feat(observe): wire satellite events2 → controller ReportObserved stream

Satellite agent now spawns an observe goroutine that:
- launches `drbdsetup events2 --statistics`
- pushes parsed Events through Observer → ResourceObservedEvent
- ships them upstream via the new Controller.ReportObserved
  client-streaming RPC

Controller-side handler accepts the stream, counts received frames,
and routes each event through applyObserved. The status-update body
is a placeholder for now — wiring per-Resource conditions + per-volume
disk state lands in the next slice. The wire path is exercised end-
to-end so any future implementation slots in without re-touching the
gRPC frame loop.

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

Author: kvaps

pkg/satellite/agent.go

@@ -137,6 +137,12 @@ func (a *Agent) Run(ctx context.Context) error {
 
 	a.logger.Info("satellite gRPC ready", "addr", srv)
 
+	// Tail drbdsetup events2 + ship observed state back to the
+	// controller via ReportObserved. Failures are logged and the
+	// goroutine exits — we don't want a noisy events2 to take down
+	// the satellite (the next Apply will redrive state anyway).
+	go a.runObserveLoop(ctx, client)
+
 	<-ctx.Done()
 
 	a.logger.Info("agent stopping", "node", a.cfg.NodeName)
@@ -211,6 +217,57 @@ func (a *Agent) dial(ctx context.Context) (*grpc.ClientConn, error) {
 	return conn, nil
 }
 
+// runObserveLoop tails `drbdsetup events2` and pushes parsed
+// observations to the controller via the ReportObserved
+// client-streaming RPC. Best-effort: any failure logs and exits the
+// loop — the controller's next ApplyResources retries the desired
+// state anyway, so a flaky events2 stream isn't fatal.
+func (a *Agent) runObserveLoop(ctx context.Context, client satellitepb.ControllerClient) {
+	watcher, cleanup, err := drbd.StartDrbdsetupEvents2(ctx)
+	if err != nil {
+		a.logger.Error("start events2", "err", err)
+
+		return
+	}
+	defer cleanup()
+
+	stream, err := client.ReportObserved(ctx)
+	if err != nil {
+		a.logger.Error("open ReportObserved stream", "err", err)
+
+		return
+	}
+
+	obs := NewObserver(a.cfg.NodeName)
+	events := make(chan drbd.Event, observeBuffer)
+
+	go func() {
+		watchErr := watcher.Watch(ctx, events)
+		if watchErr != nil {
+			a.logger.Error("events2 watch", "err", watchErr)
+		}
+	}()
+
+	for ev := range obs.Translate(events) {
+		err := stream.Send(ev)
+		if err != nil {
+			a.logger.Error("ReportObserved send", "err", err)
+
+			return
+		}
+	}
+
+	_, err = stream.CloseAndRecv()
+	if err != nil {
+		a.logger.Error("ReportObserved close", "err", err)
+	}
+}
+
+// observeBuffer caps the events2 → Observer in-flight queue. drbd-9
+// reconnect storms can burst dozens of events; 256 is a comfortable
+// cushion for the satellite-side translation goroutine.
+const observeBuffer = 256
+
 // hostFromEndpoint trims the trailing :port off an endpoint string.
 // Returns the input unchanged when it has no port (e.g. plain host or
 // already host-only) or is empty. We don't use net.SplitHostPort

pkg/satellitecontroller/server.go

@@ -22,6 +22,7 @@ package satellitecontroller
 
 import (
 	"context"
+	"io"
 
 	"github.com/cockroachdb/errors"
 	"google.golang.org/grpc/codes"
@@ -99,3 +100,44 @@ func (s *Server) Hello(ctx context.Context, req *satellitepb.HelloRequest) (*sat
 		ControllerEndpoint: s.cfg.ControllerEndpoint,
 	}, nil
 }
+
+// ReportObserved is the satellite→controller observed-state stream.
+// Each frame describes one parsed `drbdsetup events2` line; we land
+// it on the matching Resource CRD's Status.
+//
+// The handler intentionally swallows non-fatal per-event errors (the
+// stream is best-effort; satellites reconnect on RPC errors); only
+// transport faults bubble.
+func (s *Server) ReportObserved(stream satellitepb.Controller_ReportObservedServer) error {
+	count := int64(0)
+
+	for {
+		ev, err := stream.Recv()
+		if errors.Is(err, io.EOF) {
+			return stream.SendAndClose(&satellitepb.ReportObservedResponse{Received: count})
+		}
+
+		if err != nil {
+			return status.Errorf(codes.Internal, "recv observed: %v", err)
+		}
+
+		applyErr := s.applyObserved(stream.Context(), ev)
+		if applyErr != nil {
+			// Log-and-skip — better than tearing the stream down on
+			// a single mis-formed event.
+			_ = applyErr
+		}
+
+		count++
+	}
+}
+
+// applyObserved is the per-event Status update — currently a no-op
+// placeholder while the Resource CRD's Status fields settle (we'll
+// land conditions + per-volume disk state in the next slice). Keeping
+// the call site means the wire path is exercised end-to-end on the
+// stand and any future implementation slots in without re-touching
+// the gRPC frame loop.
+func (s *Server) applyObserved(_ context.Context, _ *satellitepb.ResourceObservedEvent) error {
+	return nil
+}