opensearch-project / opensearch-project/data-prepper
[BUG] Duplicate key exception when integrate with open-telemetry exporter
Nobody has claimed this yet.
- Dominant language
- Java
- Stars
- 374
- Forks
- 354
- Avg merge
- 3d 18h
- Merged PRs (30d)
- 8
Description
Describe the bug
Data prepper throws java.lang.IllegalStateException: Duplicate key xxx when ingest through log pipeline.
See the error below, due to open telemetry exporter exporting a duplicate ConnectionId key. Not sure this is expected behavior or not. Please help considering to fix this issue.
2023-12-29T07:28:02,516 [pool-13-thread-1] ERROR org.opensearch.dataprepper.plugins.source.otellogs.OTelLogsGrpcService - Failed to parse the request resource_logs {
resource {
...truncate...
scope_logs {
scope {
name: "Microsoft.Extensions.Hosting.Internal.Host"
}
log_records {
time_unix_nano: 1703834880595497100
severity_number: SEVERITY_NUMBER_DEBUG
severity_text: "Debug"
body {
string_value: "Connection id \"{ConnectionId}\" sending FIN because: \"{Reason}\""
}
attributes {
key: "ConnectionId"
value {
string_value: "0HN087UNTCNA9"
}
}
attributes {
key: "Reason"
value {
string_value: "The Socket transport\'s send loop completed gracefully."
}
}
attributes {
key: "ConnectionId"
value {
string_value: "0HN087UNTCNA9"
}
}
observed_time_unix_nano: 1703834880595497100
}
}
}
java.lang.IllegalStateException: Duplicate key log.attributes.ConnectionId (attempted merging values 0HN087UNTCNA9 and 0HN087UNTCNA9)
at java.base/java.util.stream.Collectors.duplicateKeyException(Collectors.java:135) ~[?:?]
at java.base/java.util.stream.Collectors.lambda$uniqKeysMapAccumulator$1(Collectors.java:182) ~[?:?]
at java.base/java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169) ~[?:?]
at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921) ~[?:?]
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682) ~[?:?]
at org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCodec.unpackKeyValueListLog(OTelProtoCodec.java:1094) ~[otel-proto-common-2.6.1.jar:?]
at org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCodec$OTelProtoDecoder.lambda$processLogsList$7(OTelProtoCodec.java:376) ~[otel-proto-common-2.6.1.jar:?]
To Reproduce
Steps to reproduce the behavior:
- I have pushed the sample repository, please visit here
- Run
docker-compose -f docker-compose.yml up -d curl --location 'http://localhost:5000'- Wait a couple of seconds, then check data-prepper log.
docker logs -f data-prepper
Expected behavior
Remove the duplicate key and ingest the log without issue.
Not sure whether this is a right place to fix or not ( forgive me if I'm wrong, I'm a .net developer - FYI)
Code
protected List<OpenTelemetryLog> processLogsList(final List<LogRecord> logsList,
final String serviceName,
final Map<String, Object> ils,
final Map<String, Object> resourceAttributes,
final String schemaUrl) {
return logsList.stream()
.map(log -> JacksonOtelLog.builder()
.withTime(OTelProtoCodec.convertUnixNanosToISO8601(log.getTimeUnixNano()))
.withObservedTime(OTelProtoCodec.convertUnixNanosToISO8601(log.getObservedTimeUnixNano()))
.withServiceName(serviceName)
.withAttributes(OTelProtoCodec.mergeAllAttributes(
Arrays.asList(
// would be possible to de-duplicate log.getAttributesList() first ???, so that will solve the problem.
OTelProtoCodec.unpackKeyValueListLog(log.getAttributesList()),
resourceAttributes,
ils
)
))
.withSchemaUrl(schemaUrl)
.withFlags(log.getFlags())
.withTraceId(OTelProtoCodec.convertByteStringToString(log.getTraceId()))
.withSpanId(OTelProtoCodec.convertByteStringToString(log.getSpanId()))
.withSeverityNumber(log.getSeverityNumberValue())
.withSeverityText(log.getSeverityText())
.withDroppedAttributesCount(log.getDroppedAttributesCount())
.withBody(OTelProtoCodec.convertAnyValue(log.getBody()))
.build())
.collect(Collectors.toList());
}
Screenshots
If applicable, add screenshots to help explain your problem.
Environment (please complete the following information):
- OS: Debian
- Version: 12
Additional context
N/A
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start in data-prepper-plugins/otel-proto-common/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OTelProtoCodec.java, especially unpackKeyValueListLog and processLogsList, then reproduce the failure with the linked docker-compose steps and curl request. Confirm that repeated log attributes such as ConnectionId no longer cause ingestion to fail and that the sample log is accepted.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- observability-sre
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100