[exporter/elasticsearch] Dynamically route documents by default #38500
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -55,8 +55,7 @@ func newDocumentRouter(mode MappingMode, dynamicIndex bool, defaultIndex string, | |
| var router documentRouter | ||
| if dynamicIndex { | ||
| router = dynamicDocumentRouter{ | ||
| mode: mode, | ||
| } | ||
| } else { | ||
| router = staticDocumentRouter{ | ||
|
|
@@ -94,24 +93,23 @@ func (r staticDocumentRouter) route(_ pcommon.Resource, _ pcommon.Instrumentatio | |
| } | ||
|
|
||
| type dynamicDocumentRouter struct { | ||
| mode MappingMode | ||
| } | ||
|
|
||
| func (r dynamicDocumentRouter) routeLogRecord(resource pcommon.Resource, scope pcommon.InstrumentationScope, recordAttrs pcommon.Map) (elasticsearch.Index, error) { | ||
| return routeRecord(resource, scope, recordAttrs, r.mode, defaultDataStreamTypeLogs) | ||
| } | ||
|
|
||
| func (r dynamicDocumentRouter) routeDataPoint(resource pcommon.Resource, scope pcommon.InstrumentationScope, recordAttrs pcommon.Map) (elasticsearch.Index, error) { | ||
| return routeRecord(resource, scope, recordAttrs, r.mode, defaultDataStreamTypeMetrics) | ||
| } | ||
|
|
||
| func (r dynamicDocumentRouter) routeSpan(resource pcommon.Resource, scope pcommon.InstrumentationScope, recordAttrs pcommon.Map) (elasticsearch.Index, error) { | ||
| return routeRecord(resource, scope, recordAttrs, r.mode, defaultDataStreamTypeTraces) | ||
| } | ||
|
|
||
| func (r dynamicDocumentRouter) routeSpanEvent(resource pcommon.Resource, scope pcommon.InstrumentationScope, recordAttrs pcommon.Map) (elasticsearch.Index, error) { | ||
| return routeRecord(resource, scope, recordAttrs, r.mode, defaultDataStreamTypeLogs) | ||
| } | ||
|
|
||
| type logstashDocumentRouter struct { | ||
|
|
@@ -150,20 +148,25 @@ func routeRecord( | |
| resource pcommon.Resource, | ||
| scope pcommon.InstrumentationScope, | ||
| recordAttr pcommon.Map, | ||
| mode MappingMode, | ||
| defaultDSType string, | ||
| ) (elasticsearch.Index, error) { | ||
| resourceAttr := resource.Attributes() | ||
| scopeAttr := scope.Attributes() | ||
|
|
||
| // Order: | ||
| // 1. elasticsearch.index from attributes | ||
| // 2. read data_stream.* from attributes | ||
| // 3. receiver-based routing | ||
| // 4. use default hardcoded data_stream.* | ||
| if esIndex, esIndexExists := getFromAttributes(elasticsearch.IndexAttributeName, "", recordAttr, scopeAttr, resourceAttr); esIndexExists { | ||
| // Advanced users can route documents by setting IndexAttributeName in a processor earlier in the pipeline. | ||
| // If `data_stream.*` needs to be set in the document, users should use `data_stream.*` attributes. | ||
| return elasticsearch.Index{Index: esIndex}, nil | ||
| } | ||
|
|
||
| dataset, _ := getFromAttributes(elasticsearch.DataStreamDataset, defaultDataStreamDataset, recordAttr, scopeAttr, resourceAttr) | ||
| namespace, _ := getFromAttributes(elasticsearch.DataStreamNamespace, defaultDataStreamNamespace, recordAttr, scopeAttr, resourceAttr) | ||
|
|
||
| dsType := defaultDSType | ||
| // if mapping mode is bodymap, allow overriding data_stream.type | ||
|
|
@@ -174,15 +177,6 @@ func routeRecord( | |
| } | ||
| } | ||
|
|
||
| // Receiver-based routing | ||
|
There was a problem hiding this comment. It seems that receiver-based routing has a higher precedence compared to the There was a problem hiding this comment. Nice catch. Fixed the bug and added a test: 5c9c85a |
||
| // For example, hostmetricsreceiver (or hostmetricsreceiver.otel in the OTel output mode) | ||
| // for the scope name | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should we use
logs_indexfor span events? Or maybe add another option for span events?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It isn't intuitive to have logs_index as the configuration of destination of span events. What about a new config
traces_span_events_index(that is only applicable to otel mapping mode as it sends span events as separate documents)? It should default to empty, such that dynamic routing kicks in.I have also considered
logs_indexfor span events. See reasoning below.traces_span_events_indexaslogs_index, but it seems to a big trap. Let's imagine if a user wants logs to be statically routed to anlogs_indexand traces dynamically routed. If the user sets logs_index, span events suddenly are not dynamically routed. This is not great.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
As we model span events as logs in OTel mode, I think it's only logical to use
logs_index. Ack that this may not be intuitive but documenting this seems like the solution. I think we should start simple by just usinglogs_indexand only adding something liketraces_span_events_indexif necessary.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
sgtm. It is explicitly stated that it is not recommended to set
logs_index,metrics_indexandtraces_indexanyway.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done in 61a3bfa