feat(targets): enhance event tag handling in targets manager

pkg/collector/managers/targets/targets_manager.go

@@ -47,6 +47,9 @@ type ManagedTarget struct {
 	lastError            string // last error message, protected by mu
 	outputs              map[string]struct{}
 	appliedSubscriptions []string
+	// Immutable event-tags snapshot. Replaced (never mutated) under mt.Lock
+	// when the target is created, reconnected, or tags are updated in place.
+	eventTags map[string]string
 }
 
 func (mt *ManagedTarget) setLastError(msg string) {
@@ -82,9 +85,21 @@ func newManagedTarget(name string, cfg *types.TargetConfig, tunServer *tunnel.Se
 	for _, output := range cfg.Outputs {
 		mt.outputs[output] = struct{}{}
 	}
+	mt.setEventTags(cfg.EventTags)
 	return mt
 }
 
+// setEventTags clones tags once and installs the snapshot on both the live
+// config and the reader-facing field. Callers must hold mt.Lock, except
+// newManagedTarget which is not yet published.
+func (mt *ManagedTarget) setEventTags(tags map[string]string) {
+	cloned := maps.Clone(tags)
+	if mt.T != nil && mt.T.Config != nil {
+		mt.T.Config.EventTags = cloned
+	}
+	mt.eventTags = cloned
+}
+
 // TargetsManager owns target lifecycle (connect/stop) and per-target subscriptions hookups (started by SubscriptionsManager).
 type TargetsManager struct {
 	ctx    context.Context
@@ -416,6 +431,13 @@ func (tm *TargetsManager) apply(name string, cfg *types.TargetConfig) {
 		} else {
 			tm.logger.Info("outputs unchanged", "name", name, "old", mt.T.Config.Outputs, "new", cfg.Outputs)
 		}
+		// event-tags are not part of the connection spec: replace the
+		// snapshot in place so live readers pick up the new tags without a
+		// reconnect or a per-response clone.
+		if !maps.Equal(mt.eventTags, cfg.EventTags) {
+			tm.logger.Info("event-tags changed", "name", name)
+			mt.setEventTags(cfg.EventTags)
+		}
 		return
 	}
 
@@ -427,6 +449,7 @@ func (tm *TargetsManager) apply(name string, cfg *types.TargetConfig) {
 		tm.setTargetState(name, collstore.StateFailed)
 	}
 	mt.T.Config = cfg
+	mt.setEventTags(cfg.EventTags)
 	err = tm.start(mt)
 	if err != nil {
 		tm.logger.Error("failed to start target", "name", name, "error", err)
@@ -730,6 +753,7 @@ func (tm *TargetsManager) reconcileAssignment(name string) {
 		tm.setTargetState(name, collstore.StateFailed)
 	}
 	mt.T.Config = cfg
+	mt.setEventTags(cfg.EventTags)
 	err = tm.start(mt)
 	if err != nil {
 		tm.logger.Error("failed to start target", "name", name, "error", err)
@@ -871,13 +895,13 @@ func (tm *TargetsManager) startTargetSubscription(mt *ManagedTarget, cfg *types.
 					}
 					return cp
 				}()
+				mt.RLock()
+				eventTags := mt.eventTags
+				mt.RUnlock()
 				select {
 				case tm.out <- &pipeline.Msg{
-					Msg: resp.Response,
-					Meta: outputs.Meta{
-						"source":            mt.Name,
-						"subscription-name": resp.SubscriptionName,
-					},
+					Msg:     resp.Response,
+					Meta:    pipelineMeta(mt.Name, resp.SubscriptionName, eventTags),
 					Outputs: outs,
 				}:
 				default:
@@ -905,6 +929,21 @@ func (tm *TargetsManager) startTargetSubscription(mt *ManagedTarget, cfg *types.
 	return nil
 }
 
+// pipelineMeta builds collector pipeline metadata the same way the CLI
+// subscribe path does: source and subscription-name first, then the target's
+// event-tags (which may override those keys). formatters.addMetaTags copies
+// every meta entry onto the event as a tag.
+func pipelineMeta(targetName, subscriptionName string, eventTags map[string]string) outputs.Meta {
+	meta := outputs.Meta{
+		"source":            targetName,
+		"subscription-name": subscriptionName,
+	}
+	for k, v := range eventTags {
+		meta[k] = v
+	}
+	return meta
+}
+
 func shouldReconnect(old, new *types.TargetConfig) bool {
 	if old == nil && new != nil {
 		return true

pkg/collector/managers/targets/targets_manager_test.go

@@ -5,8 +5,8 @@ import (
 	"testing"
 
 	"github.com/openconfig/gnmic/pkg/api/types"
-	"github.com/openconfig/gnmic/pkg/config"
 	collstore "github.com/openconfig/gnmic/pkg/collector/store"
+	"github.com/openconfig/gnmic/pkg/config"
 	"github.com/openconfig/gnmic/pkg/pipeline"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/zestor-dev/zestor/store"
@@ -104,6 +104,57 @@ func TestShouldReconnect(t *testing.T) {
 	if shouldReconnect(base, unchanged) {
 		t.Fatal("subscription-only change should not reconnect")
 	}
+
+	tagsOnly := base.DeepCopy()
+	tagsOnly.EventTags = map[string]string{"site": "alpha"}
+	if shouldReconnect(base, tagsOnly) {
+		t.Fatal("event-tags-only change should not reconnect")
+	}
+}
+
+func TestPipelineMetaIncludesEventTags(t *testing.T) {
+	got := pipelineMeta("dut-1", "sub1", map[string]string{"site": "alpha", "role": "edge"})
+	want := map[string]string{
+		"source":            "dut-1",
+		"subscription-name": "sub1",
+		"site":              "alpha",
+		"role":              "edge",
+	}
+	for k, v := range want {
+		if got[k] != v {
+			t.Fatalf("meta[%q] = %q, want %q (meta=%v)", k, got[k], v, got)
+		}
+	}
+	if got = pipelineMeta("dut-1", "sub1", nil); got["source"] != "dut-1" || got["subscription-name"] != "sub1" {
+		t.Fatalf("nil event-tags meta = %v", got)
+	}
+}
+
+func TestApply_eventTagsInPlace(t *testing.T) {
+	tm := newTargetsTestManager(t)
+	cfg := &types.TargetConfig{
+		Name:      "t1",
+		Address:   "10.0.0.1:57400",
+		EventTags: map[string]string{"site": "alpha"},
+	}
+	mt := newManagedTarget("t1", cfg.DeepCopy(), nil)
+	tm.mu.Lock()
+	tm.targets["t1"] = mt
+	tm.mu.Unlock()
+
+	updated := cfg.DeepCopy()
+	updated.EventTags = map[string]string{"site": "gamma"}
+	tm.apply("t1", updated)
+
+	if mt.T.Config.EventTags["site"] != "gamma" {
+		t.Fatalf("event-tags not applied in place: %#v", mt.T.Config.EventTags)
+	}
+	if mt.eventTags["site"] != "gamma" {
+		t.Fatalf("event-tags snapshot not replaced: %#v", mt.eventTags)
+	}
+	if st := tm.getTargetStateStr("t1"); st == collstore.StateFailed {
+		t.Fatal("event-tags update reconnected (start failed against a dummy address)")
+	}
 }
 
 func TestAmIAssigned_standaloneAndCluster(t *testing.T) {