Skip to content

[beatreceiver] Drop failed doc inputs from ES exporter #8738

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

Closed
khushijain21 wants to merge 4 commits into from
Closed

[beatreceiver] Drop failed doc inputs from ES exporter #8738

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fbad1ac
[beatreceiver] Drop failed doc inputs from ES exporter
khushijain21 Jun 30, 2025
d129e7b
Merge remote-tracking branch 'upstream/main' into eventLogs
khushijain21 Jul 1, 2025
610db82
add test
khushijain21 Jul 1, 2025
38b578c
Merge branch 'main' into eventLogs
khushijain21 Jul 1, 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
4 changes: 4 additions & 0 deletions internal/pkg/agent/application/monitoring/testdata/monitoring_config_full.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,10 @@ inputs:
when:
regexp:
message: ^Non-zero metrics in the last
- drop_event:
when:
regexp:
input: \{"create":\{"_index":.*
- copy_fields:
fields:
- from: data_stream.dataset
Expand Down
15 changes: 15 additions & 0 deletions internal/pkg/agent/application/monitoring/v1_monitor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -831,6 +831,8 @@ func processorsForAgentFilestream() []any {
dropEventsFromMonitoringComponentsProcessor(),
// drop periodic metrics logs (those are useful mostly in diagnostic dumps where we collect log files)
dropPeriodicMetricsLogsProcessor(),
// drop sensitive information from ES exporter ensuring we do not send it to fleet
dropSensitiveInfoFromESExporter(),
}
// if the event is from a component, use the component's dataset
processors = append(processors, useComponentDatasetProcessors()...)
Expand Down Expand Up @@ -1152,6 +1154,19 @@ func dropEcsVersionFieldProcessor() map[string]any {
}
}

// dropSensitiveInfoFromESExporter returns a processor which drops any sensitive information logged by ES exporter
func dropSensitiveInfoFromESExporter() map[string]any {
return map[string]interface{}{
"drop_event": map[string]interface{}{
"when": map[string]interface{}{
"regexp": map[string]interface{}{
"input": `\{"create":\{"_index":.*`,
Copy link
Contributor Author

@khushijain21 khushijain21 Jun 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ES exporter always uses a create action when indexing log records. If a document fails for any reason and the original input is part of the response - the log message will always contain

"input":"{\"create\":{\"_index\": some-index-name. .... // the rest of the input

We drop this event to ensure sensitive data is not sent to fleet.

Copy link
Member

@cmacknz cmacknz Jun 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When the collector is a sub-process, we could also look at directing these logs into our event log using a similar pattern to preserve the existing behavior where logs from events can't clutter the main logs. We could perhaps do this now I just know once we are a sub-process we'll be consuming the logs from the stdout/stderr stream instead of via our own in-process logger.

CC @pkoutsovasilis.

Copy link
Contributor Author

@khushijain21 khushijain21 Jul 1, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that sounds good and is possible to do.
This is the code change we would need to do when otel collector runs as sub process.

https://github.com/elastic/elastic-agent/compare/main...khushijain21:elastic-agent:collectorEvent?expand=

that would make the current changeset unnecessary - maybe we should wait until the collector runs as a subprocess by default. wdyt?

Copy link
Contributor Author

@khushijain21 khushijain21 Jul 1, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I did a quick testing and ndjson parser in filebeat monitoring https://github.com/khushijain21/elastic-agent/blob/eventLogs/internal/pkg/agent/application/monitoring/v1_monitor.go#L579 does not parse json recursively.

"input":"{\"create\":{\"_index\": some-index-name. .... // the rest of the input

This field is never parsed and indexed. We should be good even without dropping this event explicitly :))

Copy link
Contributor Author

@khushijain21 khushijain21 Jul 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cc : @cmacknz

Copy link
Member

@cmacknz cmacknz Jul 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This field is never parsed and indexed. We should be good even without dropping this event explicitly :))

OK this is nice, I think we should have an explicit test for this behavior as it is a security concern (in agent, that these errors cannot be indexed). Can you add one as a follow up?

We have similar tests of the event log functionality.

},
},
},
}
}

// addFormattedIndexProcessor returns a processor which sets the destination index for an event based on a format string.
func addFormattedIndexProcessor() map[string]any {
return map[string]any{
Expand Down
8 changes: 5 additions & 3 deletions internal/pkg/otel/translate/otelconfig.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -401,12 +401,14 @@ func translateEsOutputToExporter(cfg *config.C) (map[string]any, error) {
if err != nil {
return nil, err
}

// logs failed documents at debug level
esConfig["telemetry"] = map[string]any{
"log_failed_docs_input": true,
}
// dynamic indexing works by default

// we also want to use dynamic log ids
esConfig["logs_dynamic_id"] = map[string]any{"enabled": true}

// for compatibility with beats, we want bodymap mapping
esConfig["mapping"] = map[string]any{"mode": "bodymap"}
return esConfig, nil
}
Loading