Skip to content

Add support for configuring mutliple OTel pipelines #1167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Jul 22, 2025
Merged

Add support for configuring mutliple OTel pipelines #1167

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
7ed83e0
Add support for configuring mutliple OTel pipelines
dhurley Jul 15, 2025
e12de1a
Update load tests
dhurley Jul 15, 2025
f6b3038
Merge branch 'main' into support-otel-pipelines
dhurley Jul 16, 2025
e0c5a0e
Clean up
dhurley Jul 17, 2025
fa3c36c
Merge branch 'main' into support-otel-pipelines
dhurley Jul 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 42 additions & 49 deletions internal/collector/otel_collector_plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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.")
Expand Down Expand Up @@ -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{
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -541,44 +542,46 @@ 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{
{
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{
{
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
}
Expand All @@ -592,23 +595,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)

Expand Down Expand Up @@ -644,16 +637,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
Expand Down
73 changes: 40 additions & 33 deletions internal/collector/otel_collector_plugin_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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()
Expand All @@ -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()
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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",
},
},
},
},
Expand All @@ -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()
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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{
Expand All @@ -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, 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{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, 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)
})
}

Expand Down
Loading