merge main

Author: aphralG

internal/collector/otel_collector_plugin.go

@@ -60,7 +60,7 @@ var (
 )
 
 // NewCollector is the constructor for the Collector plugin.
-func New(conf *config.Config) (*Collector, error) {
+func NewCollector(conf *config.Config) (*Collector, error) {
 	initMutex.Lock()
 
 	defer initMutex.Unlock()
@@ -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,62 +542,64 @@ 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.NAPSysLogServer != "" {
 		if !oc.doesTcplogReceiverAlreadyExist(nginxConfigContext.NAPSysLogServer) {
-			oc.config.Collector.Receivers.TcplogReceivers = append(
-				oc.config.Collector.Receivers.TcplogReceivers,
-				config.TcplogReceiver{
-					ListenAddress: nginxConfigContext.NAPSysLogServer,
-					Operators: []config.Operator{
-						// regex captures the priority number from the log line
-						{
-							Type: "regex_parser",
-							Fields: map[string]string{
-								"regex":      "^<(?P<priority>\\d+)>",
-								"parse_from": "body",
-								"parse_to":   "attributes",
-							},
-						},
-						// filter drops all logs that have a severity above 4
-						// https://docs.secureauth.com/0902/en/how-to-read-a-syslog-message.html#severity-code-table
-						{
-							Type: "filter",
-							Fields: map[string]string{
-								"expr":       "'int(attributes.priority) % 8 > 4'",
-								"drop_ratio": "1.0",
-							},
-						},
-						{
-							Type: "add",
-							Fields: map[string]string{
-								"field": "body",
-								"value": timestampConversionExpression,
-							},
+			oc.config.Collector.Receivers.TcplogReceivers["nginx_app_protect"] = &config.TcplogReceiver{
+				ListenAddress: nginxConfigContext.NAPSysLogServer,
+				Operators: []config.Operator{
+				// regex captures the priority number from the log line
+				{
+					Type: "regex_parser",
+					Fields: map[string]string{
+						"regex":      "^<(?P<priority>\\d+)>",
+						"parse_from": "body",
+						"parse_to":   "attributes",
+					},
+				},
+				// filter drops all logs that have a severity above 4
+				// https://docs.secureauth.com/0902/en/how-to-read-a-syslog-message.html#severity-code-table
+				{
+					Type: "filter",
+					Fields: map[string]string{
+						"expr":       "'int(attributes.priority) % 8 > 4'",
+						"drop_ratio": "1.0",
+					},
+				},
+					{
+						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
 		}
@@ -610,23 +613,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)
 
@@ -662,16 +655,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

@@ -67,7 +67,7 @@ func TestCollector_New(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			collector, err := New(tt.config)
+			collector, err := NewCollector(tt.config)
 
 			if tt.expectedError != nil {
 				require.Error(t, err)
@@ -114,7 +114,7 @@ func TestCollector_Init(t *testing.T) {
 				conf.Collector.Receivers = config.Receivers{}
 			}
 
-			collector, err = New(conf)
+			collector, err = NewCollector(conf)
 			require.NoError(t, err, "NewCollector should not return an error with valid config")
 
 			collector.service = createFakeCollector()
@@ -133,7 +133,7 @@ func TestCollector_InitAndClose(t *testing.T) {
 	conf := types.OTelConfig(t)
 	conf.Collector.Log.Path = ""
 
-	collector, err := New(conf)
+	collector, err := NewCollector(conf)
 	require.NoError(t, err, "NewCollector should not return an error with valid config")
 
 	ctx := context.Background()
@@ -293,7 +293,7 @@ func TestCollector_ProcessNginxConfigUpdateTopic(t *testing.T) {
 			conf.Collector.Extensions.HeadersSetter = nil
 			conf.Collector.Exporters.PrometheusExporter = nil
 
-			collector, err := New(conf)
+			collector, err := NewCollector(conf)
 			require.NoError(tt, err, "NewCollector should not return an error with valid config")
 
 			collector.service = createFakeCollector()
@@ -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",
+							},
 						},
 					},
 				},
@@ -376,7 +378,7 @@ func TestCollector_ProcessResourceUpdateTopic(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			collector, err := New(conf)
+			collector, err := NewCollector(conf)
 			require.NoError(tt, err, "NewCollector should not return an error with valid config")
 
 			collector.service = createFakeCollector()
@@ -437,7 +439,7 @@ func TestCollector_ProcessResourceUpdateTopicFails(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			collector, err := New(conf)
+			collector, err := NewCollector(conf)
 			require.NoError(tt, err, "NewCollector should not return an error with valid config")
 
 			collector.service = createFakeCollector()
@@ -559,7 +561,7 @@ func TestCollector_updateExistingNginxOSSReceiver(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
 			conf.Collector.Receivers = test.existingReceivers
-			collector, err := New(conf)
+			collector, err := NewCollector(conf)
 			require.NoError(tt, err, "NewCollector should not return an error with valid config")
 
 			collector.service = createFakeCollector()
@@ -650,7 +652,7 @@ func TestCollector_updateExistingNginxPlusReceiver(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
 			conf.Collector.Receivers = test.existingReceivers
-			collector, err := New(conf)
+			collector, err := NewCollector(conf)
 			require.NoError(tt, err, "NewCollector should not return an error with valid config")
 
 			collector.service = createFakeCollector()
@@ -705,31 +707,32 @@ func TestCollector_updateResourceAttributes(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			collector, err := New(conf)
+			collector, err := NewCollector(conf)
 			require.NoError(tt, err, "NewCollector should not return an error with valid config")
 
 			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
 	conf.Collector.Processors.Attribute = nil
 	conf.Collector.Processors.Resource = nil
 	conf.Collector.Processors.LogsGzip = nil
-	collector, err := New(conf)
+	collector, err := NewCollector(conf)
 	require.NoError(t, err)
 
 	nginxConfigContext := &model.NginxConfigContext{
@@ -738,38 +741,42 @@ 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)
+	t.Run("Test 1: NewCollector TcplogReceiver added", func(tt *testing.T) {
+		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, 6)
+		assert.Equal(tt, "localhost:151", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
+		assert.Len(tt, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 6)
 	})
 
-	// 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, 6)
+		assert.Equal(t, "localhost:151", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
+		assert.Len(t, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 6)
 	})
 
 	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{NAPSysLogServer: "localhost:152"})
+	t.Run("Test 4: NewCollector tcplogReceiver added and deleted another", func(tt *testing.T) {
+		tcplogReceiverDeleted := collector.updateNginxAppProtectTcplogReceivers(
+			&model.NginxConfigContext{
+				NAPSysLogServer: "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, 6)
+		assert.Equal(t, "localhost:152", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
+		assert.Len(t, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 6)
 	})
 }