diff --git a/pkg/collector/managers/targets/targets_manager.go b/pkg/collector/managers/targets/targets_manager.go index a26168a4..e7f3e8a8 100644 --- a/pkg/collector/managers/targets/targets_manager.go +++ b/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 diff --git a/pkg/collector/managers/targets/targets_manager_test.go b/pkg/collector/managers/targets/targets_manager_test.go index a600c9e8..edaa0cd1 100644 --- a/pkg/collector/managers/targets/targets_manager_test.go +++ b/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) {