feat(satellite): Observer translates drbd.Event into ResourceObservedEvent

Stateless mapping from parsed events2 lines to the proto observation
type. Lets the satellite ship runtime state back to the controller
without leaking the events2 wire format upward. RPC plumbing follows
in the next slice.

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

Author: kvaps

PLAN.md

@@ -228,7 +228,7 @@ Full scope list lives in `docs/csi-api-surface.md` (to be created in Phase 1).
 - [x] `drbdadm up/down/adjust/create-md/primary/secondary` exec wrappers behind interface (`pkg/drbd/drbdadm.go`); 7 contract tests via FakeExec
 - [x] `drbdsetup events2` listener (`pkg/drbd/events2.go`): line parser + Watcher streaming `Event{Action,Kind,Fields}` to a channel; 7 contract tests
 - [x] Resource reconciler (`pkg/satellite.Reconciler`) routes DesiredResource batches: storage provider CreateVolume per volume, ConfFileBuilder writes /etc/drbd.d/<name>.res, drbdadm create-md (first activation, non-DISKLESS) + adjust. Status writeback from events2 stream is the next slice.
-- [ ] Status writeback: events2 listener pushes ResourceObservedEvent stream back to controller
+- [x] Status writeback first half: `pkg/satellite.Observer` translates parsed drbd.Event values into ResourceObservedEvent (4 contract tests). Wire-up to gRPC stream pending the controller-side handler.
 - [ ] Resource on 2 nodes replicates and goes UpToDate (real DRBD smoke)
 
 **Exit**: smoke test with two replicas, real DRBD, PVC mounted on node A then on node B (failover).

pkg/satellite/observer.go

@@ -0,0 +1,104 @@
+/*
+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 (
+	"strconv"
+	"time"
+
+	"github.com/cozystack/blockstor/pkg/drbd"
+	satellitepb "github.com/cozystack/blockstor/pkg/satellite/proto"
+)
+
+// Observation is the satellite-side view of one DRBD state change.
+// It is a thin wrapper around the proto message so the rest of the
+// satellite code can pass observations around without depending on
+// generated gRPC types directly.
+type Observation = satellitepb.ResourceObservedEvent
+
+// Observer translates parsed `drbdsetup events2` lines into the proto
+// observations the controller consumes. It is stateless across events
+// — every output is derived from one input — which keeps the wire
+// format simple and lets the controller dedupe/coalesce on its side.
+type Observer struct {
+	nodeName string
+	now      func() time.Time
+}
+
+// NewObserver constructs an Observer that stamps `time.Now()` into
+// every emitted observation.
+func NewObserver(nodeName string) *Observer {
+	return &Observer{nodeName: nodeName, now: time.Now}
+}
+
+// Translate consumes events from in, emits observations on the
+// returned channel, and closes the output when in is closed. The
+// goroutine exits when in closes; callers wanting cancellation should
+// close in themselves.
+func (o *Observer) Translate(in <-chan drbd.Event) <-chan *Observation {
+	out := make(chan *Observation)
+
+	go func() {
+		defer close(out)
+
+		for ev := range in {
+			obs, ok := o.translate(ev)
+			if !ok {
+				continue
+			}
+
+			out <- obs
+		}
+	}()
+
+	return out
+}
+
+// translate is the core mapping. Returns ok=false for events we
+// shouldn't surface (wrong kind, missing resource name, etc.).
+func (o *Observer) translate(ev drbd.Event) (*Observation, bool) {
+	if ev.Kind != "device" {
+		return nil, false
+	}
+
+	name := ev.Fields["name"]
+	if name == "" {
+		return nil, false
+	}
+
+	disk := ev.Fields["disk"]
+
+	obs := &Observation{
+		ResourceName:  name,
+		NodeName:      o.nodeName,
+		DrbdState:     disk,
+		TimestampUnix: o.now().Unix(),
+	}
+
+	volStr, hasVol := ev.Fields["volume"]
+	if hasVol {
+		volNum, err := strconv.Atoi(volStr)
+		if err == nil {
+			obs.Volumes = []*satellitepb.VolumeObservation{{
+				VolumeNumber: int32(volNum), //nolint:gosec // volume numbers are small (<32) per drbd-9 limits
+				DiskState:    disk,
+			}}
+		}
+	}
+
+	return obs, true
+}

pkg/satellite/observer_test.go

@@ -0,0 +1,167 @@
+/*
+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_test
+
+import (
+	"testing"
+
+	"github.com/cozystack/blockstor/pkg/drbd"
+	"github.com/cozystack/blockstor/pkg/satellite"
+)
+
+// TestObserverDeviceEventEmitsState: a `change device disk:UpToDate`
+// event surfaces a ResourceObservedEvent with drbd_state populated.
+func TestObserverDeviceEventEmitsState(t *testing.T) {
+	obs := satellite.NewObserver("n1")
+
+	events := []drbd.Event{
+		{
+			Action: "change",
+			Kind:   "device",
+			Fields: map[string]string{
+				"name":   "pvc-1",
+				"volume": "0",
+				"disk":   "UpToDate",
+			},
+		},
+	}
+
+	out := collectObservations(obs, events)
+
+	if len(out) != 1 {
+		t.Fatalf("len(out): got %d, want 1", len(out))
+	}
+
+	if got := out[0].GetResourceName(); got != "pvc-1" {
+		t.Errorf("ResourceName: got %q", got)
+	}
+
+	if got := out[0].GetNodeName(); got != "n1" {
+		t.Errorf("NodeName: got %q", got)
+	}
+
+	if got := out[0].GetDrbdState(); got != "UpToDate" {
+		t.Errorf("DrbdState: got %q", got)
+	}
+}
+
+// TestObserverIgnoresUnrelatedKinds: connection / peer-device kinds
+// don't carry per-resource disk state, so they don't drive observations.
+func TestObserverIgnoresUnrelatedKinds(t *testing.T) {
+	obs := satellite.NewObserver("n1")
+
+	events := []drbd.Event{
+		{
+			Action: "change",
+			Kind:   "connection",
+			Fields: map[string]string{
+				"name":       "pvc-1",
+				"connection": "Connected",
+			},
+		},
+		{
+			Action: "exists",
+			Kind:   "-",
+			Fields: map[string]string{},
+		},
+	}
+
+	out := collectObservations(obs, events)
+
+	if len(out) != 0 {
+		t.Errorf("expected no observations; got %v", out)
+	}
+}
+
+// TestObserverDropsEventsWithoutResource: a malformed event that lacks a
+// `name` field can't be matched to a resource — silently skip.
+func TestObserverDropsEventsWithoutResource(t *testing.T) {
+	obs := satellite.NewObserver("n1")
+
+	events := []drbd.Event{
+		{
+			Action: "change",
+			Kind:   "device",
+			Fields: map[string]string{
+				"disk": "UpToDate",
+			},
+		},
+	}
+
+	out := collectObservations(obs, events)
+
+	if len(out) != 0 {
+		t.Errorf("expected no observations; got %v", out)
+	}
+}
+
+// TestObserverIncludesVolumeObservation: device events carry volume
+// number — surface it in the VolumeObservation.
+func TestObserverIncludesVolumeObservation(t *testing.T) {
+	obs := satellite.NewObserver("n1")
+
+	events := []drbd.Event{
+		{
+			Action: "change",
+			Kind:   "device",
+			Fields: map[string]string{
+				"name":   "pvc-1",
+				"volume": "0",
+				"disk":   "UpToDate",
+			},
+		},
+	}
+
+	out := collectObservations(obs, events)
+
+	if len(out) != 1 {
+		t.Fatalf("len(out): got %d", len(out))
+	}
+
+	vols := out[0].GetVolumes()
+	if len(vols) != 1 {
+		t.Fatalf("len(volumes): got %d, want 1", len(vols))
+	}
+
+	if got := vols[0].GetVolumeNumber(); got != 0 {
+		t.Errorf("VolumeNumber: got %d", got)
+	}
+
+	if got := vols[0].GetDiskState(); got != "UpToDate" {
+		t.Errorf("DiskState: got %q", got)
+	}
+}
+
+// collectObservations is a shared helper to drain Observer.Observe over
+// a fixed input slice. We push into the input channel, close it, then
+// read everything off the output channel.
+func collectObservations(obs *satellite.Observer, events []drbd.Event) []*satellite.Observation {
+	in := make(chan drbd.Event, len(events))
+	for _, e := range events {
+		in <- e
+	}
+
+	close(in)
+
+	var out []*satellite.Observation
+
+	for ev := range obs.Translate(in) {
+		out = append(out, ev)
+	}
+
+	return out
+}