Generalized Write Target implementation

capabilities/writetarget/report/platform/processor/processor.go

@@ -12,7 +12,7 @@ import (
 )
 
 // Product-agnostic processors to be injected into WriteTarget Monitor
-func NewPlatformProcessors(emitter beholder.ProtoEmitter) ([]beholder.ProtoProcessor, error) {
+func NewPlatformProcessors(emitter beholder.ProtoEmitter) (map[string]beholder.ProtoProcessor, error) {
 	forwarderMetrics, err := forwarder.NewMetrics()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create new forwarder metrics: %w", err)
@@ -22,17 +22,21 @@ func NewPlatformProcessors(emitter beholder.ProtoEmitter) ([]beholder.ProtoProce
 	if err != nil {
 		return nil, fmt.Errorf("failed to create new write target metrics: %w", err)
 	}
-	return []beholder.ProtoProcessor{
-		&keystoneProcessor{
+	return map[string]beholder.ProtoProcessor{
+		"keystone": &keystoneProcessor{
 			emitter: emitter,
 			metrics: forwarderMetrics,
 		},
-		&wtProcessor{
+		"writetarget": &wtProcessor{
 			metrics: wtMetrics,
 		},
 	}, nil
 }
 
+func GetDefaultPlatformProcessors() []string {
+	return []string{"writetarget", "keystone"}
+}
+
 // Write-Target specific processor decodes write messages to derive metrics
 type wtProcessor struct {
 	metrics *wt.Metrics

capabilities/writetarget/write_target_monitor.go

@@ -26,10 +26,10 @@ func NewMonitorEmitter(lggr logger.Logger) beholder.ProtoEmitter {
 }
 
 type MonitorOpts struct {
-	Lggr                      logger.Logger
-	ProductAgnosticProcessors []beholder.ProtoProcessor
-	ProductSpecificProcessors map[string]beholder.ProtoProcessor
-	Emitter                   beholder.ProtoEmitter
+	Lggr              logger.Logger
+	Processors        map[string]beholder.ProtoProcessor
+	EnabledProcessors []string
+	Emitter           beholder.ProtoEmitter
 }
 
 // NewMonitor initializes a Beholder client for the Write Target
@@ -44,20 +44,20 @@ func NewMonitor(opts MonitorOpts) (*beholder.BeholderClient, error) {
 
 	// Proxy ProtoEmitter with additional processing
 	protoEmitterProxy := protoEmitter{
-		lggr:                      opts.Lggr,
-		emitter:                   opts.Emitter,
-		processors:                opts.ProductAgnosticProcessors,
-		productSpecificProcessors: opts.ProductSpecificProcessors,
+		lggr:              opts.Lggr,
+		emitter:           opts.Emitter,
+		processors:        opts.Processors,
+		enabledProcessors: opts.EnabledProcessors,
 	}
 	return &beholder.BeholderClient{Client: &client, ProtoEmitter: &protoEmitterProxy}, nil
 }
 
 // ProtoEmitter proxy specific to the WT
 type protoEmitter struct {
-	lggr                      logger.Logger
-	emitter                   beholder.ProtoEmitter
-	processors                []beholder.ProtoProcessor
-	productSpecificProcessors map[string]beholder.ProtoProcessor
+	lggr              logger.Logger
+	emitter           beholder.ProtoEmitter
+	processors        map[string]beholder.ProtoProcessor
+	enabledProcessors []string
 }
 
 // Emit emits a proto.Message and runs additional processing
@@ -85,7 +85,13 @@ func (e *protoEmitter) EmitWithLog(ctx context.Context, m proto.Message, attrKVs
 // Process aggregates further processing for emitted messages
 func (e *protoEmitter) Process(ctx context.Context, m proto.Message, attrKVs ...any) error {
 	// Further processing for emitted messages
-	for _, p := range e.processors {
+	for _, processorName := range e.enabledProcessors {
+		p, ok := e.processors[processorName]
+		if !ok {
+			// no processor matching configured one, log error but continue
+			e.lggr.Errorf("no required processor with name %s", processorName)
+			continue
+		}
 		err := p.Process(ctx, m, attrKVs...)
 		if err != nil {
 			// Notice: we swallow and log processing errors
@@ -102,7 +108,7 @@ func (e *protoEmitter) Process(ctx context.Context, m proto.Message, attrKVs ...
 			return nil
 		}
 
-		if p, ok := e.productSpecificProcessors[msg.MetaCapabilityProcessor]; ok {
+		if p, ok := e.processors[msg.MetaCapabilityProcessor]; ok {
 			if err := p.Process(ctx, msg, attrKVs...); err != nil {
 				e.lggr.Errorw("failed to process emitted message", "err", err)
 			}

capabilities/writetarget/write_target_monitor_test.go

@@ -22,8 +22,8 @@
 
 	m, err := writetarget.NewMonitor(writetarget.MonitorOpts{
 		lggr,
-		[]beholder.ProtoProcessor{},
 		map[string]beholder.ProtoProcessor{"test": processor},
+		[]string{},
 		writetarget.NewMonitorEmitter(lggr),
 	})
 	require.NoError(t, err)
@@ -43,7 +43,7 @@
 	})
 
 	t.Run("Logs when config name is not found", func(t *testing.T) {
-		m, err = writetarget.NewMonitor(writetarget.MonitorOpts{lggr, []beholder.ProtoProcessor{}, map[string]beholder.ProtoProcessor{"other": processor}, writetarget.NewMonitorEmitter(lggr)})
+		m, err = writetarget.NewMonitor(writetarget.MonitorOpts{lggr, map[string]beholder.ProtoProcessor{"other": processor}, []string{}, writetarget.NewMonitorEmitter(lggr)})
 
 		err = m.ProtoEmitter.EmitWithLog(t.Context(), msg)
 		require.NoError(t, err)
@@ -57,9 +57,18 @@
 		msg.MetaCapabilityProcessor = ""
 		processor.AssertNotCalled(t, "Process", mock.Anything, mock.Anything, mock.Anything)
 
 		err := m.ProtoEmitter.EmitWithLog(t.Context(), msg)
 		require.NoError(t, err)
 
 		tests.RequireLogMessage(t, observed, "No product specific processor specified; skipping.")
 	})
+
+	t.Run("Logs when required processor is not found", func(t *testing.T) {
+		m, err = writetarget.NewMonitor(writetarget.MonitorOpts{lggr, map[string]beholder.ProtoProcessor{}, []string{"other"}, writetarget.NewMonitorEmitter(lggr)})
+
+		err = m.ProtoEmitter.EmitWithLog(t.Context(), msg)
+		require.NoError(t, err)
+
+		tests.RequireLogMessage(t, observed, "no required processor with name other")
+	})
 }

capabilities/writetarget/write_target_test.go

@@ -40,11 +40,10 @@ func setupWriteTarget(
 	platformProcessors, err := processor.NewPlatformProcessors(emitter)
 	require.NoError(t, err)
 
-	var psp map[string]beholder.ProtoProcessor
 	if productSpecificProcessor {
-		psp = map[string]beholder.ProtoProcessor{"test": newMockProductSpecificProcessor(t)}
+		platformProcessors["test"] = newMockProductSpecificProcessor(t)
 	}
-	monClient, err := writetarget.NewMonitor(writetarget.MonitorOpts{lggr, platformProcessors, psp, emitter})
+	monClient, err := writetarget.NewMonitor(writetarget.MonitorOpts{lggr, platformProcessors, processor.GetDefaultPlatformProcessors(), emitter})
 	require.NoError(t, err)
 
 	pollPeriod := 100 * time.Millisecond