wu-sheng commented on code in PR #13770: URL: https://github.com/apache/skywalking/pull/13770#discussion_r3007790146
########## oap-server/server-query-plugin/traceql-plugin/src/main/java/org/apache/skywalking/oap/query/traceql/converter/SkyWalkingOTLPConverter.java: ########## @@ -0,0 +1,461 @@ +/* + * 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.query.traceql.converter; + +import com.google.protobuf.ByteString; +import io.grafana.tempo.tempopb.TraceByIDResponse; +import io.opentelemetry.proto.common.v1.AnyValue; +import io.opentelemetry.proto.common.v1.InstrumentationScope; +import io.opentelemetry.proto.common.v1.KeyValue; +import io.opentelemetry.proto.resource.v1.Resource; +import io.opentelemetry.proto.trace.v1.ResourceSpans; +import io.opentelemetry.proto.trace.v1.ScopeSpans; +import io.opentelemetry.proto.trace.v1.Status; +import java.util.LinkedHashSet; +import java.util.Set; +import org.apache.commons.codec.DecoderException; +import org.apache.commons.codec.binary.Hex; +import org.apache.skywalking.oap.query.traceql.entity.SearchResponse; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.query.type.LogEntity; +import org.apache.skywalking.oap.server.core.query.type.Ref; +import org.apache.skywalking.oap.server.core.query.type.trace.v2.TraceList; +import org.apache.skywalking.oap.server.library.util.StringUtil; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.skywalking.oap.query.traceql.handler.TraceQLApiHandler.SERVICE_NAME; +import static org.apache.skywalking.oap.query.traceql.handler.TraceQLApiHandler.SPAN_KIND; + +/** + * Converter for transforming SkyWalking trace data to OpenTelemetry Protocol (OTLP) format. + * <p> + * Note: This class uses fully qualified names for some classes to avoid naming conflicts: + * - org.apache.skywalking.oap.server.core.query.type.Trace (SkyWalking Trace) + * - org.apache.skywalking.oap.server.core.query.type.Span (SkyWalking Span) + * - org.apache.skywalking.oap.server.core.query.type.KeyValue (SkyWalking KeyValue) + * - io.grafana.tempo.tempopb.Trace (Tempo/OTLP Trace - used via fully qualified name) + * - io.opentelemetry.proto.trace.v1.Span (OTLP Span - used via fully qualified name) + * - io.opentelemetry.proto.common.v1.KeyValue (OTLP KeyValue - imported as KeyValue) + */ +public class SkyWalkingOTLPConverter { + + /** + * Convert SkyWalking trace to OTLP Protobuf format. + * + * @param traceId Trace ID of the trace to convert + * @param swTrace SkyWalking Trace object + * @return TraceByIDResponse in Protobuf format + */ + public static TraceByIDResponse convertToProtobuf(String traceId, org.apache.skywalking.oap.server.core.query.type.Trace swTrace) throws DecoderException { + if (swTrace == null || swTrace.getSpans().isEmpty()) { + return TraceByIDResponse.newBuilder().build(); + } + + io.grafana.tempo.tempopb.Trace.Builder traceBuilder = io.grafana.tempo.tempopb.Trace.newBuilder(); + + // Group spans by service + Map<String, List<org.apache.skywalking.oap.server.core.query.type.Span>> spansByService = new HashMap<>(); + for (org.apache.skywalking.oap.server.core.query.type.Span swSpan : swTrace.getSpans()) { + String serviceName = swSpan.getServiceCode(); + if (StringUtil.isEmpty(serviceName)) { + serviceName = "unknown"; + } + spansByService.computeIfAbsent(serviceName, k -> new ArrayList<>()).add(swSpan); + } + + // Convert each service group to ResourceSpans + for (Map.Entry<String, List<org.apache.skywalking.oap.server.core.query.type.Span>> entry : spansByService.entrySet()) { + String serviceName = entry.getKey(); + List<org.apache.skywalking.oap.server.core.query.type.Span> serviceSpans = entry.getValue(); + + ResourceSpans.Builder rsBuilder = ResourceSpans.newBuilder(); + + // Set resource (service) + rsBuilder.setResource(Resource.newBuilder() + .addAttributes(KeyValue.newBuilder() + .setKey("service.name") + .setValue(AnyValue.newBuilder() + .setStringValue(serviceName) + .build()) + .build()) + .build()); + + ScopeSpans.Builder ssBuilder = ScopeSpans.newBuilder(); + ssBuilder.setScope(InstrumentationScope.newBuilder() + .setName("skywalking-tracer") + .setVersion("0.1.0") + .build()); + + // Convert each SkyWalking span to OTLP Span + // Note: SkyWalking traceId already encode to hex string in query list, + // in order to make it compatible with Grafana Tempo, we keep it as hex string and + // can directly convert it to bytes for OTLP traceId + for (org.apache.skywalking.oap.server.core.query.type.Span swSpan : serviceSpans) { + io.opentelemetry.proto.trace.v1.Span.Builder spanBuilder = io.opentelemetry.proto.trace.v1.Span.newBuilder(); + spanBuilder.setTraceId(ByteString.copyFrom(OTLPConverter.hexToBytes(traceId))); + + String spanId = swSpan.getSegmentSpanId(); + if (StringUtil.isEmpty(spanId)) { + // Fallback: use segmentId + spanId + spanId = swSpan.getSegmentId() + Const.SEGMENT_SPAN_SPLIT + swSpan.getSpanId(); + } + spanBuilder.setSpanId(ByteString.copyFromUtf8(spanId)); + + // Set parent span ID + if (swSpan.getParentSpanId() >= 0 && !swSpan.isRoot()) { + String parentSpanId = swSpan.getSegmentParentSpanId(); + if (StringUtil.isEmpty(parentSpanId)) { + parentSpanId = swSpan.getSegmentId() + Const.SEGMENT_SPAN_SPLIT + swSpan.getParentSpanId(); + } + spanBuilder.setParentSpanId(ByteString.copyFromUtf8(parentSpanId)); + } else if (!swSpan.getRefs().isEmpty()) { + // Handle cross-segment reference + Ref ref = swSpan.getRefs().stream().filter(r -> r.getTraceId().equals(swSpan.getTraceId())).findFirst().orElse(null); + if (ref != null) { + String refParentSpanId = ref.getParentSegmentId() + Const.SEGMENT_SPAN_SPLIT + ref.getParentSpanId(); + spanBuilder.setParentSpanId(ByteString.copyFromUtf8(refParentSpanId)); + } + } + + // Set span name + String spanName = swSpan.getEndpointName(); + if (StringUtil.isEmpty(spanName)) { + spanName = swSpan.getType(); + } + spanBuilder.setName(StringUtil.isNotEmpty(spanName) ? spanName : "unknown"); + + // Set span kind based on type + spanBuilder.setKind(convertSpanKind(swSpan.getType())); + + // Set timestamps (convert milliseconds to nanoseconds) + spanBuilder.setStartTimeUnixNano(swSpan.getStartTime() * 1_000_000); + spanBuilder.setEndTimeUnixNano(swSpan.getEndTime() * 1_000_000); + + // Add tags as attributes + if (!swSpan.getTags().isEmpty()) { + for (org.apache.skywalking.oap.server.core.query.type.KeyValue tag : swSpan.getTags()) { + spanBuilder.addAttributes(KeyValue.newBuilder() + .setKey(tag.getKey()) + .setValue(AnyValue.newBuilder() + .setStringValue(tag.getValue()) + .build()) + .build()); + } + } + + // Add standard attributes + if (StringUtil.isNotEmpty(swSpan.getPeer())) { + spanBuilder.addAttributes(KeyValue.newBuilder() + .setKey("peer.address") + .setValue(AnyValue.newBuilder() + .setStringValue(swSpan.getPeer()) + .build()) + .build()); + } + + if (StringUtil.isNotEmpty(swSpan.getComponent())) { + spanBuilder.addAttributes(KeyValue.newBuilder() + .setKey("component") + .setValue(AnyValue.newBuilder() + .setStringValue(swSpan.getComponent()) + .build()) + .build()); + } + + if (StringUtil.isNotEmpty(swSpan.getLayer())) { + spanBuilder.addAttributes(KeyValue.newBuilder() + .setKey("layer") + .setValue(AnyValue.newBuilder() + .setStringValue(swSpan.getLayer()) + .build()) + .build()); + } + + // Set status + Status.Builder statusBuilder = Status.newBuilder(); + if (swSpan.isError()) { + statusBuilder.setCode(Status.StatusCode.STATUS_CODE_ERROR); + statusBuilder.setMessage("Error occurred"); + } else { + statusBuilder.setCode(Status.StatusCode.STATUS_CODE_OK); + } + spanBuilder.setStatus(statusBuilder.build()); + + // Convert logs to events + if (!swSpan.getLogs().isEmpty()) { + for (LogEntity log : swSpan.getLogs()) { + io.opentelemetry.proto.trace.v1.Span.Event.Builder eventBuilder = + io.opentelemetry.proto.trace.v1.Span.Event.newBuilder() + .setTimeUnixNano(log.getTime() * 1_000_000) + .setName("log"); + + // Add log data as event attributes + if (!log.getData().isEmpty()) { + for (org.apache.skywalking.oap.server.core.query.type.KeyValue data : log.getData()) { + eventBuilder.addAttributes(KeyValue.newBuilder() + .setKey(data.getKey()) + .setValue(AnyValue.newBuilder() + .setStringValue(data.getValue()) + .build()) + .build()); + } + } + + spanBuilder.addEvents(eventBuilder.build()); + } + } + + // Convert attachedEvents to OTLP events + if (!swSpan.getAttachedEvents().isEmpty()) { + for (org.apache.skywalking.oap.server.core.query.type.SpanAttachedEvent attachedEvent : swSpan.getAttachedEvents()) { + long timeUnixNano = attachedEvent.getStartTime().getSeconds() * 1_000_000_000L + + attachedEvent.getStartTime().getNanos(); + io.opentelemetry.proto.trace.v1.Span.Event.Builder eventBuilder = + io.opentelemetry.proto.trace.v1.Span.Event.newBuilder() + .setTimeUnixNano(timeUnixNano) + .setName(attachedEvent.getEvent()); + + for (org.apache.skywalking.oap.server.core.query.type.KeyValue tag : attachedEvent.getTags()) { + eventBuilder.addAttributes(KeyValue.newBuilder() + .setKey(tag.getKey()) + .setValue(AnyValue.newBuilder() + .setStringValue(tag.getValue()) + .build()) + .build()); + } + + for (org.apache.skywalking.oap.server.core.query.type.KeyNumericValue summary : attachedEvent.getSummary()) { + eventBuilder.addAttributes(KeyValue.newBuilder() + .setKey(summary.getKey()) + // convert numeric value to string for AnyValue, make it trans to JOSN format easier + .setValue(AnyValue.newBuilder() + .setStringValue(String.valueOf(summary.getValue())) + .build()) + .build()); + } + + spanBuilder.addEvents(eventBuilder.build()); + } + } + + ssBuilder.addSpans(spanBuilder.build()); + } + + rsBuilder.addScopeSpans(ssBuilder.build()); + traceBuilder.addResourceSpans(rsBuilder.build()); + } + + return TraceByIDResponse.newBuilder() + .setTrace(traceBuilder.build()) + .build(); + } + + /** + * Convert TraceList to SearchResponse format. + * + * @param traceList SkyWalking trace list + * @param allowedTags Only span attributes whose key is in this set are included; null means all + * @return SearchResponse containing the converted traces + */ + public static SearchResponse convertTraceListToSearchResponse(TraceList traceList, Set<String> allowedTags) { + SearchResponse response = new SearchResponse(); + List<SearchResponse.Trace> traces = new ArrayList<>(); + + if (traceList != null && traceList.getTraces() != null) { + for (org.apache.skywalking.oap.server.core.query.type.trace.v2.TraceV2 trace : traceList.getTraces()) { + if (trace.getSpans().isEmpty()) { + continue; + } + traces.add(convertSWTraceToSearchTrace(trace, allowedTags)); + } + } + + response.setTraces(traces); + return response; + } + + /** + * Convert a single SkyWalking TraceV2 to SearchResponse.Trace. + * + * @param swTrace SkyWalking TraceV2 + * @param allowedTags Only span attributes whose key is in this set are included; null means all + * @return SearchResponse.Trace + */ + private static SearchResponse.Trace convertSWTraceToSearchTrace( + org.apache.skywalking.oap.server.core.query.type.trace.v2.TraceV2 swTrace, Set<String> allowedTags) { + + SearchResponse.Trace trace = new SearchResponse.Trace(); + + // Find root span to get trace metadata + org.apache.skywalking.oap.server.core.query.type.Span rootSpan = + swTrace.getSpans().stream() + .filter(org.apache.skywalking.oap.server.core.query.type.Span::isRoot) + .findFirst() + .orElse(swTrace.getSpans().get(0)); + + trace.setTraceID(encodeTraceId(rootSpan.getTraceId())); + trace.setRootServiceName(rootSpan.getServiceCode()); + trace.setRootTraceName(rootSpan.getEndpointName()); + + // Calculate duration and start time + long minStartTime = Long.MAX_VALUE; + long maxEndTime = Long.MIN_VALUE; + for (org.apache.skywalking.oap.server.core.query.type.Span span : swTrace.getSpans()) { + minStartTime = Math.min(minStartTime, span.getStartTime()); + maxEndTime = Math.max(maxEndTime, span.getEndTime()); + } + trace.setStartTimeUnixNano(String.valueOf(minStartTime * 1_000_000)); + trace.setDurationMs((int) (maxEndTime - minStartTime)); + + // First pass: collect all attribute keys across all spans. + // Grafana has a bug (search.go:369) where spans missing an attribute key that other + // spans have cause a type panic: Append("") on a []*string field instead of Append(nil). + // By ensuring all spans have the same keys (padding missing ones with ""), we avoid + // the else branch in Grafana entirely. + Set<String> allSpanAttrKeys = new LinkedHashSet<>(); + for (org.apache.skywalking.oap.server.core.query.type.Span span : swTrace.getSpans()) { + for (org.apache.skywalking.oap.server.core.query.type.KeyValue tag : span.getTags()) { + allSpanAttrKeys.add(tag.getKey()); + } + } + // Restrict to allowed tags when configured + if (allowedTags != null) { + allSpanAttrKeys.retainAll(allowedTags); + } + // Add fixed tags + allSpanAttrKeys.add(SPAN_KIND); + allSpanAttrKeys.add(SERVICE_NAME); + SearchResponse.SpanSet spanSet = new SearchResponse.SpanSet(); + for (org.apache.skywalking.oap.server.core.query.type.Span span : swTrace.getSpans()) { + spanSet.getSpans().add(convertSWSpanToSearchSpan(span, allSpanAttrKeys)); + } + spanSet.setMatched(spanSet.getSpans().size()); + trace.getSpanSets().add(spanSet); + + return trace; + } + + /** + * Convert a single SkyWalking Span to SearchResponse.Span. + * All keys in {@code allSpanAttrKeys} are written (missing ones padded with "") so that + * every span in the same SpanSet has identical attribute keys — required to avoid a + * Grafana search.go:369 type panic on []*string DataFrame fields. + * + * @param swSpan SkyWalking Span + * @param allSpanAttrKeys Ordered, already-filtered set of attribute keys to output + * @return SearchResponse.Span + */ + private static SearchResponse.Span convertSWSpanToSearchSpan( + org.apache.skywalking.oap.server.core.query.type.Span swSpan, Set<String> allSpanAttrKeys) { + + SearchResponse.Span span = new SearchResponse.Span(); + span.setSpanID(swSpan.getSegmentSpanId()); + span.setStartTimeUnixNano(String.valueOf(swSpan.getStartTime() * 1_000_000)); + span.setDurationNanos(String.valueOf((swSpan.getEndTime() - swSpan.getStartTime()) * 1_000_000)); + + // Build attribute map for this span + Map<String, String> spanAttrMap = new java.util.LinkedHashMap<>(); Review Comment: ```suggestion Map<String, String> spanAttrMap = new LinkedHashMap<>(); ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
