feat(dispatcher): RD VolumeDefinitions land as DesiredVolumes on satellite

Apply now also fetches the parent ResourceDefinition. Its
VolumeDefinitions become DesiredVolumes for the target Resource —
DISKLESS replicas still get an empty list (they don't allocate local
storage). The StoragePool is sourced from the Resource's
StorPoolName prop with the RD's prop as fallback.

Lays the path for the volume-side smoke: once a satellite registers
a Provider matching that pool name, applyStorage allocates the LV
and drbdadm picks it up.

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Andrei Kvapil <kvapss@gmail.com>

Author: kvaps

internal/controller/resource_controller.go

@@ -93,7 +93,12 @@ func (r *ResourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, err
 	}
 
-	result, err := r.Dispatcher.Apply(ctx, &target, peers, nodeList.Items)
+	rdPtr, err := r.lookupRD(ctx, target.Spec.ResourceDefinitionName)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	result, err := r.Dispatcher.Apply(ctx, &target, peers, nodeList.Items, rdPtr)
 	if err != nil {
 		log.Error(err, "Apply RPC failed", "resource", target.Name, "node", target.Spec.NodeName)
 
@@ -111,6 +116,24 @@ func (r *ResourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	return ctrl.Result{}, nil
 }
 
+// lookupRD fetches the parent ResourceDefinition. A NotFound is
+// converted to (nil, nil) so the dispatcher can still push the
+// .res for connection setup; any other error bubbles.
+func (r *ResourceReconciler) lookupRD(ctx context.Context, name string) (*blockstoriov1alpha1.ResourceDefinition, error) {
+	var rd blockstoriov1alpha1.ResourceDefinition
+
+	err := r.Get(ctx, client.ObjectKey{Name: name}, &rd)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			return nil, nil //nolint:nilnil // intentional: caller treats nil RD as no-volume push
+		}
+
+		return nil, err
+	}
+
+	return &rd, nil
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *ResourceReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).

pkg/dispatcher/dispatcher.go

@@ -24,6 +24,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/binary"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -82,18 +83,21 @@ func New(dialer Dialer) *Dispatcher {
 }
 
 // Apply builds the DesiredResource for this Resource (looking up its
-// peers from the full RD-wide list) and sends it to the target
-// satellite. Returns the per-resource result the satellite reported.
+// peers from the full RD-wide list and its volumes from the parent
+// RD) and sends it to the target satellite. Returns the per-resource
+// result the satellite reported.
 //
 // nodes is the full Node CRD list — Apply uses it to resolve each
-// peer's SatelliteEndpoint property.
-func (d *Dispatcher) Apply(ctx context.Context, target *blockstoriov1alpha1.Resource, peers []blockstoriov1alpha1.Resource, nodes []blockstoriov1alpha1.Node) (*satellitepb.ResourceApplyResult, error) {
+// peer's SatelliteEndpoint property. rd may be nil; when present the
+// RD's VolumeDefinitions become DesiredVolumes for non-DISKLESS
+// replicas (DISKLESS replicas ignore them).
+func (d *Dispatcher) Apply(ctx context.Context, target *blockstoriov1alpha1.Resource, peers []blockstoriov1alpha1.Resource, nodes []blockstoriov1alpha1.Node, rd *blockstoriov1alpha1.ResourceDefinition) (*satellitepb.ResourceApplyResult, error) {
 	endpoint := lookupEndpoint(target.Spec.NodeName, nodes)
 	if endpoint == "" {
 		return nil, errors.Errorf("no SatelliteEndpoint for node %q", target.Spec.NodeName)
 	}
 
-	desired := buildDesired(target, peers, nodes)
+	desired := buildDesired(target, peers, nodes, rd)
 
 	client, closer, err := d.dialer.Dial(ctx, endpoint)
 	if err != nil {
@@ -135,7 +139,7 @@ func lookupEndpoint(nodeName string, nodes []blockstoriov1alpha1.Node) string {
 // nodes is consulted for each peer's `SatelliteEndpoint` prop so
 // `peer.<name>.address` carries a real (pod) IP rather than a 0.0.0.0
 // placeholder; drbd-9 won't replicate to 0.0.0.0.
-func buildDesired(target *blockstoriov1alpha1.Resource, peers []blockstoriov1alpha1.Resource, nodes []blockstoriov1alpha1.Node) *satellitepb.DesiredResource {
+func buildDesired(target *blockstoriov1alpha1.Resource, peers []blockstoriov1alpha1.Resource, nodes []blockstoriov1alpha1.Node, rd *blockstoriov1alpha1.ResourceDefinition) *satellitepb.DesiredResource {
 	rdName := target.Spec.ResourceDefinitionName
 	port := derivePort(rdName)
 	minor := deriveMinor(rdName)
@@ -189,10 +193,43 @@ func buildDesired(target *blockstoriov1alpha1.Resource, peers []blockstoriov1alp
 		Flags:       target.Spec.Flags,
 		Props:       target.Spec.Props,
 		Peers:       dropped,
+		Volumes:     buildVolumes(rd, target),
 		DrbdOptions: drbdOpts,
 	}
 }
 
+// buildVolumes turns the parent ResourceDefinition's VolumeDefinitions
+// into DesiredVolumes for the satellite. DISKLESS replicas get an
+// empty list — they don't allocate local storage. The StoragePool name
+// comes from the Resource's `StorPoolName` prop (LINSTOR-compatible
+// key) with the RD-level fallback.
+func buildVolumes(rd *blockstoriov1alpha1.ResourceDefinition, target *blockstoriov1alpha1.Resource) []*satellitepb.DesiredVolume {
+	if rd == nil {
+		return nil
+	}
+
+	if slices.Contains(target.Spec.Flags, "DISKLESS") {
+		return nil
+	}
+
+	pool := target.Spec.Props["StorPoolName"]
+	if pool == "" {
+		pool = rd.Spec.Props["StorPoolName"]
+	}
+
+	out := make([]*satellitepb.DesiredVolume, 0, len(rd.Spec.VolumeDefinitions))
+
+	for _, vd := range rd.Spec.VolumeDefinitions {
+		out = append(out, &satellitepb.DesiredVolume{
+			VolumeNumber: vd.VolumeNumber,
+			SizeKib:      vd.SizeKib,
+			StoragePool:  pool,
+		})
+	}
+
+	return out
+}
+
 // peerAddress looks up `nodeName`'s SatelliteEndpoint and returns
 // just the host part (no port). Falls back to the 0.0.0.0 placeholder
 // when the node is unknown or hasn't registered yet — the satellite

pkg/dispatcher/dispatcher_test.go

@@ -81,7 +81,7 @@ func TestApplyDialsTargetSatellite(t *testing.T) {
 		},
 	}}
 
-	result, err := d.Apply(t.Context(), target, nil, nodes)
+	result, err := d.Apply(t.Context(), target, nil, nodes, nil)
 	if err != nil {
 		t.Fatalf("Apply: %v", err)
 	}
@@ -108,7 +108,7 @@ func TestApplyMissingEndpoint(t *testing.T) {
 		Spec:       blockstoriov1alpha1.NodeSpec{Type: "SATELLITE"},
 	}}
 
-	_, err := d.Apply(t.Context(), target, nil, nodes)
+	_, err := d.Apply(t.Context(), target, nil, nodes, nil)
 	if err == nil {
 		t.Fatalf("expected error when endpoint missing")
 	}
@@ -141,7 +141,7 @@ func TestApplyBuildsPeers(t *testing.T) {
 		},
 	}}
 
-	_, err := d.Apply(t.Context(), target, peers, nodes)
+	_, err := d.Apply(t.Context(), target, peers, nodes, nil)
 	if err != nil {
 		t.Fatalf("Apply: %v", err)
 	}
@@ -175,7 +175,7 @@ func TestApplyDialError(t *testing.T) {
 		},
 	}}
 
-	_, err := d.Apply(t.Context(), target, nil, nodes)
+	_, err := d.Apply(t.Context(), target, nil, nodes, nil)
 	if err == nil {
 		t.Fatalf("expected dial error")
 	}