smartagent: remove event clients in favor of log pipelines

Author: Ryan Fitzpatrick

internal/receiver/smartagentreceiver/README.md

@@ -19,30 +19,58 @@ should be used.
 [Collector processors](https://github.com/open-telemetry/opentelemetry-collector/blob/master/processor/README.md).
 1. Monitors with [dimension property and tag update
 functionality](https://dev.splunk.com/observability/docs/datamodel#Creating-or-updating-custom-properties-and-tags)
-require an associated `dimensionClients` field that references the name of the SignalFx exporter you are using in your
-pipeline.
+allow an associated `dimensionClients` field that references the name of the SignalFx exporter you are using in your
+pipeline.  If you do not specify any exporters via this field, the receiver will attempt to use the associated
+pipeline.  If the next element of the pipeline isn't compatible with dimension update behavior, if a lone SignalFx
+exporter was configured for your deployment, it will be selected.  If no dimension update behavior is desired,
+you can specify the empty array `[]` to disable.
 1. Monitors with [event-sending
-functionality](https://dev.splunk.com/observability/docs/datamodel/ingest#Send-custom-events)
-require an associated `eventClients` field that references the name of the SignalFx exporter you are using in your
-pipeline.
+functionality](https://dev.splunk.com/observability/docs/datamodel/ingest#Send-custom-events) should be made members of
+a `logs` pipeline that utilizes a [SignalFx
+exporter](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/exporter/signalfxexporter/README.md)
+that will make the event submission requests.  It's recommended, and in the case of the Processlist monitor required,
+to use a Resource Detection to ensure that host identity and other useful information is made available as event
+dimensions.
 
 Example:
 
 ```yaml
-exporters:
-  signalfx:
+extensions:
+  smartagent:
 
 receivers:
-  smartagent/haproxy:
-    type: haproxy
-    host: myhaproxyinstance
-    port: 8080
   smartagent/postgresql:
     type: postgresql
     host: mypostgresinstance
     port: 5432
     dimensionClients: [signalfx]
-    eventClients: [signalfx]
+  smartagent/processlist:
+    type: processlist
+  smartagent/kafka:
+    type: collectd/kafka
+    host: mykafkabroker
+    port: 7099
+    clusterName: mykafkacluster
+    intervalSeconds: 1
+
+exporters:
+  signalfx:
+
+processors:
+  resourcedetection:
+    detectors: [system]
+
+service:
+  extensions: [smartagent]
+  pipelines:
+    metrics:
+      receivers: [smartagent/postgresql, smartagent/kafka]
+      processors: [resourcedetection]
+      exporters: [signalfx]
+    logs:
+      receivers: [smartagent/processlist]
+      processors: [resourcedetection]
+      exporters: [signalfx]
 ```
 
 The full list of settings exposed for this receiver are documented for

internal/receiver/smartagentreceiver/config.go

@@ -33,7 +33,6 @@ import (
 const defaultIntervalSeconds = 10
 
 var errDimensionClientValue = fmt.Errorf("dimensionClients must be an array of compatible exporter names")
-var errEventClientValue = fmt.Errorf("eventClients must be an array of compatible exporter names")
 
 type Config struct {
 	monitorConfig                 config.MonitorCustomConfig
@@ -42,7 +41,6 @@ type Config struct {
 	// Will expand to MonitorCustomConfig Host and Port values if unset.
 	Endpoint         string   `mapstructure:"endpoint"`
 	DimensionClients []string `mapstructure:"dimensionclients"`
-	EventClients     []string `mapstructure:"eventclients"`
 }
 
 func (rCfg *Config) validate() error {
@@ -87,10 +85,6 @@ func mergeConfigs(componentViperSection *viper.Viper, intoCfg interface{}) error
 	if err != nil {
 		return err
 	}
-	receiverCfg.EventClients, err = getStringSliceFromAllSettings(allSettings, "eventclients", errEventClientValue)
-	if err != nil {
-		return err
-	}
 
 	// monitors.ConfigTemplates is a map that all monitors use to register their custom configs in the Smart Agent.
 	// The values are always pointers to an actual custom config.

internal/receiver/smartagentreceiver/config_test.go

@@ -54,14 +54,12 @@ func TestLoadConfig(t *testing.T) {
 
 	haproxyCfg := cfg.Receivers["smartagent/haproxy"].(*Config)
 	expectedDimensionClients := []string{"exampleexporter/one", "exampleexporter/two"}
-	expectedEventClients := []string{"exampleexporter/two", "exampleexporter/three"}
 	require.Equal(t, &Config{
 		ReceiverSettings: configmodels.ReceiverSettings{
 			TypeVal: typeStr,
 			NameVal: typeStr + "/haproxy",
 		},
 		DimensionClients: expectedDimensionClients,
-		EventClients:     expectedEventClients,
 		monitorConfig: &haproxy.Config{
 			MonitorConfig: config.MonitorConfig{
 				Type:                "haproxy",
@@ -84,7 +82,6 @@ func TestLoadConfig(t *testing.T) {
 			NameVal: typeStr + "/redis",
 		},
 		DimensionClients: []string{},
-		EventClients:     []string{},
 		monitorConfig: &redis.Config{
 			MonitorConfig: config.MonitorConfig{
 				Type:                "collectd/redis",
@@ -388,18 +385,18 @@ func TestLoadInvalidConfigWithUnsupportedEndpoint(t *testing.T) {
 	require.Nil(t, cfg)
 }
 
-func TestLoadInvalidConfigWithNonArrayEventClients(t *testing.T) {
+func TestLoadInvalidConfigWithNonArrayDimensionClients(t *testing.T) {
 	factories, err := componenttest.ExampleComponents()
 	assert.Nil(t, err)
 
 	factory := NewFactory()
 	factories.Receivers[configmodels.Type(typeStr)] = factory
 	cfg, err := configtest.LoadConfigFile(
-		t, path.Join(".", "testdata", "invalid_nonarray_event_clients.yaml"), factories,
+		t, path.Join(".", "testdata", "invalid_nonarray_dimension_clients.yaml"), factories,
 	)
 	require.Error(t, err)
 	require.EqualError(t, err,
-		"error reading receivers configuration for smartagent/haproxy: eventClients must be an array of compatible exporter names")
+		"error reading receivers configuration for smartagent/haproxy: dimensionClients must be an array of compatible exporter names")
 	require.Nil(t, cfg)
 }
 

internal/receiver/smartagentreceiver/output.go

@@ -16,7 +16,6 @@ package smartagentreceiver
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	metadata "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata"
@@ -36,31 +35,33 @@ import (
 const internalTransport = "internal"
 
 // Output is an implementation of a Smart Agent FilteringOutput that receives datapoints, events, and dimension updates
-// from a configured monitor.  It will forward all datapoints to the nextConsumer, all dimension updates to the
-// nextDimensionClients, and all events to the nextEventClients as determined by the associated
-// items in Config.MetadataClients.
+// from a configured monitor.  It will forward all datapoints to the nextMetricsConsumer, all dimension updates to the
+// nextDimensionClients as determined by the associated items in Config.MetadataClients, and all events to the
+// nextLogsConsumer.
 type Output struct {
-	nextConsumer         consumer.MetricsConsumer
+	nextMetricsConsumer  consumer.MetricsConsumer
+	nextLogsConsumer     consumer.LogsConsumer
 	extraDimensions      map[string]string
 	logger               *zap.Logger
 	converter            Converter
 	monitorFiltering     *monitorFiltering
 	receiverName         string
 	nextDimensionClients []*metadata.MetadataExporter
-	nextEventClients     []*consumer.LogsConsumer
 }
 
 var _ types.Output = (*Output)(nil)
 var _ types.FilteringOutput = (*Output)(nil)
 
-func NewOutput(config Config, filtering *monitorFiltering, nextConsumer consumer.MetricsConsumer, host component.Host, logger *zap.Logger) *Output {
-	metadataExporters := getMetadataExporters(config, host, &nextConsumer, logger)
-	logConsumers := getLogsConsumers(config, host, &nextConsumer, logger)
+func NewOutput(
+	config Config, filtering *monitorFiltering, nextMetricsConsumer consumer.MetricsConsumer,
+	nextLogsConsumer consumer.LogsConsumer, host component.Host, logger *zap.Logger,
+) *Output {
+	metadataExporters := getMetadataExporters(config, host, &nextMetricsConsumer, logger)
 	return &Output{
 		receiverName:         config.Name(),
-		nextConsumer:         nextConsumer,
+		nextMetricsConsumer:  nextMetricsConsumer,
+		nextLogsConsumer:     nextLogsConsumer,
 		nextDimensionClients: metadataExporters,
-		nextEventClients:     logConsumers,
 		logger:               logger,
 		converter:            Converter{logger: logger},
 		extraDimensions:      map[string]string{},
@@ -71,11 +72,11 @@ func NewOutput(config Config, filtering *monitorFiltering, nextConsumer consumer
 // getMetadataExporters walks through obtained Config.MetadataClients and returns all matching registered MetadataExporters,
 // if any.  At this time the SignalFx exporter is the only supported use case and adopter of this type.
 func getMetadataExporters(
-	config Config, host component.Host, nextConsumer *consumer.MetricsConsumer, logger *zap.Logger,
+	config Config, host component.Host, nextMetricsConsumer *consumer.MetricsConsumer, logger *zap.Logger,
 ) []*metadata.MetadataExporter {
 	var exporters []*metadata.MetadataExporter
 
-	metadataExporters := getClientsFromMetricsExporters(config.DimensionClients, host, nextConsumer, "dimensionClients", logger)
+	metadataExporters, noClientsSpecified := getDimensionClientsFromMetricsExporters(config.DimensionClients, host, nextMetricsConsumer, logger)
 	for _, client := range metadataExporters {
 		if metadataExporter, ok := (*client).(metadata.MetadataExporter); ok {
 			exporters = append(exporters, &metadataExporter)
@@ -84,68 +85,75 @@ func getMetadataExporters(
 		}
 	}
 
+	if len(exporters) == 0 && noClientsSpecified {
+		sfxExporter := getLoneSFxExporter(host, configmodels.MetricsDataType)
+		if sfxExporter != nil {
+			if sfx, ok := sfxExporter.(metadata.MetadataExporter); ok {
+				exporters = append(exporters, &sfx)
+			}
+		}
+	}
+
 	if len(exporters) == 0 {
 		logger.Debug("no dimension updates are possible as no valid dimensionClients have been provided and next pipeline component isn't a MetadataExporter")
 	}
 
 	return exporters
 }
 
-// getLogsConsumers walks through obtained Config.EventClients and returns all matching registered LogsConsumers,
-// if any.  At this time the SignalFx exporter is the only real target use case, but it's unexported and
-// as implemented all specified combination MetricsExporters and LogsConsumers will be returned.
-func getLogsConsumers(
-	config Config, host component.Host, nextConsumer *consumer.MetricsConsumer, logger *zap.Logger,
-) []*consumer.LogsConsumer {
-	var consumers []*consumer.LogsConsumer
-
-	eventClients := getClientsFromMetricsExporters(config.EventClients, host, nextConsumer, "eventClients", logger)
-	for _, client := range eventClients {
-		if logsExporter, ok := (*client).(consumer.LogsConsumer); ok {
-			consumers = append(consumers, &logsExporter)
-		} else {
-			logger.Info("cannot send events to event client", zap.Any("client", *client))
+// getDimensionClientsFromMetricsExporters will walk through all provided config.DimensionClients and retrieve matching registered
+// MetricsExporters, the only truly supported component type.
+// If config.MetadataClients is nil, it will return a slice with nextMetricsConsumer if it's a MetricsExporter.
+func getDimensionClientsFromMetricsExporters(
+	specifiedClients []string, host component.Host, nextMetricsConsumer *consumer.MetricsConsumer, logger *zap.Logger,
+) (clients []*metadata.MetadataExporter, wasNil bool) {
+	if specifiedClients == nil {
+		wasNil = true
+		// default to nextMetricsConsumer if no clients have been provided
+		if asMetadataExporter, ok := (*nextMetricsConsumer).(metadata.MetadataExporter); ok {
+			clients = append(clients, &asMetadataExporter)
 		}
+		return
 	}
 
-	if len(consumers) == 0 {
-		logger.Debug("no SFx events are possible as no valid eventClients have been provided and next pipeline component isn't a LogsConsumer")
+	if builtExporters, ok := host.GetExporters()[configmodels.MetricsDataType]; ok {
+		for _, client := range specifiedClients {
+			var found bool
+			for exporterConfig, exporter := range builtExporters {
+				if exporterConfig.Name() == client {
+					if asMetadataExporter, ok := exporter.(metadata.MetadataExporter); ok {
+						clients = append(clients, &asMetadataExporter)
+					}
+					found = true
+				}
+			}
+			if !found {
+				logger.Info(
+					"specified dimension client is not an available exporter",
+					zap.String("client", client),
+				)
+			}
+		}
 	}
-
-	return consumers
+	return
 }
 
-// getClientsFromMetricsExporters will walk through all provided config.DimensionClients and retrieve matching registered
-// MetricsExporters, the only truly supported component type.
-// If config.MetadataClients is nil, it will return a slice with nextConsumer if it's a MetricsExporter.
-func getClientsFromMetricsExporters(
-	specifiedClients []string, host component.Host, nextConsumer *consumer.MetricsConsumer, fieldName string, logger *zap.Logger,
-) (clients []*interface{}) {
-	if specifiedClients == nil {
-		// default to nextConsumer if no clients have been provided
-		asInterface := (*nextConsumer).(interface{})
-		clients = append(clients, &asInterface)
-		return
-	}
-
-	builtExporters := host.GetExporters()[configmodels.MetricsDataType]
-	for _, client := range specifiedClients {
-		var found bool
+func getLoneSFxExporter(host component.Host, exporterType configmodels.DataType) component.Exporter {
+	var sfxExporter component.Exporter
+	if builtExporters, ok := host.GetExporters()[exporterType]; ok {
 		for exporterConfig, exporter := range builtExporters {
-			if exporterConfig.Name() == client {
-				asInterface := exporter.(interface{})
-				clients = append(clients, &asInterface)
-				found = true
+			if exporterConfig.Type() == "signalfx" {
+				if sfxExporter == nil {
+					sfxExporter = exporter
+				} else { // we've already found one so no lone instance to use as default
+					return nil
+				}
+
 			}
 		}
-		if !found {
-			logger.Info(
-				fmt.Sprintf("specified %s is not an available exporter", fieldName),
-				zap.String("client", client),
-			)
-		}
 	}
-	return clients
+	return sfxExporter
+
 }
 
 func (output *Output) AddDatapointExclusionFilter(filter dpfilters.DatapointFilter) {
@@ -177,6 +185,10 @@ func (output *Output) Copy() types.Output {
 }
 
 func (output *Output) SendDatapoints(datapoints ...*datapoint.Datapoint) {
+	if output.nextMetricsConsumer == nil {
+		return
+	}
+
 	ctx := obsreport.ReceiverContext(context.Background(), output.receiverName, internalTransport)
 	ctx = obsreport.StartMetricsReceiveOp(ctx, typeStr, internalTransport)
 
@@ -192,21 +204,19 @@ func (output *Output) SendDatapoints(datapoints ...*datapoint.Datapoint) {
 	}
 
 	_, numPoints := metrics.MetricAndDataPointCount()
-	err := output.nextConsumer.ConsumeMetrics(context.Background(), metrics)
+	err := output.nextMetricsConsumer.ConsumeMetrics(context.Background(), metrics)
 	obsreport.EndMetricsReceiveOp(ctx, typeStr, numPoints, err)
 }
 
 func (output *Output) SendEvent(event *event.Event) {
-	if len(output.nextEventClients) == 0 {
+	if output.nextLogsConsumer == nil {
 		return
 	}
 
 	logRecord := eventToLog(event, output.logger)
-	for _, logsConsumer := range output.nextEventClients {
-		err := (*logsConsumer).ConsumeLogs(context.Background(), logRecord)
-		if err != nil {
-			output.logger.Debug("SendEvent has failed", zap.Error(err))
-		}
+	err := output.nextLogsConsumer.ConsumeLogs(context.Background(), logRecord)
+	if err != nil {
+		output.logger.Debug("SendEvent has failed", zap.Error(err))
 	}
 }