Skip to content

Commit

Permalink
Update OTelProtoCodec for InstrumentationLibrary to InstrumentationSc…
Browse files Browse the repository at this point in the history 
…ope rename (#2114)

* Add support for ScopeSpans migration to otel-trace-raw-processor

Signed-off-by: Chase Engelbrecht <engechas@amazon.com>

* Remove unused import

Signed-off-by: Chase Engelbrecht <engechas@amazon.com>

* Run spotless

Signed-off-by: Chase Engelbrecht <engechas@amazon.com>

Signed-off-by: Chase Engelbrecht <engechas@amazon.com>
  • Loading branch information
@engechas
engechas authored Jan 4, 2023
1 parent ce2e7af commit 988014a
Show file tree
Hide file tree
Showing 6 changed files with 712 additions and 54 deletions.
84 changes: 58 additions & 26 deletions ...to-common/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OTelProtoCodec.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import io.opentelemetry.proto.resource.v1.Resource;
import io.opentelemetry.proto.trace.v1.InstrumentationLibrarySpans;
import io.opentelemetry.proto.trace.v1.ResourceSpans;
import io.opentelemetry.proto.trace.v1.ScopeSpans;
import io.opentelemetry.proto.trace.v1.Status;
import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.binary.Hex;
Expand All @@ -44,8 +45,8 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand Down Expand Up @@ -79,8 +80,8 @@ public class OTelProtoCodec {
private static final String AT = "@";
private static final String METRIC_ATTRIBUTES = "metric.attributes";
private static final String EXEMPLAR_ATTRIBUTES = "exemplar.attributes";
private static final String INSTRUMENTATION_SCOPE_NAME = "instrumentationScope.name";
private static final String INSTRUMENTATION_SCOPE_VERSION = "instrumentationScope.version";
static final String INSTRUMENTATION_SCOPE_NAME = "instrumentationScope.name";
static final String INSTRUMENTATION_SCOPE_VERSION = "instrumentationScope.version";

public static final Function<String, String> REPLACE_DOT_WITH_AT = i -> i.replace(DOT, AT);
/**
Expand Down Expand Up @@ -144,20 +145,51 @@ public List<Span> parseExportTraceServiceRequest(final ExportTraceServiceRequest
}

protected List<Span> parseResourceSpans(final ResourceSpans resourceSpans) {
final List<Span> spans = new LinkedList<>();

final String serviceName = getServiceName(resourceSpans.getResource()).orElse(null);
final Map<String, Object> resourceAttributes = getResourceAttributes(resourceSpans.getResource());
for (InstrumentationLibrarySpans is : resourceSpans.getInstrumentationLibrarySpansList()) {
for (io.opentelemetry.proto.trace.v1.Span sp : is.getSpansList()) {
final Span span = parseSpan(sp, is.getInstrumentationLibrary(), serviceName, resourceAttributes);
spans.add(span);
}

if (resourceSpans.getScopeSpansList().size() > 0) {
return parseScopeSpans(resourceSpans.getScopeSpansList(), serviceName, resourceAttributes);
}
return spans;

if (resourceSpans.getInstrumentationLibrarySpansList().size() > 0) {
return parseInstrumentationLibrarySpans(resourceSpans.getInstrumentationLibrarySpansList(), serviceName, resourceAttributes);
}

LOG.debug("No spans found to parse from ResourceSpans object: {}", resourceSpans);
return Collections.emptyList();
}

private List<Span> parseScopeSpans(final List<ScopeSpans> scopeSpansList, final String serviceName, final Map<String, Object> resourceAttributes) {
return scopeSpansList.stream()
.map(scopeSpans -> parseSpans(scopeSpans.getSpansList(), scopeSpans.getScope(),
OTelProtoCodec::getInstrumentationScopeAttributes, serviceName, resourceAttributes))
.flatMap(Collection::stream)
.collect(Collectors.toList());
}

private List<Span> parseInstrumentationLibrarySpans(final List<InstrumentationLibrarySpans> instrumentationLibrarySpansList,
final String serviceName, final Map<String, Object> resourceAttributes) {
return instrumentationLibrarySpansList.stream()
.map(instrumentationLibrarySpans -> parseSpans(instrumentationLibrarySpans.getSpansList(),
instrumentationLibrarySpans.getInstrumentationLibrary(), this::getInstrumentationLibraryAttributes,
serviceName, resourceAttributes))
.flatMap(Collection::stream)
.collect(Collectors.toList());
}

private <T> List<Span> parseSpans(final List<io.opentelemetry.proto.trace.v1.Span> spans, final T scope,
final Function<T, Map<String, Object>> scopeAttributesGetter,
final String serviceName, final Map<String, Object> resourceAttributes) {
return spans.stream()
.map(span -> {
final Map<String, Object> scopeAttributes = scopeAttributesGetter.apply(scope);
return parseSpan(span, scopeAttributes, serviceName, resourceAttributes);
})
.collect(Collectors.toList());
}

protected Span parseSpan(final io.opentelemetry.proto.trace.v1.Span sp, final InstrumentationLibrary instrumentationLibrary,
protected Span parseSpan(final io.opentelemetry.proto.trace.v1.Span sp, final Map<String, Object> instrumentationScopeAttributes,
final String serviceName, final Map<String, Object> resourceAttributes) {
return JacksonSpan.builder()
.withSpanId(Hex.encodeHexString(sp.getSpanId().toByteArray()))
Expand All @@ -173,7 +205,7 @@ protected Span parseSpan(final io.opentelemetry.proto.trace.v1.Span sp, final In
Arrays.asList(
getSpanAttributes(sp),
resourceAttributes,
getInstrumentationLibraryAttributes(instrumentationLibrary),
instrumentationScopeAttributes,
getSpanStatusAttributes(sp.getStatus())
)
))
Expand Down Expand Up @@ -308,10 +340,10 @@ protected TraceGroupFields getTraceGroupFields(final io.opentelemetry.proto.trac
protected Map<String, Object> getInstrumentationLibraryAttributes(final InstrumentationLibrary instrumentationLibrary) {
final Map<String, Object> instrumentationAttr = new HashMap<>();
if (!instrumentationLibrary.getName().isEmpty()) {
instrumentationAttr.put(INSTRUMENTATION_LIBRARY_NAME, instrumentationLibrary.getName());
instrumentationAttr.put(INSTRUMENTATION_SCOPE_NAME, instrumentationLibrary.getName());
}
if (!instrumentationLibrary.getVersion().isEmpty()) {
instrumentationAttr.put(INSTRUMENTATION_LIBRARY_VERSION, instrumentationLibrary.getVersion());
instrumentationAttr.put(INSTRUMENTATION_SCOPE_VERSION, instrumentationLibrary.getVersion());
}
return instrumentationAttr;
}
Expand Down Expand Up @@ -355,12 +387,12 @@ public ResourceSpans convertToResourceSpans(final Span span) throws UnsupportedE
final Map<String, Object> allAttributes = span.getAttributes();
final Resource resource = constructResource(span.getServiceName(), allAttributes);
rsBuilder.setResource(resource);
final InstrumentationLibrarySpans.Builder instrumentationLibrarySpansBuilder = InstrumentationLibrarySpans.newBuilder();
final InstrumentationLibrary instrumentationLibrary = constructInstrumentationLibrary(allAttributes);
instrumentationLibrarySpansBuilder.setInstrumentationLibrary(instrumentationLibrary);
final ScopeSpans.Builder scopeSpansBuilder = ScopeSpans.newBuilder();
final InstrumentationScope instrumentationScope = constructInstrumentationScope(allAttributes);
scopeSpansBuilder.setScope(instrumentationScope);
final io.opentelemetry.proto.trace.v1.Span otelProtoSpan = constructSpan(span, allAttributes);
instrumentationLibrarySpansBuilder.addSpans(otelProtoSpan);
rsBuilder.addInstrumentationLibrarySpans(instrumentationLibrarySpansBuilder);
scopeSpansBuilder.addSpans(otelProtoSpan);
rsBuilder.addScopeSpans(scopeSpansBuilder);
return rsBuilder.build();
}

Expand Down Expand Up @@ -411,12 +443,12 @@ protected List<KeyValue> getResourceAttributes(final Map<String, Object> attribu
return result;
}

protected InstrumentationLibrary constructInstrumentationLibrary(final Map<String, Object> attributes) {
final InstrumentationLibrary.Builder builder = InstrumentationLibrary.newBuilder();
final Optional<String> instrumentationLibraryName = Optional.ofNullable((String) attributes.get(INSTRUMENTATION_LIBRARY_NAME));
final Optional<String> instrumentationLibraryVersion = Optional.ofNullable((String) attributes.get(INSTRUMENTATION_LIBRARY_VERSION));
instrumentationLibraryName.ifPresent(builder::setName);
instrumentationLibraryVersion.ifPresent(builder::setVersion);
protected InstrumentationScope constructInstrumentationScope(final Map<String, Object> attributes) {
final InstrumentationScope.Builder builder = InstrumentationScope.newBuilder();
final Optional<String> instrumentationScopeName = Optional.ofNullable((String) attributes.get(INSTRUMENTATION_SCOPE_NAME));
final Optional<String> instrumentationScopeVersion = Optional.ofNullable((String) attributes.get(INSTRUMENTATION_SCOPE_VERSION));
instrumentationScopeName.ifPresent(builder::setName);
instrumentationScopeVersion.ifPresent(builder::setVersion);
return builder.build();
}

Expand Down
Loading

0 comments on commit 988014a

Please sign in to comment.