Skip to content

Commit 04828e6

Browse files
dhurleyspencerugbo
dhurley
authored and
spencerugbo
committed
Add support for configuring mutliple OTel pipelines (#1167)
1 parent 9f04dbf commit 04828e6

File tree

15 files changed

+664
-432
lines changed

15 files changed

+664
-432
lines changed

‎internal/collector/otel_collector_plugin.go‎

Lines changed: 42 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ var (
6060
)
6161

6262
// NewCollector is the constructor for the Collector plugin.
63-
func New(conf *config.Config) (*Collector, error) {
63+
func NewCollector(conf *config.Config) (*Collector, error) {
6464
initMutex.Lock()
6565

6666
defer initMutex.Unlock()
@@ -194,7 +194,7 @@ func (oc *Collector) Subscriptions() []string {
194194
}
195195

196196
// Process receivers and log warning for sub-optimal configurations
197-
func (oc *Collector) processReceivers(ctx context.Context, receivers []config.OtlpReceiver) {
197+
func (oc *Collector) processReceivers(ctx context.Context, receivers map[string]*config.OtlpReceiver) {
198198
for _, receiver := range receivers {
199199
if receiver.OtlpTLSConfig == nil {
200200
slog.WarnContext(ctx, "OTel receiver is configured without TLS. Connections are unencrypted.")
@@ -317,12 +317,13 @@ func (oc *Collector) updateResourceProcessor(resourceUpdateContext *v1.Resource)
317317
resourceProcessorUpdated := false
318318

319319
if oc.config.Collector.Processors.Resource == nil {
320-
oc.config.Collector.Processors.Resource = &config.Resource{
320+
oc.config.Collector.Processors.Resource = make(map[string]*config.Resource)
321+
oc.config.Collector.Processors.Resource["default"] = &config.Resource{
321322
Attributes: make([]config.ResourceAttribute, 0),
322323
}
323324
}
324325

325-
if oc.config.Collector.Processors.Resource != nil &&
326+
if oc.config.Collector.Processors.Resource["default"] != nil &&
326327
resourceUpdateContext.GetResourceId() != "" {
327328
resourceProcessorUpdated = oc.updateResourceAttributes(
328329
[]config.ResourceAttribute{
@@ -431,7 +432,7 @@ func (oc *Collector) checkForNewReceivers(ctx context.Context, nginxConfigContex
431432
}
432433

433434
if oc.config.IsFeatureEnabled(pkgConfig.FeatureLogsNap) {
434-
tcplogReceiversFound := oc.updateTcplogReceivers(nginxConfigContext)
435+
tcplogReceiversFound := oc.updateNginxAppProtectTcplogReceivers(nginxConfigContext)
435436
if tcplogReceiversFound {
436437
reloadCollector = true
437438
}
@@ -541,44 +542,46 @@ func (oc *Collector) updateExistingNginxOSSReceiver(
541542
return nginxReceiverFound, reloadCollector
542543
}
543544

544-
func (oc *Collector) updateTcplogReceivers(nginxConfigContext *model.NginxConfigContext) bool {
545+
func (oc *Collector) updateNginxAppProtectTcplogReceivers(nginxConfigContext *model.NginxConfigContext) bool {
545546
newTcplogReceiverAdded := false
547+
548+
if oc.config.Collector.Receivers.TcplogReceivers == nil {
549+
oc.config.Collector.Receivers.TcplogReceivers = make(map[string]*config.TcplogReceiver)
550+
}
551+
546552
if nginxConfigContext.NAPSysLogServer != "" {
547553
if !oc.doesTcplogReceiverAlreadyExist(nginxConfigContext.NAPSysLogServer) {
548-
oc.config.Collector.Receivers.TcplogReceivers = append(
549-
oc.config.Collector.Receivers.TcplogReceivers,
550-
config.TcplogReceiver{
551-
ListenAddress: nginxConfigContext.NAPSysLogServer,
552-
Operators: []config.Operator{
553-
{
554-
Type: "add",
555-
Fields: map[string]string{
556-
"field": "body",
557-
"value": timestampConversionExpression,
558-
},
554+
oc.config.Collector.Receivers.TcplogReceivers["nginx_app_protect"] = &config.TcplogReceiver{
555+
ListenAddress: nginxConfigContext.NAPSysLogServer,
556+
Operators: []config.Operator{
557+
{
558+
Type: "add",
559+
Fields: map[string]string{
560+
"field": "body",
561+
"value": timestampConversionExpression,
559562
},
560-
{
561-
Type: "syslog_parser",
562-
Fields: map[string]string{
563-
"protocol": "rfc3164",
564-
},
563+
},
564+
{
565+
Type: "syslog_parser",
566+
Fields: map[string]string{
567+
"protocol": "rfc3164",
565568
},
566-
{
567-
Type: "remove",
568-
Fields: map[string]string{
569-
"field": "attributes.message",
570-
},
569+
},
570+
{
571+
Type: "remove",
572+
Fields: map[string]string{
573+
"field": "attributes.message",
571574
},
572-
{
573-
Type: "add",
574-
Fields: map[string]string{
575-
"field": "resource[\"instance.id\"]",
576-
"value": nginxConfigContext.InstanceID,
577-
},
575+
},
576+
{
577+
Type: "add",
578+
Fields: map[string]string{
579+
"field": "resource[\"instance.id\"]",
580+
"value": nginxConfigContext.InstanceID,
578581
},
579582
},
580583
},
581-
)
584+
}
582585

583586
newTcplogReceiverAdded = true
584587
}
@@ -592,23 +595,13 @@ func (oc *Collector) updateTcplogReceivers(nginxConfigContext *model.NginxConfig
592595
func (oc *Collector) areNapReceiversDeleted(nginxConfigContext *model.NginxConfigContext) bool {
593596
listenAddressesToBeDeleted := oc.configDeletedNapReceivers(nginxConfigContext)
594597
if len(listenAddressesToBeDeleted) != 0 {
595-
oc.deleteNapReceivers(listenAddressesToBeDeleted)
598+
delete(oc.config.Collector.Receivers.TcplogReceivers, "nginx_app_protect")
596599
return true
597600
}
598601

599602
return false
600603
}
601604

602-
func (oc *Collector) deleteNapReceivers(listenAddressesToBeDeleted map[string]bool) {
603-
filteredReceivers := (oc.config.Collector.Receivers.TcplogReceivers)[:0]
604-
for _, receiver := range oc.config.Collector.Receivers.TcplogReceivers {
605-
if !listenAddressesToBeDeleted[receiver.ListenAddress] {
606-
filteredReceivers = append(filteredReceivers, receiver)
607-
}
608-
}
609-
oc.config.Collector.Receivers.TcplogReceivers = filteredReceivers
610-
}
611-
612605
func (oc *Collector) configDeletedNapReceivers(nginxConfigContext *model.NginxConfigContext) map[string]bool {
613606
elements := make(map[string]bool)
614607

@@ -644,16 +637,16 @@ func (oc *Collector) updateResourceAttributes(
644637
) (actionUpdated bool) {
645638
actionUpdated = false
646639

647-
if oc.config.Collector.Processors.Resource.Attributes != nil {
640+
if oc.config.Collector.Processors.Resource["default"].Attributes != nil {
648641
OUTER:
649642
for _, toAdd := range attributesToAdd {
650-
for _, action := range oc.config.Collector.Processors.Resource.Attributes {
643+
for _, action := range oc.config.Collector.Processors.Resource["default"].Attributes {
651644
if action.Key == toAdd.Key {
652645
continue OUTER
653646
}
654647
}
655-
oc.config.Collector.Processors.Resource.Attributes = append(
656-
oc.config.Collector.Processors.Resource.Attributes,
648+
oc.config.Collector.Processors.Resource["default"].Attributes = append(
649+
oc.config.Collector.Processors.Resource["default"].Attributes,
657650
toAdd,
658651
)
659652
actionUpdated = true

‎internal/collector/otel_collector_plugin_test.go‎

Lines changed: 40 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ func TestCollector_New(t *testing.T) {
6767

6868
for _, tt := range tests {
6969
t.Run(tt.name, func(t *testing.T) {
70-
collector, err := New(tt.config)
70+
collector, err := NewCollector(tt.config)
7171

7272
if tt.expectedError != nil {
7373
require.Error(t, err)
@@ -114,7 +114,7 @@ func TestCollector_Init(t *testing.T) {
114114
conf.Collector.Receivers = config.Receivers{}
115115
}
116116

117-
collector, err = New(conf)
117+
collector, err = NewCollector(conf)
118118
require.NoError(t, err, "NewCollector should not return an error with valid config")
119119

120120
collector.service = createFakeCollector()
@@ -133,7 +133,7 @@ func TestCollector_InitAndClose(t *testing.T) {
133133
conf := types.OTelConfig(t)
134134
conf.Collector.Log.Path = ""
135135

136-
collector, err := New(conf)
136+
collector, err := NewCollector(conf)
137137
require.NoError(t, err, "NewCollector should not return an error with valid config")
138138

139139
ctx := context.Background()
@@ -293,7 +293,7 @@ func TestCollector_ProcessNginxConfigUpdateTopic(t *testing.T) {
293293
conf.Collector.Extensions.HeadersSetter = nil
294294
conf.Collector.Exporters.PrometheusExporter = nil
295295

296-
collector, err := New(conf)
296+
collector, err := NewCollector(conf)
297297
require.NoError(tt, err, "NewCollector should not return an error with valid config")
298298

299299
collector.service = createFakeCollector()
@@ -349,12 +349,14 @@ func TestCollector_ProcessResourceUpdateTopic(t *testing.T) {
349349
Data: protos.HostResource(),
350350
},
351351
processors: config.Processors{
352-
Resource: &config.Resource{
353-
Attributes: []config.ResourceAttribute{
354-
{
355-
Key: "resource.id",
356-
Action: "insert",
357-
Value: "1234",
352+
Resource: map[string]*config.Resource{
353+
"default": {
354+
Attributes: []config.ResourceAttribute{
355+
{
356+
Key: "resource.id",
357+
Action: "insert",
358+
Value: "1234",
359+
},
358360
},
359361
},
360362
},
@@ -376,7 +378,7 @@ func TestCollector_ProcessResourceUpdateTopic(t *testing.T) {
376378

377379
for _, test := range tests {
378380
t.Run(test.name, func(tt *testing.T) {
379-
collector, err := New(conf)
381+
collector, err := NewCollector(conf)
380382
require.NoError(tt, err, "NewCollector should not return an error with valid config")
381383

382384
collector.service = createFakeCollector()
@@ -437,7 +439,7 @@ func TestCollector_ProcessResourceUpdateTopicFails(t *testing.T) {
437439

438440
for _, test := range tests {
439441
t.Run(test.name, func(tt *testing.T) {
440-
collector, err := New(conf)
442+
collector, err := NewCollector(conf)
441443
require.NoError(tt, err, "NewCollector should not return an error with valid config")
442444

443445
collector.service = createFakeCollector()
@@ -559,7 +561,7 @@ func TestCollector_updateExistingNginxOSSReceiver(t *testing.T) {
559561
for _, test := range tests {
560562
t.Run(test.name, func(tt *testing.T) {
561563
conf.Collector.Receivers = test.existingReceivers
562-
collector, err := New(conf)
564+
collector, err := NewCollector(conf)
563565
require.NoError(tt, err, "NewCollector should not return an error with valid config")
564566

565567
collector.service = createFakeCollector()
@@ -650,7 +652,7 @@ func TestCollector_updateExistingNginxPlusReceiver(t *testing.T) {
650652
for _, test := range tests {
651653
t.Run(test.name, func(tt *testing.T) {
652654
conf.Collector.Receivers = test.existingReceivers
653-
collector, err := New(conf)
655+
collector, err := NewCollector(conf)
654656
require.NoError(tt, err, "NewCollector should not return an error with valid config")
655657

656658
collector.service = createFakeCollector()
@@ -705,31 +707,32 @@ func TestCollector_updateResourceAttributes(t *testing.T) {
705707

706708
for _, test := range tests {
707709
t.Run(test.name, func(tt *testing.T) {
708-
collector, err := New(conf)
710+
collector, err := NewCollector(conf)
709711
require.NoError(tt, err, "NewCollector should not return an error with valid config")
710712

711713
collector.service = createFakeCollector()
712714

713715
// set up Actions
714-
conf.Collector.Processors.Resource = &config.Resource{Attributes: test.setup}
716+
conf.Collector.Processors.Resource = make(map[string]*config.Resource)
717+
conf.Collector.Processors.Resource["default"] = &config.Resource{Attributes: test.setup}
715718

716719
reloadRequired := collector.updateResourceAttributes(test.attributes)
717720
assert.Equal(tt,
718721
test.expectedAttribs,
719-
conf.Collector.Processors.Resource.Attributes)
722+
conf.Collector.Processors.Resource["default"].Attributes)
720723
assert.Equal(tt, test.expectedReloadRequired, reloadRequired)
721724
})
722725
}
723726
}
724727

725-
func TestCollector_updateTcplogReceivers(t *testing.T) {
728+
func TestCollector_updateNginxAppProtectTcplogReceivers(t *testing.T) {
726729
conf := types.OTelConfig(t)
727730
conf.Collector.Log.Path = ""
728731
conf.Collector.Processors.Batch = nil
729732
conf.Collector.Processors.Attribute = nil
730733
conf.Collector.Processors.Resource = nil
731734
conf.Collector.Processors.LogsGzip = nil
732-
collector, err := New(conf)
735+
collector, err := NewCollector(conf)
733736
require.NoError(t, err)
734737

735738
nginxConfigContext := &model.NginxConfigContext{
@@ -738,38 +741,42 @@ func TestCollector_updateTcplogReceivers(t *testing.T) {
738741

739742
assert.Empty(t, conf.Collector.Receivers.TcplogReceivers)
740743

741-
t.Run("Test 1: New TcplogReceiver added", func(tt *testing.T) {
742-
tcplogReceiverAdded := collector.updateTcplogReceivers(nginxConfigContext)
744+
t.Run("Test 1: NewCollector TcplogReceiver added", func(tt *testing.T) {
745+
tcplogReceiverAdded := collector.updateNginxAppProtectTcplogReceivers(nginxConfigContext)
743746

744747
assert.True(tt, tcplogReceiverAdded)
745748
assert.Len(tt, conf.Collector.Receivers.TcplogReceivers, 1)
746-
assert.Equal(tt, "localhost:151", conf.Collector.Receivers.TcplogReceivers[0].ListenAddress)
747-
assert.Len(tt, conf.Collector.Receivers.TcplogReceivers[0].Operators, 4)
749+
assert.Equal(tt, "localhost:151", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
750+
assert.Len(tt, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 4)
748751
})
749752

750-
// Calling updateTcplogReceivers shouldn't update the TcplogReceivers slice
753+
// Calling updateNginxAppProtectTcplogReceivers shouldn't update the TcplogReceivers slice
751754
// since there is already a receiver with the same ListenAddress
752755
t.Run("Test 2: TcplogReceiver already exists", func(tt *testing.T) {
753-
tcplogReceiverAdded := collector.updateTcplogReceivers(nginxConfigContext)
756+
tcplogReceiverAdded := collector.updateNginxAppProtectTcplogReceivers(nginxConfigContext)
754757
assert.False(t, tcplogReceiverAdded)
755758
assert.Len(t, conf.Collector.Receivers.TcplogReceivers, 1)
756-
assert.Equal(t, "localhost:151", conf.Collector.Receivers.TcplogReceivers[0].ListenAddress)
757-
assert.Len(t, conf.Collector.Receivers.TcplogReceivers[0].Operators, 4)
759+
assert.Equal(t, "localhost:151", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
760+
assert.Len(t, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 4)
758761
})
759762

760763
t.Run("Test 3: TcplogReceiver deleted", func(tt *testing.T) {
761-
tcplogReceiverDeleted := collector.updateTcplogReceivers(&model.NginxConfigContext{})
764+
tcplogReceiverDeleted := collector.updateNginxAppProtectTcplogReceivers(&model.NginxConfigContext{})
762765
assert.True(t, tcplogReceiverDeleted)
763766
assert.Empty(t, conf.Collector.Receivers.TcplogReceivers)
764767
})
765768

766-
t.Run("Test 4: New tcplogReceiver added and deleted another", func(tt *testing.T) {
767-
tcplogReceiverDeleted := collector.
768-
updateTcplogReceivers(&model.NginxConfigContext{NAPSysLogServer: "localhost:152"})
769+
t.Run("Test 4: NewCollector tcplogReceiver added and deleted another", func(tt *testing.T) {
770+
tcplogReceiverDeleted := collector.updateNginxAppProtectTcplogReceivers(
771+
&model.NginxConfigContext{
772+
NAPSysLogServer: "localhost:152",
773+
},
774+
)
775+
769776
assert.True(t, tcplogReceiverDeleted)
770777
assert.Len(t, conf.Collector.Receivers.TcplogReceivers, 1)
771-
assert.Equal(t, "localhost:152", conf.Collector.Receivers.TcplogReceivers[0].ListenAddress)
772-
assert.Len(t, conf.Collector.Receivers.TcplogReceivers[0].Operators, 4)
778+
assert.Equal(t, "localhost:152", conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].ListenAddress)
779+
assert.Len(t, conf.Collector.Receivers.TcplogReceivers["nginx_app_protect"].Operators, 4)
773780
})
774781
}
775782

0 commit comments

Comments
 (0)