feat(grpc): split into Controller + Satellite services; satellite hosts gRPC server

The original single Satellite service had Hello (satellite→controller)
mixed with ApplyResources/etc (controller→satellite), which forced
both ends to be both client AND server through a single proto. Split
into:
- service Controller — Hello + ReportObserved (satellite-streaming
  observed events). Satellite is the client.
- service Satellite — ApplyResources, ApplyStoragePools, snapshot
  RPCs, ShipSnapshot. Controller is the client.

Agent.Run now optionally hosts the Satellite service on
cfg.ListenAddr; cmd/satellite gains --listen, --lvm-pool-name,
--lvm-vg, --lvm-thinpool flags so the agent can register an LVM-thin
provider at startup. The Reconciler is reused — same code path that
unit tests already cover.

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

Author: kvaps

.golangci.yml

@@ -214,7 +214,7 @@ linters:
           - cyclop
           - gocyclo
           - maintidx
-        path: cmd/main.go
+        path: cmd/(main|satellite/main)\.go
       # api/v1alpha1 and internal/controller are kubebuilder scaffold; the
       # boilerplate (dot-imports for ginkgo/gomega, generated DeepCopy globals,
       # ginkgo BeforeSuite TODOs, wrapcheck on framework returns) is not

cmd/satellite/main.go

@@ -34,6 +34,8 @@ import (
 	"github.com/cockroachdb/errors"
 
 	"github.com/cozystack/blockstor/pkg/satellite"
+	"github.com/cozystack/blockstor/pkg/storage"
+	"github.com/cozystack/blockstor/pkg/storage/lvm"
 )
 
 func main() {
@@ -48,6 +50,10 @@ func run() int {
 		controllerAddr string
 		nodeName       string
 		stateDir       string
+		listenAddr     string
+		lvmPoolName    string
+		lvmVG          string
+		lvmThinPool    string
 	)
 
 	flag.StringVar(&controllerAddr, "controller", "blockstor-controller:7000",
@@ -56,6 +62,14 @@ func run() int {
 		"name this satellite registers under (defaults to NODE_NAME env)")
 	flag.StringVar(&stateDir, "state-dir", "/var/lib/blockstor-satellite",
 		"directory the satellite uses to persist DRBD .res files and per-resource state")
+	flag.StringVar(&listenAddr, "listen", ":7000",
+		"bind address for the satellite-side gRPC server (controller dials this for ApplyResources)")
+	flag.StringVar(&lvmPoolName, "lvm-pool-name", "",
+		"register an LVM-thin pool under this LINSTOR pool name (empty disables LVM)")
+	flag.StringVar(&lvmVG, "lvm-vg", "",
+		"LVM volume group backing the lvm-pool-name pool")
+	flag.StringVar(&lvmThinPool, "lvm-thinpool", "",
+		"LVM thinpool LV backing the lvm-pool-name pool")
 	flag.Parse()
 
 	logger := slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
@@ -70,18 +84,41 @@ func run() int {
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer cancel()
 
+	providers := map[string]storage.Provider{}
+
+	if lvmPoolName != "" {
+		if lvmVG == "" || lvmThinPool == "" {
+			logger.Error("lvm-pool-name set but lvm-vg / lvm-thinpool missing")
+
+			return 1
+		}
+
+		providers[lvmPoolName] = lvm.NewThin(
+			lvm.ThinConfig{VolumeGroup: lvmVG, ThinPool: lvmThinPool},
+			storage.RealExec{})
+	}
+
 	agent := satellite.NewAgent(satellite.Config{
 		NodeName:       nodeName,
 		ControllerAddr: controllerAddr,
+		ListenAddr:     listenAddr,
 		StateDir:       stateDir,
+		Providers:      providers,
 		DialTimeout:    10 * time.Second,
 		Logger:         logger,
 	})
 
+	providerNames := make([]string, 0, len(providers))
+	for name := range providers {
+		providerNames = append(providerNames, name)
+	}
+
 	logger.Info("blockstor-satellite starting",
 		"node_name", nodeName,
 		"controller", controllerAddr,
-		"state_dir", stateDir)
+		"state_dir", stateDir,
+		"listen", listenAddr,
+		"providers", providerNames)
 
 	err := agent.Run(ctx)
 	if err != nil && !errors.Is(err, context.Canceled) {

pkg/satellite/agent.go

@@ -26,13 +26,16 @@ package satellite
 import (
 	"context"
 	"log/slog"
+	"net"
 	"time"
 
 	"github.com/cockroachdb/errors"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
+	"github.com/cozystack/blockstor/pkg/drbd"
 	satellitepb "github.com/cozystack/blockstor/pkg/satellite/proto"
+	"github.com/cozystack/blockstor/pkg/storage"
 	"github.com/cozystack/blockstor/pkg/version"
 )
 
@@ -45,10 +48,21 @@ type Config struct {
 	// ControllerAddr is the gRPC dial address of the blockstor-controller.
 	ControllerAddr string
 
+	// ListenAddr is the bind address for the satellite's own gRPC server
+	// (the side that the controller dials for ApplyResources, snapshot
+	// RPCs, ship). Empty disables the server (useful for unit tests).
+	ListenAddr string
+
 	// StateDir is the on-disk directory the satellite uses for DRBD .res
 	// files and per-resource state. Required.
 	StateDir string
 
+	// Providers maps storage-pool name → provider implementation. The
+	// satellite reconciler uses this to resolve which backend a
+	// DesiredVolume's StoragePool refers to. Seeded at startup from CLI
+	// flags / env (LVM-thin, ZFS, file).
+	Providers map[string]storage.Provider
+
 	// DialTimeout caps gRPC connection establishment per attempt.
 	DialTimeout time.Duration
 
@@ -65,6 +79,8 @@ type Agent struct {
 }
 
 // NewAgent constructs an Agent without yet dialling the controller.
+//
+//nolint:gocritic // value receiver is the ergonomic public API; Config is the binary's flag bundle.
 func NewAgent(cfg Config) *Agent {
 	logger := cfg.Logger
 	if logger == nil {
@@ -75,11 +91,9 @@ func NewAgent(cfg Config) *Agent {
 }
 
 // Run is the agent's main loop. It dials the controller, performs the
-// hello handshake to register the node, then keeps long-running apply
-// and observe loops supervised. When ctx is cancelled the loops drain
-// and Run returns ctx.Err().
-//
-// Phase 3.1: hello is the only real RPC; reconcile loops still no-op.
+// hello handshake to register the node, starts the satellite-side
+// gRPC server (so the controller can push desired state), then waits
+// for ctx to cancel.
 func (a *Agent) Run(ctx context.Context) error {
 	if a.cfg.NodeName == "" {
 		return errors.New("NodeName is required")
@@ -88,28 +102,84 @@ func (a *Agent) Run(ctx context.Context) error {
 	a.logger.Info("agent starting",
 		"node", a.cfg.NodeName,
 		"blockstor_version", version.Version,
-		"controller", a.cfg.ControllerAddr)
+		"controller", a.cfg.ControllerAddr,
+		"listen", a.cfg.ListenAddr)
 
 	conn, err := a.dial(ctx)
 	if err != nil {
 		return errors.Wrap(err, "dial controller")
 	}
 	defer func() { _ = conn.Close() }()
 
-	client := satellitepb.NewSatelliteClient(conn)
+	client := satellitepb.NewControllerClient(conn)
 
 	err = a.hello(ctx, client)
 	if err != nil {
 		return errors.Wrap(err, "hello")
 	}
 
+	// Bring up the satellite-side gRPC server so the controller can push
+	// ApplyResources / snapshot RPCs at us. The Reconciler is wired with
+	// the configured providers + drbdadm wrapper + state dir.
+	srv, stop, err := a.startGRPCServer(ctx)
+	if err != nil {
+		return errors.Wrap(err, "start gRPC server")
+	}
+	defer stop()
+
+	a.logger.Info("satellite gRPC ready", "addr", srv)
+
 	<-ctx.Done()
 
 	a.logger.Info("agent stopping", "node", a.cfg.NodeName)
 
 	return ctx.Err() //nolint:wrapcheck // bubbling ctx.Err() unwrapped is the convention
 }
 
+// startGRPCServer binds the satellite's `service Satellite` listener.
+// Empty cfg.ListenAddr disables the server (returns a no-op stop) so
+// unit tests that only exercise Hello don't need a free port.
+func (a *Agent) startGRPCServer(ctx context.Context) (string, func(), error) {
+	if a.cfg.ListenAddr == "" {
+		return "<disabled>", func() {}, nil
+	}
+
+	rec := NewReconciler(ReconcilerConfig{
+		Providers: a.cfg.Providers,
+		Adm:       drbd.NewAdm(storage.RealExec{}),
+		StateDir:  a.cfg.StateDir,
+		NodeName:  a.cfg.NodeName,
+	})
+
+	listenCfg := &net.ListenConfig{}
+
+	listener, err := listenCfg.Listen(ctx, "tcp", a.cfg.ListenAddr)
+	if err != nil {
+		return "", nil, errors.Wrapf(err, "listen %s", a.cfg.ListenAddr)
+	}
+
+	gs := grpc.NewServer()
+	satellitepb.RegisterSatelliteServer(gs, NewGRPCServer(rec))
+
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		err := gs.Serve(listener)
+		if err != nil {
+			a.logger.Error("gRPC Serve returned", "err", err)
+		}
+	}()
+
+	stop := func() {
+		gs.GracefulStop()
+		<-done
+	}
+
+	return listener.Addr().String(), stop, nil
+}
+
 // dial opens an insecure gRPC connection to the controller. TLS comes in
 // Phase 6 alongside the rest of the encryption work; cluster traffic is
 // expected to ride a private k8s network until then.
@@ -134,7 +204,7 @@ func (a *Agent) dial(ctx context.Context) (*grpc.ClientConn, error) {
 // hello is the registration handshake. The satellite tells the controller
 // who it is and what layers / providers it can drive; the controller
 // upserts the corresponding Node CRD and replies with the cluster id.
-func (a *Agent) hello(ctx context.Context, client satellitepb.SatelliteClient) error {
+func (a *Agent) hello(ctx context.Context, client satellitepb.ControllerClient) error {
 	rpcCtx, cancel := context.WithTimeout(ctx, a.cfg.DialTimeout)
 	defer cancel()
 

pkg/satellite/agent_test.go

@@ -119,7 +119,7 @@ func startServer(t *testing.T, st store.Store, clusterID string) (string, func()
 
 	srv := satellitecontroller.New(st, satellitecontroller.Config{ClusterID: clusterID})
 	gs := grpc.NewServer()
-	satellitepb.RegisterSatelliteServer(gs, srv)
+	satellitepb.RegisterControllerServer(gs, srv)
 
 	errCh := make(chan error, 1)
 	go func() { errCh <- gs.Serve(ln) }()

pkg/satellite/grpc_server.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2026 Cozystack contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package satellite
+
+import (
+	"context"
+
+	satellitepb "github.com/cozystack/blockstor/pkg/satellite/proto"
+)
+
+// GRPCServer is the satellite-side implementation of `service Satellite`.
+// It glues controller→satellite RPCs (ApplyResources, ApplyStoragePools,
+// CreateSnapshot, DeleteSnapshot, ShipSnapshot) onto an in-process
+// Reconciler. Agent.Run wires this into a gRPC listener so the
+// controller can push desired state.
+type GRPCServer struct {
+	satellitepb.UnimplementedSatelliteServer
+
+	rec *Reconciler
+}
+
+// NewGRPCServer constructs a server backed by rec.
+func NewGRPCServer(rec *Reconciler) *GRPCServer {
+	return &GRPCServer{rec: rec}
+}
+
+// ApplyResources delegates to Reconciler.Apply. Per-resource failures
+// are surfaced via ResourceApplyResult.Ok=false rather than as gRPC
+// errors so a single bad replica doesn't sink a batched apply.
+func (g *GRPCServer) ApplyResources(ctx context.Context, req *satellitepb.ApplyResourcesRequest) (*satellitepb.ApplyResourcesResponse, error) {
+	results, err := g.rec.Apply(ctx, req.GetResources())
+	if err != nil {
+		return nil, err
+	}
+
+	return &satellitepb.ApplyResourcesResponse{Results: results}, nil
+}
+
+// ApplyStoragePools is a placeholder that just OKs every requested pool
+// for now. The satellite uses a Provider registry seeded at startup
+// (CLI flags / env), so the controller's pool spec is informational
+// today; we'll wire dynamic pool wiring once the per-pool runtime
+// state lands.
+func (g *GRPCServer) ApplyStoragePools(_ context.Context, req *satellitepb.ApplyStoragePoolsRequest) (*satellitepb.ApplyStoragePoolsResponse, error) {
+	pools := req.GetPools()
+	results := make([]*satellitepb.StoragePoolApplyResult, 0, len(pools))
+
+	for _, pool := range pools {
+		results = append(results, &satellitepb.StoragePoolApplyResult{
+			Name: pool.GetName(),
+			Ok:   true,
+		})
+	}
+
+	return &satellitepb.ApplyStoragePoolsResponse{Results: results}, nil
+}
+
+// CreateSnapshot routes through the Reconciler's existing snapshot
+// path. Per-snapshot errors land in the response body (Ok=false), the
+// gRPC error path is reserved for context cancellation / transport
+// faults.
+func (g *GRPCServer) CreateSnapshot(ctx context.Context, req *satellitepb.CreateSnapshotRequest) (*satellitepb.CreateSnapshotResponse, error) {
+	return g.rec.CreateSnapshot(ctx, req)
+}
+
+// DeleteSnapshot mirrors CreateSnapshot.
+func (g *GRPCServer) DeleteSnapshot(ctx context.Context, req *satellitepb.DeleteSnapshotRequest) (*satellitepb.DeleteSnapshotResponse, error) {
+	return g.rec.DeleteSnapshot(ctx, req)
+}
+
+// ShipSnapshot routes through the Reconciler's ship dispatch.
+func (g *GRPCServer) ShipSnapshot(ctx context.Context, req *satellitepb.ShipSnapshotRequest) (*satellitepb.ShipSnapshotResponse, error) {
+	return g.rec.ShipSnapshot(ctx, req)
+}