Add support for configuring mutliple OTel pipelines

Author: dhurley

internal/collector/otel_collector_plugin.go

@@ -59,7 +59,7 @@ var (
 	initMutex            = &sync.Mutex{}
 )
 
-// NewCollector is the constructor for the Collector plugin.
+// New is the constructor for the Collector plugin.
 func New(conf *config.Config) (*Collector, error) {
 	initMutex.Lock()
 
@@ -194,7 +194,7 @@ func (oc *Collector) Subscriptions() []string {
 }
 
 // Process receivers and log warning for sub-optimal configurations
-func (oc *Collector) processReceivers(ctx context.Context, receivers []config.OtlpReceiver) {
+func (oc *Collector) processReceivers(ctx context.Context, receivers map[string]*config.OtlpReceiver) {
 	for _, receiver := range receivers {
 		if receiver.OtlpTLSConfig == nil {
 			slog.WarnContext(ctx, "OTel receiver is configured without TLS. Connections are unencrypted.")
@@ -317,12 +317,13 @@ func (oc *Collector) updateResourceProcessor(resourceUpdateContext *v1.Resource)
 	resourceProcessorUpdated := false
 
 	if oc.config.Collector.Processors.Resource == nil {
-		oc.config.Collector.Processors.Resource = &config.Resource{
+		oc.config.Collector.Processors.Resource = make(map[string]*config.Resource)
+		oc.config.Collector.Processors.Resource["default"] = &config.Resource{
 			Attributes: make([]config.ResourceAttribute, 0),
 		}
 	}
 
-	if oc.config.Collector.Processors.Resource != nil &&
+	if oc.config.Collector.Processors.Resource["default"] != nil &&
 		resourceUpdateContext.GetResourceId() != "" {
 		resourceProcessorUpdated = oc.updateResourceAttributes(
 			[]config.ResourceAttribute{
@@ -431,7 +432,7 @@ func (oc *Collector) checkForNewReceivers(ctx context.Context, nginxConfigContex
 	}
 
 	if oc.config.IsFeatureEnabled(pkgConfig.FeatureLogsNap) {
-		tcplogReceiversFound := oc.updateTcplogReceivers(nginxConfigContext)
+		tcplogReceiversFound := oc.updateNginxAppProtectTcplogReceivers(nginxConfigContext)
 		if tcplogReceiversFound {
 			reloadCollector = true
 		}
@@ -541,49 +542,51 @@ func (oc *Collector) updateExistingNginxOSSReceiver(
 	return nginxReceiverFound, reloadCollector
 }
 
-func (oc *Collector) updateTcplogReceivers(nginxConfigContext *model.NginxConfigContext) bool {
+func (oc *Collector) updateNginxAppProtectTcplogReceivers(nginxConfigContext *model.NginxConfigContext) bool {
 	newTcplogReceiverAdded := false
+
+	if oc.config.Collector.Receivers.TcplogReceivers == nil {
+		oc.config.Collector.Receivers.TcplogReceivers = make(map[string]*config.TcplogReceiver)
+	}
+
 	if nginxConfigContext.NAPSysLogServers != nil {
 	napLoop:
 		for _, napSysLogServer := range nginxConfigContext.NAPSysLogServers {
 			if oc.doesTcplogReceiverAlreadyExist(napSysLogServer) {
 				continue napLoop
 			}
 
-			oc.config.Collector.Receivers.TcplogReceivers = append(
-				oc.config.Collector.Receivers.TcplogReceivers,
-				config.TcplogReceiver{
-					ListenAddress: napSysLogServer,
-					Operators: []config.Operator{
-						{
-							Type: "add",
-							Fields: map[string]string{
-								"field": "body",
-								"value": timestampConversionExpression,
-							},
+			oc.config.Collector.Receivers.TcplogReceivers["nginx_app_protect"] = &config.TcplogReceiver{
+				ListenAddress: napSysLogServer,
+				Operators: []config.Operator{
+					{
+						Type: "add",
+						Fields: map[string]string{
+							"field": "body",
+							"value": timestampConversionExpression,
 						},
-						{
-							Type: "syslog_parser",
-							Fields: map[string]string{
-								"protocol": "rfc3164",
-							},
+					},
+					{
+						Type: "syslog_parser",
+						Fields: map[string]string{
+							"protocol": "rfc3164",
 						},
-						{
-							Type: "remove",
-							Fields: map[string]string{
-								"field": "attributes.message",
-							},
+					},
+					{
+						Type: "remove",
+						Fields: map[string]string{
+							"field": "attributes.message",
 						},
-						{
-							Type: "add",
-							Fields: map[string]string{
-								"field": "resource[\"instance.id\"]",
-								"value": nginxConfigContext.InstanceID,
-							},
+					},
+					{
+						Type: "add",
+						Fields: map[string]string{
+							"field": "resource[\"instance.id\"]",
+							"value": nginxConfigContext.InstanceID,
 						},
 					},
 				},
-			)
+			}
 
 			newTcplogReceiverAdded = true
 		}
@@ -597,23 +600,13 @@ func (oc *Collector) updateTcplogReceivers(nginxConfigContext *model.NginxConfig
 func (oc *Collector) areNapReceiversDeleted(nginxConfigContext *model.NginxConfigContext) bool {
 	listenAddressesToBeDeleted := oc.configDeletedNapReceivers(nginxConfigContext)
 	if len(listenAddressesToBeDeleted) != 0 {
-		oc.deleteNapReceivers(listenAddressesToBeDeleted)
+		delete(oc.config.Collector.Receivers.TcplogReceivers, "nginx_app_protect")
 		return true
 	}
 
 	return false
 }
 
-func (oc *Collector) deleteNapReceivers(listenAddressesToBeDeleted map[string]bool) {
-	filteredReceivers := (oc.config.Collector.Receivers.TcplogReceivers)[:0]
-	for _, receiver := range oc.config.Collector.Receivers.TcplogReceivers {
-		if !listenAddressesToBeDeleted[receiver.ListenAddress] {
-			filteredReceivers = append(filteredReceivers, receiver)
-		}
-	}
-	oc.config.Collector.Receivers.TcplogReceivers = filteredReceivers
-}
-
 func (oc *Collector) configDeletedNapReceivers(nginxConfigContext *model.NginxConfigContext) map[string]bool {
 	elements := make(map[string]bool)
 
@@ -651,16 +644,16 @@ func (oc *Collector) updateResourceAttributes(
 ) (actionUpdated bool) {
 	actionUpdated = false
 
-	if oc.config.Collector.Processors.Resource.Attributes != nil {
+	if oc.config.Collector.Processors.Resource["default"].Attributes != nil {
 	OUTER:
 		for _, toAdd := range attributesToAdd {
-			for _, action := range oc.config.Collector.Processors.Resource.Attributes {
+			for _, action := range oc.config.Collector.Processors.Resource["default"].Attributes {
 				if action.Key == toAdd.Key {
 					continue OUTER
 				}
 			}
-			oc.config.Collector.Processors.Resource.Attributes = append(
-				oc.config.Collector.Processors.Resource.Attributes,
+			oc.config.Collector.Processors.Resource["default"].Attributes = append(
+				oc.config.Collector.Processors.Resource["default"].Attributes,
 				toAdd,
 			)
 			actionUpdated = true

internal/collector/otel_collector_plugin_test.go

@@ -349,12 +349,14 @@ func TestCollector_ProcessResourceUpdateTopic(t *testing.T) {
 				Data:  protos.HostResource(),
 			},
 			processors: config.Processors{
-				Resource: &config.Resource{
-					Attributes: []config.ResourceAttribute{
-						{
-							Key:    "resource.id",
-							Action: "insert",
-							Value:  "1234",
+				Resource: map[string]*config.Resource{
+					"default": {
+						Attributes: []config.ResourceAttribute{
+							{
+								Key:    "resource.id",
+								Action: "insert",
+								Value:  "1234",
+							},
 						},
 					},
 				},
@@ -711,18 +713,19 @@ func TestCollector_updateResourceAttributes(t *testing.T) {
 			collector.service = createFakeCollector()
 
 			// set up Actions
-			conf.Collector.Processors.Resource = &config.Resource{Attributes: test.setup}
+			conf.Collector.Processors.Resource = make(map[string]*config.Resource)
+			conf.Collector.Processors.Resource["default"] = &config.Resource{Attributes: test.setup}
 
 			reloadRequired := collector.updateResourceAttributes(test.attributes)
 			assert.Equal(tt,
 				test.expectedAttribs,
-				conf.Collector.Processors.Resource.Attributes)
+				conf.Collector.Processors.Resource["default"].Attributes)
 			assert.Equal(tt, test.expectedReloadRequired, reloadRequired)
 		})
 	}
 }
 
-func TestCollector_updateTcplogReceivers(t *testing.T) {
+func TestCollector_updateNginxAppProtectTcplogReceivers(t *testing.T) {
 	conf := types.OTelConfig(t)
 	conf.Collector.Log.Path = ""
 	conf.Collector.Processors.Batch = nil
@@ -741,38 +744,43 @@ func TestCollector_updateTcplogReceivers(t *testing.T) {
 	assert.Empty(t, conf.Collector.Receivers.TcplogReceivers)
 
 	t.Run("Test 1: New TcplogReceiver added", func(tt *testing.T) {
-		tcplogReceiverAdded := collector.updateTcplogReceivers(nginxConfigContext)
+		tcplogReceiverAdded := collector.updateNginxAppProtectTcplogReceivers(nginxConfigContext)
 
 		assert.True(tt, tcplogReceiverAdded)
 		assert.Len(tt, conf.Collector.Receivers.TcplogReceivers, 1)
-		assert.Equal(tt, "localhost:151", conf.Collector.Receivers.TcplogReceivers[0].ListenAddress)
-		assert.Len(tt, conf.Collector.Receivers.TcplogReceivers[0].Operators, 4)
+		assert.Equal(tt, "localhost:151", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
+		assert.Len(tt, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 4)
 	})
 
-	// Calling updateTcplogReceivers shouldn't update the TcplogReceivers slice
+	// Calling updateNginxAppProtectTcplogReceivers shouldn't update the TcplogReceivers slice
 	// since there is already a receiver with the same ListenAddress
 	t.Run("Test 2: TcplogReceiver already exists", func(tt *testing.T) {
-		tcplogReceiverAdded := collector.updateTcplogReceivers(nginxConfigContext)
+		tcplogReceiverAdded := collector.updateNginxAppProtectTcplogReceivers(nginxConfigContext)
 		assert.False(t, tcplogReceiverAdded)
 		assert.Len(t, conf.Collector.Receivers.TcplogReceivers, 1)
-		assert.Equal(t, "localhost:151", conf.Collector.Receivers.TcplogReceivers[0].ListenAddress)
-		assert.Len(t, conf.Collector.Receivers.TcplogReceivers[0].Operators, 4)
+		assert.Equal(t, "localhost:151", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
+		assert.Len(t, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 4)
 	})
 
 	t.Run("Test 3: TcplogReceiver deleted", func(tt *testing.T) {
-		tcplogReceiverDeleted := collector.updateTcplogReceivers(&model.NginxConfigContext{})
+		tcplogReceiverDeleted := collector.updateNginxAppProtectTcplogReceivers(&model.NginxConfigContext{})
 		assert.True(t, tcplogReceiverDeleted)
 		assert.Empty(t, conf.Collector.Receivers.TcplogReceivers)
 	})
 
 	t.Run("Test 4: New tcplogReceiver added and deleted another", func(tt *testing.T) {
-		tcplogReceiverDeleted := collector.updateTcplogReceivers(&model.NginxConfigContext{NAPSysLogServers: []string{
-			"localhost:152",
-		}})
+		tcplogReceiverDeleted := collector.updateNginxAppProtectTcplogReceivers(
+			&model.NginxConfigContext{
+				NAPSysLogServers: []string{
+					"localhost:152",
+				},
+			},
+		)
+
 		assert.True(t, tcplogReceiverDeleted)
 		assert.Len(t, conf.Collector.Receivers.TcplogReceivers, 1)
-		assert.Equal(t, "localhost:152", conf.Collector.Receivers.TcplogReceivers[0].ListenAddress)
-		assert.Len(t, conf.Collector.Receivers.TcplogReceivers[0].Operators, 4)
+		assert.Equal(t, "localhost:152", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
+		assert.Len(t, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 4)
 	})
 }