From f96e32816d5b40a45fd9a70a9d4ced18ac755d59 Mon Sep 17 00:00:00 2001 From: Wu Sheng Date: Fri, 13 Nov 2026 17:57:48 +0800 Subject: [PATCH] Keep the trace and span an OTLP log was written in, and query logs by an OTLP or Zipkin span The OTLP log receiver dropped a record's own trace_id and span_id, so an OTLP log could not be found from its trace. It now stores the trace ID as the log's trace ID and the span ID in a new otlp_span_id column: the numeric span_id column is a SkyWalking span's number within its segment and cannot hold a 16-hex OTLP or Zipkin span ID. TraceScopeCondition.otlpSpanId (apache/skywalking-query-protocol#181) reads the logs one OTLP or Zipkin span wrote, filtered in the BanyanDB, Elasticsearch and JDBC DAOs and the debugging query API. A related trace is traceId alone, traceId with segmentId and spanId, or traceId with otlpSpanId; any other mix is refused, including a segment or a span number alone, which were accepted before. LAL sets it with the otlpSpanId output field of the default output type and reads it as log.traceContext.otlpSpanId. OTLP/JSON log IDs are converted from hex before parsing, and with otlpTraceStorage: zipkin a trace ID whose upper 64 bits are zero is written in Zipkin's 16-character form. An existing BanyanDB log stream gains the tag at boot (allowBootReshape); JDBC log tables need the column added by hand, recorded as a breaking change. --- docs/en/changes/changes.md | 3 + docs/en/concepts-and-designs/lal.md | 25 ++- docs/en/debugging/query-tracing.md | 5 +- docs/en/setup/backend/log-otlp.md | 9 + .../v2/compiler/LALCodegenHelper.java | 1 + .../v2/compiler/LALScriptExecutionTest.java | 13 ++ .../feature-cases/execution-basic.data.yaml | 24 +++ .../feature-cases/execution-basic.yaml | 30 +++ .../admin/status/DebuggingHTTPHandler.java | 5 + .../manual/log/AbstractLogRecord.java | 13 ++ .../core/analysis/manual/log/LogRecord.java | 2 +- .../manual/log/LogRecordDispatcher.java | 1 + .../server/core/query/LogQueryService.java | 27 +++ .../core/query/input/TraceScopeCondition.java | 1 + .../oap/server/core/source/AbstractLog.java | 1 + .../oap/server/core/source/Log.java | 1 + .../oap/server/core/source/LogBuilder.java | 11 ++ .../oap/server/core/source/LogMetadata.java | 8 +- .../core/query/LogQueryServiceTest.java | 108 +++++++++++ .../src/main/resources/query-protocol | 2 +- .../otlp/OpenTelemetryLogHTTPHandler.java | 3 +- .../otel/otlp/OpenTelemetryLogHandler.java | 48 ++++- ...enTelemetryLogHandlerTraceContextTest.java | 182 ++++++++++++++++++ .../server/receiver/otel/otlp/OTLPValues.java | 4 +- .../banyandb/stream/BanyanDBLogQueryDAO.java | 3 + .../elasticsearch/query/LogQueryEsDAO.java | 3 + .../jdbc/common/dao/JDBCLogQueryDAO.java | 5 + .../jdbc/common/dao/JDBCLogQueryDAOTest.java | 13 ++ .../otlp-traces/native/banyandb/e2e.yaml | 1 + .../cases/otlp-traces/native/es/e2e.yaml | 1 + .../native/expected/logs-by-otlp-span.yml | 26 +++ .../cases/otlp-traces/native/log-cases.yaml | 35 ++++ .../cases/otlp-traces/native/mysql/e2e.yaml | 1 + 33 files changed, 604 insertions(+), 11 deletions(-) create mode 100644 oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/query/LogQueryServiceTest.java create mode 100644 oap-server/server-receiver-plugin/otel-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandlerTraceContextTest.java create mode 100644 test/e2e-v2/cases/otlp-traces/native/expected/logs-by-otlp-span.yml create mode 100644 test/e2e-v2/cases/otlp-traces/native/log-cases.yaml diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index 2b75e8bbc12f..719ab4a0345a 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -20,6 +20,9 @@ * Fix a webhook alarm hook not being posted when the alarm also names a hook of another type: the webhook callback looked up the other hook's name in the webhook settings and threw a `NullPointerException`. Depending on the hook names, the webhook was never posted (for example `webhook.default` with `gRPC.default`), or an error was logged on every alarm. * Add `alarm.default.grpcTimeout` / `SW_ALARM_GRPC_TIMEOUT`, the deadline in seconds of each gRPC alarm hook stream (default `10`, which was the fixed deadline before). * Record gRPC alarm hook deliveries on the alarm delivery self-observability metrics (`alarm_http_delivery_total` and `alarm_http_delivery_latency_seconds`) with `hook_type=gRPC`, so the OAP dashboards show the gRPC hook as one more channel. +* Keep the trace an OTLP log was written in: the OTLP log receiver now stores a record's `trace_id` as the log's trace ID, which it dropped before, and its `span_id` in a new `otlp_span_id` column. The log query's `TraceScopeCondition` takes `otlpSpanId` to read the logs one OTLP or Zipkin span wrote, which the numeric `spanId` cannot name, and a LAL rule sets it with `otlpSpanId`, an output field of the default output type. An existing BanyanDB log stream and Elasticsearch log index gain the column at boot. +* A log query's `relatedTrace` is one of `traceId` alone, `traceId` with `segmentId` and `spanId` for a SkyWalking native span, or `traceId` with `otlpSpanId` for an OTLP or Zipkin span; any other mix is refused rather than read. This also refuses two forms accepted before: `segmentId` without `spanId`, which read the logs of a whole segment, and `spanId` without `segmentId`. The debugging query API refuses `segmentId`, `spanId` or `otlpSpanId` without `traceId`, which it ignored before. +* **Breaking Change** — JDBC storage (MySQL, PostgreSQL): the log tables gain the `otlp_span_id` column, and OAP adds it at boot only to the current day's table. Every other existing log table keeps its old shape — including the next day's, when OAP had already created it before the upgrade — and inserting a log into one fails, for logs from every source. When upgrading, add the column to the existing log tables, `ALTER TABLE log_ ADD COLUMN otlp_span_id VARCHAR(64);`, or drop them, which deletes their logs, so OAP creates them anew. #### UI diff --git a/docs/en/concepts-and-designs/lal.md b/docs/en/concepts-and-designs/lal.md index d233c10b3cd1..65e11e65fa79 100644 --- a/docs/en/concepts-and-designs/lal.md +++ b/docs/en/concepts-and-designs/lal.md @@ -394,7 +394,30 @@ not dropped) and is used to associate with traces / metrics. - `spanId` `spanId` extracts the span ID from the `parsed` result, and set it into the `LogData`, which will be persisted (if not -dropped) and is used to associate with traces / metrics. +dropped) and is used to associate with traces / metrics. It is a SkyWalking span ID: a number, counted within the segment +`segmentId` names. + +- `otlpSpanId` + +`otlpSpanId` sets the span ID of an OTLP or Zipkin span, 16 lowercase hex characters, which `spanId` cannot hold. It is +persisted with the log, so the log can be queried by the span through `relatedTrace { traceId, otlpSpanId }`. It is an +[output field](#output-fields) of the default output type rather than a keyword, so it works in rules without a custom +`outputType`: + +``` +filter { + text { + regexp $/trace_id=(?[0-9a-f]{32}) span_id=(?[0-9a-f]{16})/$ + } + extractor { + traceId parsed.traceId as String + otlpSpanId parsed.spanId as String + } +} +``` + +The [OTLP log receiver](../setup/backend/log-otlp.md) sets it from the record's own `span_id`; a rule reads it as +`log.traceContext.otlpSpanId`. - `timestamp` diff --git a/docs/en/debugging/query-tracing.md b/docs/en/debugging/query-tracing.md index af2f96f3dcc3..8149ea23c24c 100644 --- a/docs/en/debugging/query-tracing.md +++ b/docs/en/debugging/query-tracing.md @@ -474,8 +474,9 @@ debuggingTrace: | serviceInstance | The service instance name | No, require service | | endpoint | The endpoint name | No, require service | | traceId | The trace ID | No | - | segmentId | The segment ID | No, require traceId | - | spanId | The span ID | No, require traceId | + | segmentId | The segment ID | No, require traceId and spanId | + | spanId | The span ID | No, require traceId and segmentId | + | otlpSpanId | The OTLP or Zipkin span ID, 16 lowercase hex characters | No, require traceId; not with segmentId or spanId | | queryOrder | The order of the query result, `ASC`, `DES` | No, default `DES` | | tags | The tags of the trace, `key1=value1,key2=value2` | No | | pageNum | The page number of the query result | Yes | diff --git a/docs/en/setup/backend/log-otlp.md b/docs/en/setup/backend/log-otlp.md index 5390adac41c5..5b3c68e87170 100644 --- a/docs/en/setup/backend/log-otlp.md +++ b/docs/en/setup/backend/log-otlp.md @@ -55,3 +55,12 @@ And several attributes are optional as add-on information for the logs before an Note, that these attributes should be set manually through OpenTelemetry SDK or through [attribute#insert in OpenTelemetry Collector](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/attributesprocessor/README.md). The OTLP log record body is delivered to [LAL](../../concepts-and-designs/lal.md) as a plain-text body, even when the string content is JSON-shaped. A LAL rule can still parse such content with the `json {}` parser — when the JSON body is empty and a text body is present, `json {}` tries the text as JSON. On a successful parse the log is normalized to a JSON body, so it is stored and rendered as JSON. See [`json`](../../concepts-and-designs/lal.md#json) for details. + +A log record written inside a span carries that span's `trace_id` and `span_id`. Both are stored with the log, as +lowercase hex: the trace ID as the log's trace ID, and the span ID as its OTLP span ID — not the `spanId`, which is a +SkyWalking span's number within its segment. So the log can be queried by the trace that wrote it, through +`relatedTrace { traceId }`, and by the span, through `relatedTrace { traceId, otlpSpanId }`. A LAL rule reads the span ID as +`log.traceContext.otlpSpanId`, and may set it with [`otlpSpanId`](../../concepts-and-designs/lal.md#extractor). + +The trace ID is written the way this OAP stores the OTLP trace itself: with `otlpTraceStorage: zipkin`, an ID whose upper +64 bits are zero — a trace started by a tracer with 64-bit IDs — keeps only its lower 16 hex characters, as Zipkin does. diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALCodegenHelper.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALCodegenHelper.java index 09a889e9af10..115b87688ad5 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALCodegenHelper.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALCodegenHelper.java @@ -53,6 +53,7 @@ final class LALCodegenHelper { METADATA_TRACE_GETTERS.put("traceId", "getTraceId"); METADATA_TRACE_GETTERS.put("traceSegmentId", "getTraceSegmentId"); METADATA_TRACE_GETTERS.put("spanId", "getSpanId"); + METADATA_TRACE_GETTERS.put("otlpSpanId", "getOtlpSpanId"); LOG_GETTERS.put("body", "getBody"); LOG_GETTERS.put("tags", "getTags"); diff --git a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java index 65a8c0ba6d70..72449d44b720 100644 --- a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java +++ b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALScriptExecutionTest.java @@ -38,6 +38,7 @@ import org.apache.skywalking.oap.server.core.config.ConfigService; import org.apache.skywalking.oap.server.core.config.NamingControl; import org.apache.skywalking.oap.server.core.source.LALOutputBuilder; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.core.source.SourceReceiver; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.module.ModuleProviderHolder; @@ -188,6 +189,18 @@ private void executeAndValidate( srcAttrs != null ? org.apache.skywalking.oap.server.core.source.LogMetadataUtils.fromLogData(testLogBuilder, srcAttrs) : org.apache.skywalking.oap.server.core.source.LogMetadataUtils.fromLogData(testLog); + final Object otlpSpanId = inputData != null ? inputData.get("otlp-span-id") : null; + if (otlpSpanId != null) { + // No native protocol carries an OTLP span id; the OTLP receiver puts it on the metadata. + final LogMetadata.TraceContext tc = v2Metadata.getTraceContext(); + v2Metadata.setTraceContext( + LogMetadata.TraceContext.builder() + .traceId(tc.getTraceId()) + .traceSegmentId(tc.getTraceSegmentId()) + .spanId(tc.getSpanId()) + .otlpSpanId(String.valueOf(otlpSpanId)) + .build()); + } final Object v2Input = extraLog != null ? extraLog : testLogBuilder; v2Ctx = new org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext(); v2Ctx.init(v2Metadata, v2Input); diff --git a/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml b/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml index c338c1f6fa37..1c65caf260b0 100644 --- a/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml +++ b/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.data.yaml @@ -270,6 +270,30 @@ trace-context-read: save: true abort: false +trace-otlp-span-read: + body-type: json + body: '{"service":"trace-svc"}' + trace-id: 0a1b2c3d4e5f60718293a4b5c6d7e8f9 + otlp-span-id: 0123456789abcdef + expect: + service: trace-svc + tag.osid: 0123456789abcdef + traceId: 0a1b2c3d4e5f60718293a4b5c6d7e8f9 + outputField.otlpSpanId: 0123456789abcdef + save: true + abort: false + +trace-otlp-span-write: + body-type: json + body: '{"service":"trace-svc","tid":"4bf92f3577b34da6a3ce929d0e0e4736","osp":"00f067aa0ba902b7"}' + otlp-span-id: 0123456789abcdef + expect: + service: trace-svc + traceId: 4bf92f3577b34da6a3ce929d0e0e4736 + outputField.otlpSpanId: 00f067aa0ba902b7 + save: true + abort: false + trace-segment-override: body-type: json body: '{"service":"trace-svc","seg":"segment-from-body"}' diff --git a/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml b/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml index b0a953dbd62c..cdef3bd0dd94 100644 --- a/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml +++ b/oap-server/analyzer/log-analyzer/src/test/resources/scripts/lal/test-lal/feature-cases/execution-basic.yaml @@ -417,6 +417,36 @@ rules: sink {} } + # An OTLP or Zipkin span id has no keyword: it is an output field of the default output + # type, LogBuilder (setOtlpSpanId). Read it from the metadata's TraceContext, where the OTLP + # receiver puts it, and let LogBuilder copy it into the output untouched. + - name: trace-otlp-span-read + layer: GENERAL + dsl: | + filter { + json {} + extractor { + service parsed.service as String + tag 'osid': log.traceContext.otlpSpanId + } + sink {} + } + + # The write direction: a span id written into the log text — a Zipkin or OpenTelemetry + # logging context — wins over the one the metadata carries. + - name: trace-otlp-span-write + layer: GENERAL + dsl: | + filter { + json {} + extractor { + service parsed.service as String + traceId parsed.tid as String + otlpSpanId parsed.osp as String + } + sink {} + } + # segmentId alone — the single-statement shape that used to NPE. - name: trace-segment-override layer: GENERAL diff --git a/oap-server/server-admin/status/src/main/java/org/apache/skywalking/oap/server/admin/status/DebuggingHTTPHandler.java b/oap-server/server-admin/status/src/main/java/org/apache/skywalking/oap/server/admin/status/DebuggingHTTPHandler.java index 4daf851a0dba..eb264338a366 100644 --- a/oap-server/server-admin/status/src/main/java/org/apache/skywalking/oap/server/admin/status/DebuggingHTTPHandler.java +++ b/oap-server/server-admin/status/src/main/java/org/apache/skywalking/oap/server/admin/status/DebuggingHTTPHandler.java @@ -551,6 +551,7 @@ public String queryLogs(@Param("service") Optional service, @Param("traceId") Optional traceId, @Param("segmentId") Optional segmentId, @Param("spanId") Optional spanId, + @Param("otlpSpanId") Optional otlpSpanId, @Param("tags") Optional tags, @Param("pageNum") int pageNum, @Param("pageSize") int pageSize, @@ -559,11 +560,15 @@ public String queryLogs(@Param("service") Optional service, @Param("queryOrder") Optional queryOrder) { LogQueryCondition condition = new LogQueryCondition(); condition.setPaging(new Pagination(pageNum, pageSize)); + if (traceId.isEmpty() && (segmentId.isPresent() || spanId.isPresent() || otlpSpanId.isPresent())) { + return "traceId is required with segmentId, spanId or otlpSpanId"; + } if (traceId.isPresent()) { TraceScopeCondition traceScopeCondition = new TraceScopeCondition(); traceScopeCondition.setTraceId(traceId.get()); segmentId.ifPresent(traceScopeCondition::setSegmentId); spanId.ifPresent(traceScopeCondition::setSpanId); + otlpSpanId.ifPresent(traceScopeCondition::setOtlpSpanId); condition.setRelatedTrace(traceScopeCondition); } else { if (startTime.isEmpty() || endTime.isEmpty() || step.isEmpty()) { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java index 15726e3d7ce3..3b2f4a872eaa 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java @@ -44,6 +44,7 @@ public abstract class AbstractLogRecord extends Record { public static final String TRACE_ID = "trace_id"; public static final String TRACE_SEGMENT_ID = "trace_segment_id"; public static final String SPAN_ID = "span_id"; + public static final String OTLP_SPAN_ID = "otlp_span_id"; public static final String CONTENT_TYPE = "content_type"; public static final String CONTENT = "content"; public static final String TAGS_RAW_DATA = "tags_raw_data"; @@ -79,6 +80,16 @@ public abstract class AbstractLogRecord extends Record { @Column(name = SPAN_ID) @BanyanDB.NoIndexing private int spanId; + /** + * The span that wrote the log when it is an OTLP or Zipkin span: its 16 lowercase hex characters. + * A SkyWalking native span is {@link #traceSegmentId} + {@link #spanId} instead, as a native span id + * is a number counted within its segment. + */ + @Setter + @Getter + @Column(name = OTLP_SPAN_ID, length = 64) + @BanyanDB.NoIndexing + private String otlpSpanId; @Setter @Getter @Column(name = CONTENT_TYPE, storageOnly = true) @@ -120,6 +131,7 @@ protected void map2Data(T record, final Convert2Entity converter) { record.setTraceId((String) converter.get(TRACE_ID)); record.setTraceSegmentId((String) converter.get(TRACE_SEGMENT_ID)); record.setSpanId(((Number) converter.get(SPAN_ID)).intValue()); + record.setOtlpSpanId((String) converter.get(OTLP_SPAN_ID)); record.setContentType(((Number) converter.get(CONTENT_TYPE)).intValue()); record.setContent(new LongText((String) converter.get(CONTENT))); record.setTimestamp(((Number) converter.get(TIMESTAMP)).longValue()); @@ -134,6 +146,7 @@ protected void data2Map(final T record, final Convert2Storage converter) { converter.accept(TRACE_ID, record.getTraceId()); converter.accept(TRACE_SEGMENT_ID, record.getTraceSegmentId()); converter.accept(SPAN_ID, record.getSpanId()); + converter.accept(OTLP_SPAN_ID, record.getOtlpSpanId()); converter.accept(TIME_BUCKET, record.getTimeBucket()); converter.accept(CONTENT_TYPE, record.getContentType()); converter.accept(CONTENT, record.getContent()); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/LogRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/LogRecord.java index 890ab5fc7bf8..5c1f5d1c8b67 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/LogRecord.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/LogRecord.java @@ -33,7 +33,7 @@ import static org.apache.skywalking.oap.server.core.storage.StorageData.TIME_BUCKET; @SuperDataset -@Stream(name = LogRecord.INDEX_NAME, scopeId = DefaultScopeDefine.LOG, builder = LogRecord.Builder.class, processor = RecordStreamProcessor.class) +@Stream(name = LogRecord.INDEX_NAME, scopeId = DefaultScopeDefine.LOG, builder = LogRecord.Builder.class, processor = RecordStreamProcessor.class, allowBootReshape = true) @SQLDatabase.ExtraColumn4AdditionalEntity(additionalTable = AbstractLogRecord.ADDITIONAL_TAG_TABLE, parentColumn = TIME_BUCKET) @BanyanDB.TimestampColumn(AbstractLogRecord.TIMESTAMP) @BanyanDB.Group(streamGroup = BanyanDB.StreamGroup.RECORDS_LOG) diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/LogRecordDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/LogRecordDispatcher.java index 95736d66f905..a13f0adffa43 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/LogRecordDispatcher.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/LogRecordDispatcher.java @@ -37,6 +37,7 @@ public void dispatch(final Log source) { record.setTraceId(source.getTraceId()); record.setTraceSegmentId(source.getTraceSegmentId()); record.setSpanId(source.getSpanId()); + record.setOtlpSpanId(source.getOtlpSpanId()); record.setContentType(source.getContentType().value()); record.setContent(new LongText(source.getContent())); record.setTagsRawData(source.getTagsRawData()); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/LogQueryService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/LogQueryService.java index 6ed55d455452..0e06c90b7234 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/LogQueryService.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/LogQueryService.java @@ -69,6 +69,7 @@ public Logs queryLogs(String serviceId, final List tags, List keywordsOfContent, List excludingKeywordsOfContent) throws IOException { + validateRelatedTrace(relatedTrace); DebuggingTraceContext traceContext = TRACE_CONTEXT.get(); DebuggingSpan span = null; try { @@ -98,6 +99,32 @@ public Logs queryLogs(String serviceId, } } + /** + * A related trace is one of three forms: {@code traceId} alone for the whole trace, {@code traceId} + + * {@code segmentId} + {@code spanId} for a SkyWalking native span, or {@code traceId} + {@code otlpSpanId} for an + * OTLP or Zipkin span. Any other mix names a partial or contradictory span, which would answer "no logs", or a + * different set of logs, rather than report the mistake. A condition with nothing set, as the LogQL handler passes + * without a trace ID label, is no condition. + */ + static void validateRelatedTrace(final TraceScopeCondition relatedTrace) { + if (relatedTrace == null) { + return; + } + final boolean segment = StringUtil.isNotEmpty(relatedTrace.getSegmentId()); + final boolean span = nonNull(relatedTrace.getSpanId()); + final boolean otlpSpan = StringUtil.isNotEmpty(relatedTrace.getOtlpSpanId()); + if (!segment && !span && !otlpSpan) { + return; + } + final boolean nativeSpan = segment && span && !otlpSpan; + final boolean otlpSpanOnly = otlpSpan && !segment && !span; + if (StringUtil.isEmpty(relatedTrace.getTraceId()) || !(nativeSpan || otlpSpanOnly)) { + throw new IllegalArgumentException( + "relatedTrace takes traceId alone, traceId with segmentId and spanId for a SkyWalking native span, " + + "or traceId with otlpSpanId for an OTLP or Zipkin span"); + } + } + private Logs invokeQueryLogs(String serviceId, String serviceInstanceId, String endpointId, diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/input/TraceScopeCondition.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/input/TraceScopeCondition.java index b5b2aecc3080..431893399ef5 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/input/TraceScopeCondition.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/input/TraceScopeCondition.java @@ -29,4 +29,5 @@ public class TraceScopeCondition { private String traceId; private String segmentId; private Integer spanId; + private String otlpSpanId; } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/AbstractLog.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/AbstractLog.java index b5b902d29282..97690ead86a9 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/AbstractLog.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/AbstractLog.java @@ -38,6 +38,7 @@ public abstract class AbstractLog extends Source { private String traceId; private String traceSegmentId; private int spanId; + private String otlpSpanId; private ContentType contentType = ContentType.NONE; private String content; private byte[] tagsRawData; diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Log.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Log.java index 1ec78255909f..72819d5ac04e 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Log.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Log.java @@ -43,6 +43,7 @@ public String toJson() { obj.addProperty("traceId", getTraceId()); obj.addProperty("traceSegmentId", getTraceSegmentId()); obj.addProperty("spanId", getSpanId()); + obj.addProperty("otlpSpanId", getOtlpSpanId()); obj.addProperty("contentType", contentType == null ? null : contentType.name()); obj.addProperty("content", getContent()); obj.addProperty("error", isError()); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogBuilder.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogBuilder.java index 78cd39b2531c..38cf7f455d1b 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogBuilder.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogBuilder.java @@ -73,6 +73,8 @@ public class LogBuilder implements LALOutputBuilder { private String traceId; @Setter private String segmentId; + @Setter + private String otlpSpanId; @Getter @Setter private long timestamp; @@ -159,6 +161,9 @@ protected void initFromMetadata(final LogMetadata metadata) { if (this.spanId < 0) { this.spanId = tc.getSpanId(); } + if (this.otlpSpanId == null) { + this.otlpSpanId = tc.getOtlpSpanId(); + } } if (this.timestamp == 0) { this.timestamp = metadata.getTimestamp(); @@ -206,6 +211,9 @@ public Log toLog() { log.setSpanId(spanId); } } + if (StringUtil.isNotEmpty(otlpSpanId)) { + log.setOtlpSpanId(otlpSpanId); + } // content (only when input is LogData) if (logData != null) { final LogDataBody body = logData.getBody(); @@ -309,6 +317,9 @@ public String outputToJson() { if (spanId >= 0) { obj.addProperty("spanId", spanId); } + if (StringUtil.isNotEmpty(otlpSpanId)) { + obj.addProperty("otlpSpanId", otlpSpanId); + } obj.addProperty("timestamp", timestamp); appendBodyContent(obj); appendMergedTags(obj); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadata.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadata.java index df2f5acfd8ee..a6aab1155a25 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadata.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadata.java @@ -51,12 +51,18 @@ public class LogMetadata { @Builder public static class TraceContext { static final TraceContext EMPTY = - TraceContext.builder().traceId("").traceSegmentId("").spanId(0).build(); + TraceContext.builder().traceId("").traceSegmentId("").spanId(0).otlpSpanId("").build(); @Builder.Default private String traceId = ""; @Builder.Default private String traceSegmentId = ""; private int spanId; + /** + * An OTLP or Zipkin span id, hex. No native protocol carries one; a receiver of those + * protocols sets it. + */ + @Builder.Default + private String otlpSpanId = ""; } } diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/query/LogQueryServiceTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/query/LogQueryServiceTest.java new file mode 100644 index 000000000000..9923be87f295 --- /dev/null +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/query/LogQueryServiceTest.java @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.core.query; + +import java.lang.reflect.Field; +import org.apache.skywalking.oap.server.core.query.enumeration.Order; +import org.apache.skywalking.oap.server.core.query.input.TraceScopeCondition; +import org.apache.skywalking.oap.server.core.query.type.Logs; +import org.apache.skywalking.oap.server.core.query.type.Pagination; +import org.apache.skywalking.oap.server.core.storage.query.ILogQueryDAO; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * A related trace is the whole trace, a SkyWalking native span by segment ID and span ID, or an OTLP or Zipkin span by + * its hex ID, each with the trace ID. + */ +class LogQueryServiceTest { + private static final String TRACE_ID = "0af7651916cd43dd8448eb211c80319c"; + private static final String OTLP_SPAN = "b7ad6b7169203331"; + + @Test + void theAcceptedFormsAreRead() throws Exception { + for (final TraceScopeCondition accepted : new TraceScopeCondition[] { + condition(TRACE_ID, null, null, null), + condition(TRACE_ID, "segment-1", 0, null), + condition(TRACE_ID, null, null, OTLP_SPAN), + // The LogQL handler passes an empty condition when the query names no trace ID. + condition(null, null, null, null), + condition("", "", null, "") + }) { + final ILogQueryDAO dao = dao(); + + query(dao, accepted); + + verify(dao).queryLogsDebuggable(any(), any(), any(), eq(accepted), any(), anyInt(), anyInt(), any(), any(), + any(), any()); + } + } + + @Test + void anyOtherMixIsRefusedBeforeStorageIsAsked() throws Exception { + for (final TraceScopeCondition refused : new TraceScopeCondition[] { + condition(TRACE_ID, "segment-1", 0, OTLP_SPAN), + condition(TRACE_ID, "segment-1", null, OTLP_SPAN), + condition(TRACE_ID, null, 0, OTLP_SPAN), + condition(TRACE_ID, "segment-1", null, null), + condition(TRACE_ID, null, 0, null), + condition(null, "segment-1", 0, null), + condition("", null, null, OTLP_SPAN), + condition(null, null, 0, null) + }) { + final ILogQueryDAO dao = dao(); + assertThrows(IllegalArgumentException.class, () -> query(dao, refused), refused.toString()); + verify(dao, never()).queryLogsDebuggable(any(), any(), any(), any(), any(), anyInt(), anyInt(), any(), + any(), any(), any()); + } + } + + private static TraceScopeCondition condition(final String traceId, final String segmentId, final Integer spanId, + final String otlpSpanId) { + final TraceScopeCondition condition = new TraceScopeCondition(); + condition.setTraceId(traceId); + condition.setSegmentId(segmentId); + condition.setSpanId(spanId); + condition.setOtlpSpanId(otlpSpanId); + return condition; + } + + private static ILogQueryDAO dao() throws Exception { + final ILogQueryDAO dao = mock(ILogQueryDAO.class); + when(dao.queryLogsDebuggable(any(), any(), any(), any(), any(), anyInt(), anyInt(), any(), any(), any(), + any())).thenReturn(new Logs()); + return dao; + } + + private static void query(final ILogQueryDAO dao, final TraceScopeCondition condition) throws Exception { + final LogQueryService service = new LogQueryService(null); + final Field field = LogQueryService.class.getDeclaredField("logQueryDAO"); + field.setAccessible(true); + field.set(service, dao); + service.queryLogs(null, null, null, condition, new Pagination(1, 10), Order.DES, null, null, null, null); + } +} diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol index 35155de7564a..966fb2cec582 160000 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol @@ -1 +1 @@ -Subproject commit 35155de7564af962144e662020700cedcdf2ab9d +Subproject commit 966fb2cec582846a4edb1a2f44af440a4803d479 diff --git a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHTTPHandler.java b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHTTPHandler.java index 8d3122bb7b00..b34b6be204f6 100644 --- a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHTTPHandler.java +++ b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHTTPHandler.java @@ -69,8 +69,9 @@ public HttpResponse collectJson(AggregatedHttpRequest request) { try { final ExportLogsServiceRequest.Builder builder = ExportLogsServiceRequest.newBuilder(); + // OTLP/JSON writes a record's traceId and spanId as hex, which the protobuf parser would read as base64. JsonFormat.parser().ignoringUnknownFields().merge( - request.contentUtf8(), builder); + OTLPValues.hexIdsToBase64(request.contentUtf8()), builder); logHandler.processExport(builder.build()); return HttpResponse.of(HttpStatus.OK, MediaType.JSON_UTF_8, "{}"); } catch (InvalidProtocolBufferException e) { diff --git a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java index 3bde2ae13bed..c934b12366f2 100644 --- a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java +++ b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java @@ -24,6 +24,7 @@ import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceResponse; import io.opentelemetry.proto.collector.logs.v1.LogsServiceGrpc; import io.opentelemetry.proto.common.v1.KeyValue; +import com.google.protobuf.ByteString; import io.opentelemetry.proto.logs.v1.LogRecord; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -33,6 +34,8 @@ import org.apache.skywalking.apm.network.logging.v3.LogDataBody; import org.apache.skywalking.apm.network.logging.v3.LogTags; import org.apache.skywalking.apm.network.logging.v3.TextLog; +import org.apache.skywalking.apm.network.logging.v3.TraceContext; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.core.source.LogMetadataUtils; import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService; @@ -70,6 +73,12 @@ public class OpenTelemetryLogHandler */ private List serviceGroupKeys = Collections.emptyList(); + /** + * Whether this OAP stores OTLP spans as Zipkin spans ({@code otlpTraceStorage: zipkin}), which shortens a + * trace ID whose upper 64 bits are zero; a log's trace ID has to be written the way its trace is. + */ + private boolean zipkinTraceIds; + @Getter(lazy = true) private final MetricsCreator metricsCreator = manager.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class); @@ -85,6 +94,7 @@ public class OpenTelemetryLogHandler public void init(ModuleManager manager, OtelMetricReceiverConfig config) { this.manager = manager; this.serviceGroupKeys = config.getOtlpServiceGroupKeys(); + this.zipkinTraceIds = !config.isOtlpTraceStorageNative(); } @Override @@ -162,13 +172,47 @@ private void doAnalysisQuietly(String service, String layer, String serviceInsta .setTags(buildTags(logRecord)) .setBody(buildBody(logRecord)) .setLayer(layer); - logAnalyzerService().doAnalysis( - LogMetadataUtils.fromLogData(builder, resourceAttributes), builder); + final String traceId = traceIdOf(logRecord.getTraceId()); + if (!traceId.isEmpty()) { + builder.setTraceContext(TraceContext.newBuilder().setTraceId(traceId)); + } + final LogMetadata metadata = LogMetadataUtils.fromLogData(builder, resourceAttributes); + final String spanId = idOf(logRecord.getSpanId()); + if (!spanId.isEmpty()) { + // A new context: the one a log without a trace id gets is a shared constant. + metadata.setTraceContext(LogMetadata.TraceContext.builder() + .traceId(traceId) + .otlpSpanId(spanId) + .build()); + } + logAnalyzerService().doAnalysis(metadata, builder); } catch (Exception e) { log.error("Failed to analyze logs", e); } } + /** + * The hex form of an OTLP trace or span id, or empty when the record carries none — the + * protocol marks an absent id as empty or all-zero bytes. + */ + static String idOf(final ByteString id) { + for (int i = 0; i < id.size(); i++) { + if (id.byteAt(i) != 0) { + return OTLPValues.hexId(id); + } + } + return ""; + } + + /** + * The trace ID as the store of this OAP's OTLP traces writes it: Zipkin keeps only the lower 64 bits of an ID + * whose upper 64 are zero. + */ + private String traceIdOf(final ByteString id) { + final String hex = idOf(id); + return zipkinTraceIds && hex.length() == 32 && hex.startsWith("0000000000000000") ? hex.substring(16) : hex; + } + private static LogDataBody buildBody(LogRecord logRecord) { return LogDataBody.newBuilder().setText( TextLog.newBuilder().setText( diff --git a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandlerTraceContextTest.java b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandlerTraceContextTest.java new file mode 100644 index 000000000000..fa58331de363 --- /dev/null +++ b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandlerTraceContextTest.java @@ -0,0 +1,182 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.receiver.otel.otlp; + +import com.google.protobuf.ByteString; +import com.linecorp.armeria.common.AggregatedHttpRequest; +import com.linecorp.armeria.common.HttpMethod; +import com.linecorp.armeria.common.MediaType; +import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest; +import io.opentelemetry.proto.common.v1.AnyValue; +import io.opentelemetry.proto.common.v1.KeyValue; +import io.opentelemetry.proto.logs.v1.LogRecord; +import io.opentelemetry.proto.logs.v1.ResourceLogs; +import io.opentelemetry.proto.logs.v1.ScopeLogs; +import io.opentelemetry.proto.resource.v1.Resource; +import java.util.ArrayList; +import java.util.List; +import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; +import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService; +import org.apache.skywalking.oap.server.core.source.LogMetadata; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.library.module.ModuleProviderHolder; +import org.apache.skywalking.oap.server.library.module.ModuleServiceHolder; +import org.apache.skywalking.oap.server.library.module.Service; +import org.apache.skywalking.oap.server.receiver.otel.OtelMetricReceiverConfig; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics; +import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * An OTLP log record's own trace id and span id reach the log analysis, so the stored log is found + * from its trace and from its span. + */ +class OpenTelemetryLogHandlerTraceContextTest { + private static final byte[] TRACE_ID = { + 0x0a, 0x1b, 0x2c, 0x3d, 0x4e, 0x5f, 0x60, 0x71, (byte) 0x82, (byte) 0x93, (byte) 0xa4, (byte) 0xb5, + (byte) 0xc6, (byte) 0xd7, (byte) 0xe8, (byte) 0xf9 + }; + private static final byte[] SPAN_ID = {0x01, 0x23, 0x45, 0x67, (byte) 0x89, (byte) 0xab, (byte) 0xcd, (byte) 0xef}; + + private final List metadata = new ArrayList<>(); + private final List inputs = new ArrayList<>(); + + @Test + void aRecordInATraceCarriesItsTraceIdAndSpanId() { + export(LogRecord.newBuilder() + .setTraceId(ByteString.copyFrom(TRACE_ID)) + .setSpanId(ByteString.copyFrom(SPAN_ID))); + + final LogMetadata.TraceContext context = metadata.get(0).getTraceContext(); + assertEquals("0a1b2c3d4e5f60718293a4b5c6d7e8f9", context.getTraceId()); + assertEquals("0123456789abcdef", context.getOtlpSpanId()); + assertEquals("", context.getTraceSegmentId(), "an OTLP span is not a native segment"); + assertEquals("0a1b2c3d4e5f60718293a4b5c6d7e8f9", inputs.get(0).getTraceContext().getTraceId(), + "LAL reads the trace id off the LogData too"); + } + + @Test + void aRecordOutsideATraceCarriesNeither() { + export(LogRecord.newBuilder()); + export(LogRecord.newBuilder() + .setTraceId(ByteString.copyFrom(new byte[16])) + .setSpanId(ByteString.copyFrom(new byte[8]))); + + for (final LogMetadata m : metadata) { + assertEquals("", m.getTraceContext().getTraceId(), "an absent or all-zero trace id is no trace id"); + assertEquals("", m.getTraceContext().getOtlpSpanId(), "an absent or all-zero span id is no span id"); + } + } + + @Test + void aSpanIdWithoutATraceIdLeavesTheSharedEmptyContextAlone() { + export(LogRecord.newBuilder().setSpanId(ByteString.copyFrom(SPAN_ID))); + export(LogRecord.newBuilder()); + + assertEquals("0123456789abcdef", metadata.get(0).getTraceContext().getOtlpSpanId()); + assertEquals("", metadata.get(1).getTraceContext().getOtlpSpanId(), + "the next record without ids must not inherit the previous one's span id"); + } + + @Test + void anOtlpJsonRecordKeepsTheIdsItsSenderWrote() { + final OpenTelemetryLogHandler handler = new OpenTelemetryLogHandler(); + handler.init(manager(), new OtelMetricReceiverConfig()); + new OpenTelemetryLogHTTPHandler(handler).collectJson(AggregatedHttpRequest.of( + HttpMethod.POST, "/v1/logs", MediaType.JSON_UTF_8, + "{\"resourceLogs\":[{\"resource\":{\"attributes\":[{\"key\":\"service.name\"," + + "\"value\":{\"stringValue\":\"checkout\"}}]},\"scopeLogs\":[{\"logRecords\":[{" + + "\"timeUnixNano\":\"1700000000000000000\",\"body\":{\"stringValue\":\"a line\"}," + + "\"traceId\":\"0a1b2c3d4e5f60718293a4b5c6d7e8f9\",\"spanId\":\"0123456789abcdef\"}]}]}]}")); + + assertEquals("0a1b2c3d4e5f60718293a4b5c6d7e8f9", metadata.get(0).getTraceContext().getTraceId(), + "OTLP/JSON writes ids as hex, not as the base64 protobuf's JSON parser expects"); + assertEquals("0123456789abcdef", metadata.get(0).getTraceContext().getOtlpSpanId()); + } + + @Test + void aTraceIdWithAZeroUpperHalfIsWrittenAsTheTraceStoreWritesIt() { + final byte[] sixtyFourBit = {0, 0, 0, 0, 0, 0, 0, 0, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}; + export(LogRecord.newBuilder().setTraceId(ByteString.copyFrom(sixtyFourBit)), OtelMetricReceiverConfig.OTLP_TRACE_STORAGE_OTLP); + export(LogRecord.newBuilder().setTraceId(ByteString.copyFrom(sixtyFourBit)), OtelMetricReceiverConfig.OTLP_TRACE_STORAGE_ZIPKIN); + + assertEquals("00000000000000000102030405060708", metadata.get(0).getTraceContext().getTraceId(), + "the native OTLP span store keeps all 32 characters"); + assertEquals("0102030405060708", metadata.get(1).getTraceContext().getTraceId(), + "Zipkin keeps the lower 64 bits of an ID whose upper 64 are zero"); + } + + private void export(final LogRecord.Builder record) { + export(record, OtelMetricReceiverConfig.OTLP_TRACE_STORAGE_OTLP); + } + + private void export(final LogRecord.Builder record, final String traceStorage) { + final OpenTelemetryLogHandler handler = new OpenTelemetryLogHandler(); + final OtelMetricReceiverConfig config = new OtelMetricReceiverConfig(); + config.setOtlpTraceStorage(traceStorage); + handler.init(manager(), config); + handler.processExport(ExportLogsServiceRequest.newBuilder().addResourceLogs( + ResourceLogs.newBuilder() + .setResource(Resource.newBuilder().addAttributes( + KeyValue.newBuilder().setKey("service.name") + .setValue(AnyValue.newBuilder().setStringValue("checkout")))) + .addScopeLogs(ScopeLogs.newBuilder().addLogRecords( + record.setTimeUnixNano(1_700_000_000_000_000_000L) + .setBody(AnyValue.newBuilder().setStringValue("a line"))))) + .build()); + } + + private ModuleManager manager() { + final ILogAnalyzerService analyzer = mock(ILogAnalyzerService.class); + doAnswer(inv -> { + metadata.add(inv.getArgument(0)); + inputs.add(inv.getArgument(1)); + return null; + }).when(analyzer).doAnalysis(any(), any()); + final MetricsCreator metricsCreator = mock(MetricsCreator.class); + when(metricsCreator.createHistogramMetric(anyString(), anyString(), any(), any())) + .thenReturn(new HistogramMetrics() { + @Override + public void observe(final double value) { + } + }); + final ModuleManager manager = mock(ModuleManager.class); + provide(manager, LogAnalyzerModule.NAME, ILogAnalyzerService.class, analyzer); + provide(manager, TelemetryModule.NAME, MetricsCreator.class, metricsCreator); + return manager; + } + + private static void provide(final ModuleManager manager, final String module, + final Class type, final T service) { + final ModuleServiceHolder services = mock(ModuleServiceHolder.class); + when(services.getService(type)).thenReturn(service); + final ModuleProviderHolder provider = mock(ModuleProviderHolder.class); + when(provider.provider()).thenReturn(services); + when(manager.find(module)).thenReturn(provider); + } +} diff --git a/oap-server/server-receiver-plugin/receiver-proto/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OTLPValues.java b/oap-server/server-receiver-plugin/receiver-proto/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OTLPValues.java index cf45683bf077..ef16064265d4 100644 --- a/oap-server/server-receiver-plugin/receiver-proto/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OTLPValues.java +++ b/oap-server/server-receiver-plugin/receiver-proto/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OTLPValues.java @@ -102,8 +102,8 @@ public static Map toStringMap(final List attributes) { "spanId", "parentSpanId", "span_id", "parent_span_id"); /** - * Rewrite the ids of an OTLP/JSON trace body from hex to base64, so protobuf's JSON parser reads them. OTLP/JSON - * encodes {@code traceId}, {@code spanId} and {@code parentSpanId} as hex, unlike the standard protobuf JSON + * Rewrite the ids of an OTLP/JSON trace or log body from hex to base64, so protobuf's JSON parser reads them. + * OTLP/JSON encodes {@code traceId}, {@code spanId} and {@code parentSpanId} as hex, unlike the standard protobuf JSON * mapping, which reads every {@code bytes} field as base64. Hex digits are base64 characters too, so the * unconverted body parses without an error, into ids of the wrong length that match nothing the SDK logged and * nothing a protobuf sender reports for the same trace. diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java index 559c860f6f07..3ecb626975e9 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java @@ -97,6 +97,9 @@ public Logs queryLogs(String serviceId, String serviceInstanceId, String endpoin if (Objects.nonNull(relatedTrace.getSpanId())) { where.eq(AbstractLogRecord.SPAN_ID, (long) relatedTrace.getSpanId()); } + if (StringUtil.isNotEmpty(relatedTrace.getOtlpSpanId())) { + where.eq(AbstractLogRecord.OTLP_SPAN_ID, relatedTrace.getOtlpSpanId()); + } } if (CollectionUtils.isNotEmpty(tags)) { List tagsConditions = new ArrayList<>(tags.size()); diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java index 2fe233e23516..8acf4b56d3a9 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java @@ -106,6 +106,9 @@ public Logs queryLogs(final String serviceId, if (nonNull(relatedTrace.getSpanId())) { query.must(Query.term(AbstractLogRecord.SPAN_ID, relatedTrace.getSpanId())); } + if (isNotEmpty(relatedTrace.getOtlpSpanId())) { + query.must(Query.term(AbstractLogRecord.OTLP_SPAN_ID, relatedTrace.getOtlpSpanId())); + } } if (CollectionUtils.isNotEmpty(tags)) { diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCLogQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCLogQueryDAO.java index 1a04a461663f..fabbad866f09 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCLogQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCLogQueryDAO.java @@ -67,6 +67,7 @@ import static org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord.TIMESTAMP; import static org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord.TRACE_ID; import static org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord.TRACE_SEGMENT_ID; +import static org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord.OTLP_SPAN_ID; @Slf4j @RequiredArgsConstructor @@ -237,6 +238,10 @@ protected SQLAndParameters buildSQL( sql.append(" and ").append(SPAN_ID).append(" = ?"); parameters.add(relatedTrace.getSpanId()); } + if (StringUtil.isNotEmpty(relatedTrace.getOtlpSpanId())) { + sql.append(" and ").append(OTLP_SPAN_ID).append(" = ?"); + parameters.add(relatedTrace.getOtlpSpanId()); + } } if (CollectionUtils.isNotEmpty(tags)) { diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCLogQueryDAOTest.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCLogQueryDAOTest.java index f65c3574a7af..dbeabe8beb39 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCLogQueryDAOTest.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCLogQueryDAOTest.java @@ -153,6 +153,19 @@ void buildSQL_withSegmentIdAndSpanId_shouldIncludeBothConditions() { assertThat(result.parameters()).contains(1); } + @Test + void buildSQL_withOtlpSpanId_shouldIncludeItsCondition() { + final TraceScopeCondition traceCondition = new TraceScopeCondition(); + traceCondition.setTraceId("trace-abc"); + traceCondition.setOtlpSpanId("0123456789abcdef"); + + final SQLAndParameters result = dao.buildSQL( + null, null, null, traceCondition, Order.DES, 0, 10, null, null, null, null, TABLE); + + assertThat(result.sql()).contains("and " + AbstractLogRecord.OTLP_SPAN_ID + " = ?"); + assertThat(result.parameters()).contains("0123456789abcdef"); + } + @Test void buildSQL_withSingleTag_shouldUseInnerJoin() { final List tags = Collections.singletonList(new Tag("level", "ERROR")); diff --git a/test/e2e-v2/cases/otlp-traces/native/banyandb/e2e.yaml b/test/e2e-v2/cases/otlp-traces/native/banyandb/e2e.yaml index a7659be6d22f..2030f7b7941b 100644 --- a/test/e2e-v2/cases/otlp-traces/native/banyandb/e2e.yaml +++ b/test/e2e-v2/cases/otlp-traces/native/banyandb/e2e.yaml @@ -48,3 +48,4 @@ verify: - includes: - ../traceql-cases.yaml - ../tal-cases.yaml + - ../log-cases.yaml diff --git a/test/e2e-v2/cases/otlp-traces/native/es/e2e.yaml b/test/e2e-v2/cases/otlp-traces/native/es/e2e.yaml index 556c22b3b82d..993577251f7f 100644 --- a/test/e2e-v2/cases/otlp-traces/native/es/e2e.yaml +++ b/test/e2e-v2/cases/otlp-traces/native/es/e2e.yaml @@ -48,3 +48,4 @@ verify: - includes: - ../traceql-cases.yaml - ../tal-cases.yaml + - ../log-cases.yaml diff --git a/test/e2e-v2/cases/otlp-traces/native/expected/logs-by-otlp-span.yml b/test/e2e-v2/cases/otlp-traces/native/expected/logs-by-otlp-span.yml new file mode 100644 index 000000000000..8615d0bf751b --- /dev/null +++ b/test/e2e-v2/cases/otlp-traces/native/expected/logs-by-otlp-span.yml @@ -0,0 +1,26 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +errors: 0 +# A query OAP rejected, or one that found nothing, has no rows for `contains` to check — so count them. +bySpanCount: {{ gt .bySpanCount 0 }} +bySpan: +{{- contains .bySpan }} +- serviceName: e2e-otlp-log + traceId: 0af7651916cd43dd8448eb211c80319c + content: a line written in a span +{{- end }} +byOtherSpan: 0 +mixedRefused: true diff --git a/test/e2e-v2/cases/otlp-traces/native/log-cases.yaml b/test/e2e-v2/cases/otlp-traces/native/log-cases.yaml new file mode 100644 index 000000000000..7b87dd62de9e --- /dev/null +++ b/test/e2e-v2/cases/otlp-traces/native/log-cases.yaml @@ -0,0 +1,35 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +cases: + # An OTLP log record written in a span keeps the span's trace ID and span ID, so the log is read by its span + # (relatedTrace.otlpSpanId) and not by another span of the same trace. Sent as OTLP/JSON, whose IDs are hex + # rather than the base64 protobuf's JSON mapping reads. A condition naming an OTLP span and a native span at + # once is refused: the two address a span in conflicting ways. + - query: | + curl -s -X POST http://${oap_host}:${oap_12800}/v1/logs -H 'Content-Type: application/json' \ + -d '{"resourceLogs":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"e2e-otlp-log"}}]},"scopeLogs":[{"logRecords":[{"timeUnixNano":"'$(date +%s)'000000000","body":{"stringValue":"a line written in a span"},"traceId":"0af7651916cd43dd8448eb211c80319c","spanId":"b7ad6b7169203331"}]}]}]}' > /dev/null + sleep 5 + logs_of_span() { + curl -s -X POST http://${oap_host}:${oap_12800}/graphql -H 'Content-Type: application/json' \ + -d '{"query":"query($c: LogQueryCondition){ queryLogs(condition: $c){ logs { serviceName traceId content } } }","variables":{"c":{"relatedTrace":{"traceId":"0af7651916cd43dd8448eb211c80319c","otlpSpanId":"'"$1"'"'"$2"'},"paging":{"pageNum":1,"pageSize":10}}}}' + } + jq -n --argjson span "$(logs_of_span b7ad6b7169203331)" --argjson other "$(logs_of_span 00f067aa0ba902b7)" \ + --argjson mixed "$(logs_of_span b7ad6b7169203331 ',"segmentId":"a-segment","spanId":0')" \ + '{errors: ((($span.errors // []) + ($other.errors // [])) | length), + bySpanCount: (($span.data.queryLogs.logs // []) | length), bySpan: ($span.data.queryLogs.logs // []), + byOtherSpan: (($other.data.queryLogs.logs // []) | length), + mixedRefused: ((($mixed.errors // []) | length) > 0 and ($mixed.data.queryLogs == null))}' | yq -P + expected: expected/logs-by-otlp-span.yml diff --git a/test/e2e-v2/cases/otlp-traces/native/mysql/e2e.yaml b/test/e2e-v2/cases/otlp-traces/native/mysql/e2e.yaml index 189f6b11eb45..4dda17ced0c3 100644 --- a/test/e2e-v2/cases/otlp-traces/native/mysql/e2e.yaml +++ b/test/e2e-v2/cases/otlp-traces/native/mysql/e2e.yaml @@ -48,3 +48,4 @@ verify: - includes: - ../traceql-cases.yaml - ../tal-cases.yaml + - ../log-cases.yaml