[m3msg] Set key on value watches (#2144)

Allows for more verbose errors.
m3db · Feb 12, 2020 · 4072128 · 4072128
1 parent cca763b
commit 4072128
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 3 deletions.
diff --git a/src/aggregator/aggregator/handler/trafficcontrol/controller.go b/src/aggregator/aggregator/handler/trafficcontrol/controller.go
@@ -81,7 +81,8 @@ func NewTrafficEnabler(opts Options) Controller {
 		SetInstrumentOptions(iOpts).
 		SetNewUpdatableFn(newUpdatableFn).
 		SetGetUpdateFn(getFn).
-		SetProcessFn(processFn)
+		SetProcessFn(processFn).
+		SetKey(opts.RuntimeKey())
 	return &trafficEnabler{
 		enabled: enabled,
 		value:   watch.NewValue(vOptions),

diff --git a/src/msg/producer/writer/consumer_service_writer.go b/src/msg/producer/writer/consumer_service_writer.go
@@ -215,7 +215,8 @@ func (w *consumerServiceWriterImpl) Init(t initType) error {
 		SetInstrumentOptions(w.opts.InstrumentOptions()).
 		SetNewUpdatableFn(updatableFn).
 		SetGetUpdateFn(getFn).
-		SetProcessFn(w.processFn)
+		SetProcessFn(w.processFn).
+		SetKey(w.opts.TopicName())
 	w.value = watch.NewValue(vOptions)
 	err := w.value.Watch()
 	if err == nil {

diff --git a/src/msg/producer/writer/writer.go b/src/msg/producer/writer/writer.go
@@ -137,7 +137,8 @@ func (w *writer) Init() error {
 		SetInstrumentOptions(w.opts.InstrumentOptions()).
 		SetNewUpdatableFn(newUpdatableFn).
 		SetGetUpdateFn(getUpdateFn).
-		SetProcessFn(w.processFn)
+		SetProcessFn(w.processFn).
+		SetKey(w.opts.TopicName())
 	w.value = watch.NewValue(vOptions)
 	if err := w.value.Watch(); err != nil {
 		return fmt.Errorf("writer init error: %v", err)