diff --git a/README.md b/README.md index ce0fda406..0c612a428 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,7 @@ including 1. Java, .NET Core, NodeJS and PHP auto-instrument agents in SkyWalking format 1. Istio telemetry format 1. Zipkin v1/v2 format +1. Jaeger gRPC format. 1. Envoy metrics format (the metric entries itself is prometheus client [metric family](https://github.com/prometheus/client_model/blob/fd36f4220a901265f90734c3183c5f0c91daa0b8/metrics.proto#L77)) diff --git a/docs/en/setup/backend/backend-receivers.md b/docs/en/setup/backend/backend-receivers.md index a4302a99e..3c5ea57f5 100644 --- a/docs/en/setup/backend/backend-receivers.md +++ b/docs/en/setup/backend/backend-receivers.md @@ -12,6 +12,7 @@ We have following receivers, and `default` implementors are provided in our Apac 1. **istio-telemetry**. Istio telemetry is from Istio official bypass adaptor, this receiver match its gRPC services. 1. **envoy-metric**. Envoy `metrics_service` supported by this receiver. OAL script support all GAUGE type metrics. 1. **receiver_zipkin**. See [details](#zipkin-receiver). +1. **receiver_jaeger**. See [details](#jaeger-receiver). The sample settings of these receivers should be already in default `application.yml`, and also list here ```yaml @@ -69,7 +70,17 @@ But it wouldn't analysis metric from them. In most case, I suggest you could use Notice, in this mode, Zipkin receiver requires `zipkin-elasticsearch` storage implementation active. Read [this](backend-storage.md#elasticsearch-6-with-zipkin-trace-extension) to know how to active. -1. Analysis mode(Not production ready), receive Zipkin v1/v2 formats through HTTP service. Transform the trace to skywalking + +Use following config to active. +```yaml +receiver_zipkin: + default: + host: ${SW_RECEIVER_ZIPKIN_HOST:0.0.0.0} + port: ${SW_RECEIVER_ZIPKIN_PORT:9411} + contextPath: ${SW_RECEIVER_ZIPKIN_CONTEXT_PATH:/} +``` + +2. Analysis mode(Not production ready), receive Zipkin v1/v2 formats through HTTP service. Transform the trace to skywalking native format, and analysis like skywalking trace. This feature can't work in production env right now, because of Zipkin tag/endpoint value unpredictable, we can't make sure it fits production env requirements. @@ -81,4 +92,22 @@ receiver_zipkin: port: ${SW_RECEIVER_ZIPKIN_PORT:9411} contextPath: ${SW_RECEIVER_ZIPKIN_CONTEXT_PATH:/} needAnalysis: true -``` \ No newline at end of file +``` + +## Jaeger receiver +Jaeger receiver right now only works in `Tracing Mode`, and no analysis. +Jaeger receiver provides extra gRPC host/port, if absent, sharing-server host/port will be used, then core gRPC host/port. +Receiver requires `jaeger-elasticsearch` storage implementation active. +Read [this](backend-storage.md#elasticsearch-6-with-jaeger-trace-extension) to know how to active. + +Right now, you need [jaeger agent](https://www.jaegertracing.io/docs/1.11/architecture/#agent) to batch +send spans to SkyWalking oap server. Read [Jaeger Architecture](https://www.jaegertracing.io/docs/1.11/architecture/) +to get more details. + +Active the receiver. +```yaml +receiver_jaeger: + default: + gRPCHost: ${SW_RECEIVER_JAEGER_HOST:0.0.0.0} + gRPCPort: ${SW_RECEIVER_JAEGER_PORT:14250} +``` \ No newline at end of file diff --git a/docs/en/setup/backend/backend-storage.md b/docs/en/setup/backend/backend-storage.md index f769690fb..806215ba2 100644 --- a/docs/en/setup/backend/backend-storage.md +++ b/docs/en/setup/backend/backend-storage.md @@ -68,6 +68,25 @@ storage: concurrentRequests: ${SW_STORAGE_ES_CONCURRENT_REQUESTS:2} # the number of concurrent requests ``` +### ElasticSearch 6 with Jaeger trace extension +This implementation shares most of `elasticsearch`, just extend to support zipkin span storage. +It has all same configs. +```yaml +storage: + jaeger-elasticsearch: + nameSpace: ${SW_NAMESPACE:""} + clusterNodes: ${SW_STORAGE_ES_CLUSTER_NODES:localhost:9200} + user: ${SW_ES_USER:""} + password: ${SW_ES_PASSWORD:""} + indexShardsNumber: ${SW_STORAGE_ES_INDEX_SHARDS_NUMBER:2} + indexReplicasNumber: ${SW_STORAGE_ES_INDEX_REPLICAS_NUMBER:0} + # Batch process setting, refer to https://www.elastic.co/guide/en/elasticsearch/client/java-api/5.5/java-docs-bulk-processor.html + bulkActions: ${SW_STORAGE_ES_BULK_ACTIONS:2000} # Execute the bulk every 2000 requests + bulkSize: ${SW_STORAGE_ES_BULK_SIZE:20} # flush the bulk every 20mb + flushInterval: ${SW_STORAGE_ES_FLUSH_INTERVAL:10} # flush the bulk every 10 seconds whatever the number of requests + concurrentRequests: ${SW_STORAGE_ES_CONCURRENT_REQUESTS:2} # the number of concurrent requests +``` + ### About Namespace When namespace is set, names of all indexes in ElasticSearch will use it as prefix. diff --git a/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALLexer.g4 b/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALLexer.g4 index 8888c5e09..b9cb869ca 100644 --- a/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALLexer.g4 +++ b/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALLexer.g4 @@ -52,6 +52,9 @@ SRC_SERVICE_RELATION_SERVER_SIDE: 'service_relation_server_side'; SRC_SERVICE_RELATION_CLIENT_SIDE: 'service_relation_client_side'; SRC_ALARM_RECORD: 'alarm_record'; +SRC_ZIPKIN_SPAN: 'zipkin_span'; +SRC_JAEGER_SPAN: 'jaeger_span'; + // Literals BOOL_LITERAL: 'true' diff --git a/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALParser.g4 b/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALParser.g4 index 74fa29406..2622156d3 100644 --- a/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALParser.g4 +++ b/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALParser.g4 @@ -54,7 +54,8 @@ source SRC_SERVICE_RELATION | SRC_SERVICE_INSTANCE_RELATION | SRC_ENDPOINT_RELATION | SRC_SERVICE_INSTANCE_JVM_CPU | SRC_SERVICE_INSTANCE_JVM_MEMORY | SRC_SERVICE_INSTANCE_JVM_MEMORY_POOL | SRC_SERVICE_INSTANCE_JVM_GC |// JVM source of service instance SRC_SERVICE_INSTANCE_CLR_CPU | SRC_SERVICE_INSTANCE_CLR_GC | SRC_SERVICE_INSTANCE_CLR_THREAD | - SRC_ENVOY_INSTANCE_METRIC + SRC_ENVOY_INSTANCE_METRIC | + SRC_ZIPKIN_SPAN | SRC_JAEGER_SPAN ; disableSource diff --git a/oap-server/generated-analysis/src/main/resources/official_analysis.oal b/oap-server/generated-analysis/src/main/resources/official_analysis.oal index 211f296ec..66b664ab2 100644 --- a/oap-server/generated-analysis/src/main/resources/official_analysis.oal +++ b/oap-server/generated-analysis/src/main/resources/official_analysis.oal @@ -101,4 +101,6 @@ envoy_parent_connections_used = from(EnvoyInstanceMetric.value).filter(metricNam ///////// // disable(segment); // disable(endpoint_relation_server_side); -// disable(top_n_database_statement); \ No newline at end of file +// disable(top_n_database_statement); +// disable(zipkin_span); +// disable(jaeger_span); \ No newline at end of file diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java index e584cd564..0cfe991a3 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java @@ -91,6 +91,7 @@ public class CoreModuleProvider extends ModuleProvider { AnnotationScan scopeScan = new AnnotationScan(); scopeScan.registerListener(new DefaultScopeDefine.Listener()); scopeScan.registerListener(DisableRegister.INSTANCE); + scopeScan.registerListener(new DisableRegister.SingleDisableScanListener()); try { scopeScan.scan(null); } catch (IOException e) { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DisableRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DisableRegister.java index cc3041ed2..d90462e08 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DisableRegister.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DisableRegister.java @@ -51,4 +51,15 @@ public class DisableRegister implements AnnotationListener { public boolean include(String name) { return disableEntitySet.contains(name); } + + public static class SingleDisableScanListener implements AnnotationListener { + @Override public Class annotation() { + return Disable.class; + } + + @Override public void notify(Class aClass) { + String name = ((Disable)aClass.getAnnotation(Disable.class)).value(); + DisableRegister.INSTANCE.disableEntitySet.add(name); + } + } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java index aec1665f0..b9c323915 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java @@ -60,6 +60,7 @@ public class DefaultScopeDefine { public static final int SERVICE_INSTANCE_CLR_THREAD = 21; public static final int ENVOY_INSTANCE_METRIC = 22; public static final int ZIPKIN_SPAN = 23; + public static final int JAEGER_SPAN = 24; /** * Catalog of scope, the indicator processor could use this to group all generated indicators by oal tool. diff --git a/oap-server/server-receiver-plugin/jaeger-receiver-plugin/pom.xml b/oap-server/server-receiver-plugin/jaeger-receiver-plugin/pom.xml index a4afaa6a7..05e6c8d37 100644 --- a/oap-server/server-receiver-plugin/jaeger-receiver-plugin/pom.xml +++ b/oap-server/server-receiver-plugin/jaeger-receiver-plugin/pom.xml @@ -40,5 +40,10 @@ skywalking-sharing-server-plugin ${project.version} + + org.apache.skywalking + storage-jaeger-plugin + ${project.version} + \ No newline at end of file diff --git a/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerGRPCHandler.java b/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerGRPCHandler.java index e1dfd3f52..9cb52270c 100644 --- a/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerGRPCHandler.java +++ b/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerGRPCHandler.java @@ -18,9 +18,18 @@ package org.apache.skywalking.aop.server.receiver.jaeger; +import com.google.gson.JsonObject; +import com.google.protobuf.ByteString; import io.grpc.stub.StreamObserver; import io.jaegertracing.api_v2.*; -import org.apache.skywalking.oap.server.library.module.ModuleManager; +import java.time.Instant; +import java.util.Base64; +import org.apache.skywalking.apm.util.StringUtil; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.source.*; +import org.apache.skywalking.oap.server.library.util.*; +import org.apache.skywalking.oap.server.receiver.sharing.server.CoreRegisterLinker; +import org.apache.skywalking.oap.server.storage.plugin.jaeger.JaegerSpan; import org.slf4j.*; /** @@ -29,18 +38,106 @@ import org.slf4j.*; public class JaegerGRPCHandler extends CollectorServiceGrpc.CollectorServiceImplBase { private static final Logger logger = LoggerFactory.getLogger(JaegerGRPCHandler.class); - public JaegerGRPCHandler(ModuleManager manager) { + private SourceReceiver receiver; + private JaegerReceiverConfig config; + public JaegerGRPCHandler(SourceReceiver receiver, + JaegerReceiverConfig config) { + this.receiver = receiver; + this.config = config; } public void postSpans(Collector.PostSpansRequest request, StreamObserver responseObserver) { request.getBatch().getSpansList().forEach(span -> { - logger.debug(span.toString()); + try { + if (logger.isDebugEnabled()) { + logger.debug(span.toString()); + } + + JaegerSpan jaegerSpan = new JaegerSpan(); + jaegerSpan.setTraceId(format(span.getTraceId())); + jaegerSpan.setSpanId(format(span.getSpanId())); + Model.Process process = span.getProcess(); + int serviceId = Const.NONE; + String serviceName = null; + if (process != null) { + serviceName = process.getServiceName(); + } + if (StringUtil.isEmpty(serviceName)) { + serviceName = "UNKNOWN"; + } + serviceId = CoreRegisterLinker.getServiceInventoryCache().getServiceId(serviceName); + if (serviceId != Const.NONE) { + jaegerSpan.setServiceId(serviceId); + } else { + JsonObject properties = new JsonObject(); + if (process != null) { + process.getTagsList().forEach(keyValue -> { + String key = keyValue.getKey(); + Model.ValueType valueVType = keyValue.getVType(); + switch (valueVType) { + case STRING: + properties.addProperty(key, keyValue.getVStr()); + break; + case INT64: + properties.addProperty(key, keyValue.getVInt64()); + break; + case BOOL: + properties.addProperty(key, keyValue.getVBool()); + break; + case FLOAT64: + properties.addProperty(key, keyValue.getVFloat64()); + break; + } + }); + } + CoreRegisterLinker.getServiceInventoryRegister().getOrCreate(serviceName, properties); + } + + long duration = span.getDuration().getNanos() / 1_000_000; + jaegerSpan.setStartTime(Instant.ofEpochSecond(span.getStartTime().getSeconds(), span.getStartTime().getNanos()).toEpochMilli()); + long timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(jaegerSpan.getStartTime()); + jaegerSpan.setTimeBucket(timeBucket); + jaegerSpan.setEndTime(jaegerSpan.getStartTime() + duration); + jaegerSpan.setLatency((int)duration); + jaegerSpan.setDataBinary(span.toByteArray()); + jaegerSpan.setEndpointName(span.getOperationName()); + + int finalServiceId = serviceId; + span.getTagsList().forEach(tag -> { + String key = tag.getKey(); + if ("error".equals(key)) { + boolean status = tag.getVBool(); + jaegerSpan.setIsError(BooleanUtils.booleanToValue(status)); + } else if ("span.kind".equals(key)) { + String kind = tag.getVStr(); + if ("server".equals(kind) || "consumer".equals(kind)) { + String endpointName = span.getOperationName(); + jaegerSpan.setEndpointName(endpointName); + int endpointId = CoreRegisterLinker.getEndpointInventoryCache().getEndpointId(finalServiceId, endpointName, + DetectPoint.SERVER.ordinal()); + if (endpointId != Const.NONE) { + CoreRegisterLinker.getEndpointInventoryRegister().getOrCreate(finalServiceId, endpointName, DetectPoint.SERVER); + } + } + } + }); + + receiver.receive(jaegerSpan); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } }); responseObserver.onNext(Collector.PostSpansResponse.newBuilder().build()); responseObserver.onCompleted(); } + + private String format(ByteString bytes) { + Base64.Encoder encoder = Base64.getEncoder(); + return encoder.encodeToString(bytes.toByteArray()); + } + } diff --git a/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerReceiverConfig.java b/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerReceiverConfig.java index 275e9649d..780ea097f 100644 --- a/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerReceiverConfig.java +++ b/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerReceiverConfig.java @@ -24,5 +24,9 @@ import org.apache.skywalking.oap.server.library.module.ModuleConfig; @Setter @Getter public class JaegerReceiverConfig extends ModuleConfig { + private String gRPCHost = null; + private int gRPCPort = -1; + private int maxConcurrentCallsPerConnection; + private int maxMessageSize; private boolean registerJaegerEndpoint = true; } diff --git a/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerReceiverProvider.java b/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerReceiverProvider.java index 8f4931574..69ac9dfde 100644 --- a/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerReceiverProvider.java +++ b/oap-server/server-receiver-plugin/jaeger-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/jaeger/JaegerReceiverProvider.java @@ -18,15 +18,23 @@ package org.apache.skywalking.aop.server.receiver.jaeger; +import java.util.Objects; +import org.apache.logging.log4j.util.Strings; +import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister; +import org.apache.skywalking.oap.server.core.source.SourceReceiver; import org.apache.skywalking.oap.server.library.module.*; -import org.apache.skywalking.oap.server.receiver.sharing.server.SharingServerModule; +import org.apache.skywalking.oap.server.library.server.ServerException; +import org.apache.skywalking.oap.server.library.server.grpc.GRPCServer; +import org.apache.skywalking.oap.server.receiver.sharing.server.*; /** * @author wusheng */ public class JaegerReceiverProvider extends ModuleProvider { public static final String NAME = "default"; + private JaegerReceiverConfig config; + private GRPCServer grpcServer = null; @Override public String name() { return NAME; @@ -37,20 +45,45 @@ public class JaegerReceiverProvider extends ModuleProvider { } @Override public ModuleConfig createConfigBeanIfAbsent() { - return null; + config = new JaegerReceiverConfig(); + return config; } @Override public void prepare() throws ServiceNotProvidedException, ModuleStartException { - + if (config.getGRPCPort() > 0) { + grpcServer = new GRPCServer(Strings.isBlank(config.getGRPCHost()) ? "0.0.0.0" : config.getGRPCHost(), config.getGRPCPort()); + if (config.getMaxMessageSize() > 0) { + grpcServer.setMaxMessageSize(config.getMaxMessageSize()); + } + if (config.getMaxConcurrentCallsPerConnection() > 0) { + grpcServer.setMaxConcurrentCallsPerConnection(config.getMaxConcurrentCallsPerConnection()); + } + grpcServer.initialize(); + } } @Override public void start() throws ServiceNotProvidedException, ModuleStartException { - GRPCHandlerRegister grpcHandlerRegister = getManager().find(SharingServerModule.NAME).provider().getService(GRPCHandlerRegister.class); - grpcHandlerRegister.addHandler(new JaegerGRPCHandler(getManager())); + CoreRegisterLinker.setModuleManager(getManager()); + + SourceReceiver sourceReceiver = getManager().find(CoreModule.NAME).provider().getService(SourceReceiver.class); + + if (Objects.nonNull(grpcServer)) { + grpcServer.addHandler(new JaegerGRPCHandler(sourceReceiver, config)); + } else { + GRPCHandlerRegister grpcHandlerRegister = getManager().find(SharingServerModule.NAME).provider().getService(GRPCHandlerRegister.class); + grpcHandlerRegister.addHandler(new JaegerGRPCHandler(sourceReceiver, config)); + } + } @Override public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException { - + try { + if (Objects.nonNull(grpcServer)) { + grpcServer.start(); + } + } catch (ServerException e) { + throw new ModuleStartException(e.getMessage(), e); + } } @Override public String[] requiredModules() { diff --git a/oap-server/server-receiver-plugin/receiver-proto/pom.xml b/oap-server/server-receiver-plugin/receiver-proto/pom.xml index 4d6a634b4..37587649c 100644 --- a/oap-server/server-receiver-plugin/receiver-proto/pom.xml +++ b/oap-server/server-receiver-plugin/receiver-proto/pom.xml @@ -30,14 +30,6 @@ receiver-proto jar - - - org.apache.skywalking - skywalking-sharing-server-plugin - ${project.version} - - - diff --git a/oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/sharing/server/CoreRegisterLinker.java b/oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/sharing/server/CoreRegisterLinker.java index 9617ab4ea..f894590e7 100644 --- a/oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/sharing/server/CoreRegisterLinker.java +++ b/oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/sharing/server/CoreRegisterLinker.java @@ -19,6 +19,7 @@ package org.apache.skywalking.oap.server.receiver.sharing.server; import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.cache.*; import org.apache.skywalking.oap.server.core.register.service.*; import org.apache.skywalking.oap.server.library.module.ModuleManager; @@ -27,6 +28,8 @@ public class CoreRegisterLinker { private static volatile IServiceInventoryRegister SERVICE_INVENTORY_REGISTER; private static volatile IServiceInstanceInventoryRegister SERVICE_INSTANCE_INVENTORY_REGISTER; private static volatile IEndpointInventoryRegister ENDPOINT_INVENTORY_REGISTER; + private static volatile ServiceInventoryCache SERVICE_INVENTORY_CACHE; + private static volatile EndpointInventoryCache ENDPOINT_INVENTORY_CACHE; public static void setModuleManager(ModuleManager moduleManager) { CoreRegisterLinker.MODULE_MANAGER = moduleManager; @@ -52,4 +55,18 @@ public class CoreRegisterLinker { } return ENDPOINT_INVENTORY_REGISTER; } + + public static ServiceInventoryCache getServiceInventoryCache() { + if (SERVICE_INVENTORY_CACHE == null) { + SERVICE_INVENTORY_CACHE = MODULE_MANAGER.find(CoreModule.NAME).provider().getService(ServiceInventoryCache.class); + } + return SERVICE_INVENTORY_CACHE; + } + + public static EndpointInventoryCache getEndpointInventoryCache() { + if (ENDPOINT_INVENTORY_CACHE == null) { + ENDPOINT_INVENTORY_CACHE = MODULE_MANAGER.find(CoreModule.NAME).provider().getService(EndpointInventoryCache.class); + } + return ENDPOINT_INVENTORY_CACHE; + } } diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml index 0b575c73f..c1c14bd4d 100644 --- a/oap-server/server-starter/src/main/resources/application.yml +++ b/oap-server/server-starter/src/main/resources/application.yml @@ -103,8 +103,10 @@ envoy-metric: # host: ${SW_RECEIVER_ZIPKIN_HOST:0.0.0.0} # port: ${SW_RECEIVER_ZIPKIN_PORT:9411} # contextPath: ${SW_RECEIVER_ZIPKIN_CONTEXT_PATH:/} -receiver_jaeger: - default: +#receiver_jaeger: +# default: +# gRPCHost: ${SW_RECEIVER_JAEGER_HOST:0.0.0.0} +# gRPCPort: ${SW_RECEIVER_JAEGER_PORT:14250} query: graphql: path: ${SW_QUERY_GRAPHQL_PATH:/graphql} diff --git a/oap-server/server-storage-plugin/pom.xml b/oap-server/server-storage-plugin/pom.xml index 99291e12e..27867ac7c 100644 --- a/oap-server/server-storage-plugin/pom.xml +++ b/oap-server/server-storage-plugin/pom.xml @@ -31,6 +31,7 @@ storage-jdbc-hikaricp-plugin storage-elasticsearch-plugin storage-zipkin-plugin + storage-jaeger-plugin \ No newline at end of file diff --git a/oap-server/server-storage-plugin/storage-jaeger-plugin/pom.xml b/oap-server/server-storage-plugin/storage-jaeger-plugin/pom.xml new file mode 100644 index 000000000..93975e941 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jaeger-plugin/pom.xml @@ -0,0 +1,44 @@ + + + + + + server-storage-plugin + org.apache.skywalking + 6.1.0-SNAPSHOT + + 4.0.0 + + storage-jaeger-plugin + + + + org.apache.skywalking + storage-elasticsearch-plugin + ${project.version} + + + org.apache.skywalking + receiver-proto + ${project.version} + + + \ No newline at end of file diff --git a/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpan.java b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpan.java new file mode 100644 index 000000000..46e62edf9 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpan.java @@ -0,0 +1,52 @@ +/* + * 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.storage.plugin.jaeger; + +import lombok.*; +import org.apache.skywalking.oap.server.core.source.*; + +import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.*; + +/** + * @author peng-yongsheng + */ +@ScopeDeclaration(id = JAEGER_SPAN, name = "JaegerSpan") +public class JaegerSpan extends Source { + + @Override public int scope() { + return DefaultScopeDefine.JAEGER_SPAN; + } + + @Override public String getEntityId() { + return traceId + spanId; + } + + @Setter @Getter private String traceId; + @Setter @Getter private String spanId; + @Setter @Getter private int serviceId; + @Setter @Getter private int serviceInstanceId; + @Setter @Getter private String endpointName; + @Setter @Getter private int endpointId; + @Setter @Getter private long startTime; + @Setter @Getter private long endTime; + @Setter @Getter private int latency; + @Setter @Getter private int isError; + @Setter @Getter private byte[] dataBinary; + @Setter @Getter private int encode; +} diff --git a/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpanRecord.java b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpanRecord.java new file mode 100644 index 000000000..69c34e1a6 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpanRecord.java @@ -0,0 +1,112 @@ +/* + * 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.storage.plugin.jaeger; + +import java.util.*; +import lombok.*; +import org.apache.skywalking.apm.util.StringUtil; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.analysis.record.Record; +import org.apache.skywalking.oap.server.core.analysis.record.annotation.RecordType; +import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine; +import org.apache.skywalking.oap.server.core.storage.StorageBuilder; +import org.apache.skywalking.oap.server.core.storage.annotation.*; +import org.apache.skywalking.oap.server.library.util.CollectionUtils; + +@RecordType +@StorageEntity(name = JaegerSpanRecord.INDEX_NAME, builder = JaegerSpanRecord.Builder.class, sourceScopeId = DefaultScopeDefine.JAEGER_SPAN) +public class JaegerSpanRecord extends Record { + public static final String INDEX_NAME = "jaeger_span"; + public static final String TRACE_ID = "trace_id"; + public static final String SPAN_ID = "span_id"; + public static final String SERVICE_ID = "service_id"; + public static final String SERVICE_INSTANCE_ID = "service_instance_id"; + public static final String ENDPOINT_NAME = "endpoint_name"; + public static final String ENDPOINT_ID = "endpoint_id"; + public static final String START_TIME = "start_time"; + public static final String END_TIME = "end_time"; + public static final String LATENCY = "latency"; + public static final String IS_ERROR = "is_error"; + public static final String DATA_BINARY = "data_binary"; + public static final String ENCODE = "encode"; + + @Setter @Getter @Column(columnName = TRACE_ID) @IDColumn private String traceId; + @Setter @Getter @Column(columnName = SPAN_ID) @IDColumn private String spanId; + @Setter @Getter @Column(columnName = SERVICE_ID) @IDColumn private int serviceId; + @Setter @Getter @Column(columnName = SERVICE_INSTANCE_ID) @IDColumn private int serviceInstanceId; + @Setter @Getter @Column(columnName = ENDPOINT_NAME, matchQuery = true) @IDColumn private String endpointName; + @Setter @Getter @Column(columnName = ENDPOINT_ID) @IDColumn private int endpointId; + @Setter @Getter @Column(columnName = START_TIME) @IDColumn private long startTime; + @Setter @Getter @Column(columnName = END_TIME) @IDColumn private long endTime; + @Setter @Getter @Column(columnName = LATENCY) @IDColumn private int latency; + @Setter @Getter @Column(columnName = IS_ERROR) @IDColumn private int isError; + @Setter @Getter @Column(columnName = DATA_BINARY) @IDColumn private byte[] dataBinary; + @Setter @Getter @Column(columnName = ENCODE) @IDColumn private int encode; + + @Override public String id() { + return traceId + "-" + spanId; + } + + public static class Builder implements StorageBuilder { + + @Override public Map data2Map(JaegerSpanRecord storageData) { + Map map = new HashMap<>(); + map.put(TRACE_ID, storageData.getTraceId()); + map.put(SPAN_ID, storageData.getSpanId()); + map.put(SERVICE_ID, storageData.getServiceId()); + map.put(SERVICE_INSTANCE_ID, storageData.getServiceInstanceId()); + map.put(ENDPOINT_NAME, storageData.getEndpointName()); + map.put(ENDPOINT_ID, storageData.getEndpointId()); + map.put(START_TIME, storageData.getStartTime()); + map.put(END_TIME, storageData.getEndTime()); + map.put(LATENCY, storageData.getLatency()); + map.put(IS_ERROR, storageData.getIsError()); + map.put(TIME_BUCKET, storageData.getTimeBucket()); + if (CollectionUtils.isEmpty(storageData.getDataBinary())) { + map.put(DATA_BINARY, Const.EMPTY_STRING); + } else { + map.put(DATA_BINARY, new String(Base64.getEncoder().encode(storageData.getDataBinary()))); + } + map.put(ENCODE, storageData.getEncode()); + return map; + } + + @Override public JaegerSpanRecord map2Data(Map dbMap) { + JaegerSpanRecord record = new JaegerSpanRecord(); + record.setTraceId((String)dbMap.get(TRACE_ID)); + record.setSpanId((String)dbMap.get(SPAN_ID)); + record.setServiceId(((Number)dbMap.get(SERVICE_ID)).intValue()); + record.setServiceInstanceId(((Number)dbMap.get(SERVICE_INSTANCE_ID)).intValue()); + record.setEndpointName((String)dbMap.get(ENDPOINT_NAME)); + record.setEndpointId(((Number)dbMap.get(ENDPOINT_ID)).intValue()); + record.setStartTime(((Number)dbMap.get(START_TIME)).longValue()); + record.setEndTime(((Number)dbMap.get(END_TIME)).longValue()); + record.setLatency(((Number)dbMap.get(LATENCY)).intValue()); + record.setIsError(((Number)dbMap.get(IS_ERROR)).intValue()); + record.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue()); + if (StringUtil.isEmpty((String)dbMap.get(DATA_BINARY))) { + record.setDataBinary(new byte[] {}); + } else { + record.setDataBinary(Base64.getDecoder().decode((String)dbMap.get(DATA_BINARY))); + } + record.setEncode(((Number)dbMap.get(ENCODE)).intValue()); + return record; + } + } +} diff --git a/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpanRecordDispatcher.java b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpanRecordDispatcher.java new file mode 100644 index 000000000..60e227c26 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/JaegerSpanRecordDispatcher.java @@ -0,0 +1,48 @@ +/* + * 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.storage.plugin.jaeger; + +import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher; +import org.apache.skywalking.oap.server.core.analysis.worker.RecordProcess; + +/** + * Dispatch for Zipkin native mode spans. + * + * @author wusheng + */ +public class JaegerSpanRecordDispatcher implements SourceDispatcher { + @Override public void dispatch(JaegerSpan source) { + JaegerSpanRecord segment = new JaegerSpanRecord(); + segment.setTraceId(source.getTraceId()); + segment.setSpanId(source.getSpanId()); + segment.setServiceId(source.getServiceId()); + segment.setServiceInstanceId(source.getServiceInstanceId()); + segment.setEndpointName(source.getEndpointName()); + segment.setEndpointId(source.getEndpointId()); + segment.setStartTime(source.getStartTime()); + segment.setEndTime(source.getEndTime()); + segment.setLatency(source.getLatency()); + segment.setIsError(source.getIsError()); + segment.setDataBinary(source.getDataBinary()); + segment.setTimeBucket(source.getTimeBucket()); + segment.setEncode(source.getEncode()); + + RecordProcess.INSTANCE.in(segment); + } +} diff --git a/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/elasticsearch/JaegerStorageModuleElasticsearchProvider.java b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/elasticsearch/JaegerStorageModuleElasticsearchProvider.java new file mode 100644 index 000000000..74519fa68 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/elasticsearch/JaegerStorageModuleElasticsearchProvider.java @@ -0,0 +1,57 @@ +/* + * 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.storage.plugin.jaeger.elasticsearch; + +import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.cache.ServiceInventoryCache; +import org.apache.skywalking.oap.server.core.storage.query.ITraceQueryDAO; +import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchProvider; +import org.slf4j.*; + +/** + * @author peng-yongsheng + */ +public class JaegerStorageModuleElasticsearchProvider extends StorageModuleElasticsearchProvider { + + private static final Logger logger = LoggerFactory.getLogger(JaegerStorageModuleElasticsearchProvider.class); + private JaegerTraceQueryEsDAO traceQueryEsDAO; + + @Override + public String name() { + return "jaeger-elasticsearch"; + } + + @Override + public void prepare() throws ServiceNotProvidedException { + super.prepare(); + traceQueryEsDAO = new JaegerTraceQueryEsDAO(elasticSearchClient); + this.registerServiceImplementation(ITraceQueryDAO.class, traceQueryEsDAO); + } + + @Override public void notifyAfterCompleted() { + super.notifyAfterCompleted(); + traceQueryEsDAO.setServiceInventoryCache(getManager().find(CoreModule.NAME).provider().getService(ServiceInventoryCache.class)); + } + + @Override + public String[] requiredModules() { + return new String[] {CoreModule.NAME}; + } +} diff --git a/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/elasticsearch/JaegerTraceQueryEsDAO.java b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/elasticsearch/JaegerTraceQueryEsDAO.java new file mode 100644 index 000000000..cf9a24471 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jaeger/elasticsearch/JaegerTraceQueryEsDAO.java @@ -0,0 +1,274 @@ +/* + * 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.storage.plugin.jaeger.elasticsearch; + +import com.google.common.base.Strings; +import com.google.protobuf.ByteString; +import io.jaegertracing.api_v2.Model; +import java.io.IOException; +import java.time.Instant; +import java.util.*; +import lombok.Setter; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord; +import org.apache.skywalking.oap.server.core.cache.ServiceInventoryCache; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.core.storage.query.ITraceQueryDAO; +import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient; +import org.apache.skywalking.oap.server.library.util.BooleanUtils; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO; +import org.apache.skywalking.oap.server.storage.plugin.jaeger.JaegerSpanRecord; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.index.query.*; +import org.elasticsearch.search.SearchHit; +import org.elasticsearch.search.aggregations.*; +import org.elasticsearch.search.aggregations.bucket.terms.*; +import org.elasticsearch.search.aggregations.metrics.max.Max; +import org.elasticsearch.search.aggregations.metrics.min.Min; +import org.elasticsearch.search.builder.SearchSourceBuilder; +import org.elasticsearch.search.sort.SortOrder; + +import static org.apache.skywalking.oap.server.core.analysis.record.Record.TIME_BUCKET; +import static org.apache.skywalking.oap.server.storage.plugin.jaeger.JaegerSpanRecord.*; + +public class JaegerTraceQueryEsDAO extends EsDAO implements ITraceQueryDAO { + @Setter + private ServiceInventoryCache serviceInventoryCache; + + public JaegerTraceQueryEsDAO( + ElasticSearchClient client) { + super(client); + } + + @Override + public TraceBrief queryBasicTraces(long startSecondTB, long endSecondTB, long minDuration, long maxDuration, + String endpointName, int serviceId, int serviceInstanceId, int endpointId, String traceId, int limit, int from, + TraceState traceState, QueryOrder queryOrder) throws IOException { + + SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); + + BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); + sourceBuilder.query(boolQueryBuilder); + List mustQueryList = boolQueryBuilder.must(); + + if (startSecondTB != 0 && endSecondTB != 0) { + mustQueryList.add(QueryBuilders.rangeQuery(TIME_BUCKET).gte(startSecondTB).lte(endSecondTB)); + } + + if (minDuration != 0 || maxDuration != 0) { + RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(LATENCY); + if (minDuration != 0) { + rangeQueryBuilder.gte(minDuration); + } + if (maxDuration != 0) { + rangeQueryBuilder.lte(maxDuration); + } + boolQueryBuilder.must().add(rangeQueryBuilder); + } + if (!Strings.isNullOrEmpty(endpointName)) { + mustQueryList.add(QueryBuilders.matchPhraseQuery(ENDPOINT_NAME, endpointName)); + } + if (serviceId != 0) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(SERVICE_ID, serviceId)); + } + if (serviceInstanceId != 0) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(SERVICE_INSTANCE_ID, serviceInstanceId)); + } + if (endpointId != 0) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(ENDPOINT_ID, endpointId)); + } + if (!Strings.isNullOrEmpty(traceId)) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(TRACE_ID, traceId)); + } + switch (traceState) { + case ERROR: + mustQueryList.add(QueryBuilders.matchQuery(IS_ERROR, BooleanUtils.TRUE)); + break; + case SUCCESS: + mustQueryList.add(QueryBuilders.matchQuery(IS_ERROR, BooleanUtils.FALSE)); + break; + } + + TermsAggregationBuilder builder = AggregationBuilders.terms(TRACE_ID).field(TRACE_ID).size(limit) + .subAggregation( + AggregationBuilders.max(LATENCY).field(LATENCY) + ) + .subAggregation( + AggregationBuilders.min(START_TIME).field(START_TIME) + ); + switch (queryOrder) { + case BY_START_TIME: + builder.order(BucketOrder.aggregation(START_TIME, false)); + break; + case BY_DURATION: + builder.order(BucketOrder.aggregation(LATENCY, false)); + break; + } + sourceBuilder.aggregation(builder); + + SearchResponse response = getClient().search(JaegerSpanRecord.INDEX_NAME, sourceBuilder); + + TraceBrief traceBrief = new TraceBrief(); + + Terms terms = response.getAggregations().get(TRACE_ID); + + for (Terms.Bucket termsBucket : terms.getBuckets()) { + BasicTrace basicTrace = new BasicTrace(); + + basicTrace.setSegmentId(termsBucket.getKeyAsString()); + Min startTime = termsBucket.getAggregations().get(START_TIME); + Max latency = termsBucket.getAggregations().get(LATENCY); + basicTrace.setStart(String.valueOf((long)startTime.getValue())); + basicTrace.getEndpointNames().add(""); + basicTrace.setDuration((int)latency.getValue()); + basicTrace.setError(false); + basicTrace.getTraceIds().add(termsBucket.getKeyAsString()); + traceBrief.getTraces().add(basicTrace); + } + + return traceBrief; + } + + @Override public List queryByTraceId(String traceId) throws IOException { + return Collections.emptyList(); + } + + @Override public List doFlexibleTraceQuery( + String traceId) throws IOException { + SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); + sourceBuilder.query(QueryBuilders.termQuery(TRACE_ID, traceId)); + sourceBuilder.sort(START_TIME, SortOrder.ASC); + sourceBuilder.size(1000); + + SearchResponse response = getClient().search(JaegerSpanRecord.INDEX_NAME, sourceBuilder); + + List spanList = new ArrayList<>(); + + for (SearchHit searchHit : response.getHits().getHits()) { + int serviceId = ((Number)searchHit.getSourceAsMap().get(SERVICE_ID)).intValue(); + long startTime = ((Number)searchHit.getSourceAsMap().get(START_TIME)).longValue(); + long endTime = ((Number)searchHit.getSourceAsMap().get(END_TIME)).longValue(); + String dataBinaryBase64 = (String)searchHit.getSourceAsMap().get(SegmentRecord.DATA_BINARY); + + Model.Span jaegerSpan = Model.Span.newBuilder().mergeFrom(Base64.getDecoder().decode(dataBinaryBase64)).build(); + + Span swSpan = new Span(); + + swSpan.setTraceId(format(jaegerSpan.getTraceId())); + swSpan.setEndpointName(jaegerSpan.getOperationName()); + swSpan.setStartTime(startTime); + swSpan.setEndTime(endTime); + jaegerSpan.getTagsList().forEach(keyValue -> { + String key = keyValue.getKey(); + Model.ValueType valueVType = keyValue.getVType(); + switch (valueVType) { + case STRING: + swSpan.getTags().add(new KeyValue(key, keyValue.getVStr())); + break; + case INT64: + swSpan.getTags().add(new KeyValue(key, keyValue.getVInt64() + "")); + break; + case BOOL: + swSpan.getTags().add(new KeyValue(key, keyValue.getVBool() + "")); + break; + case FLOAT64: + swSpan.getTags().add(new KeyValue(key, keyValue.getVFloat64() + "")); + break; + } + swSpan.setType("Local"); + if ("span.kind".equals(key)) { + String kind = keyValue.getVStr(); + if ("server".equals(kind) || "consumer".equals(kind)) { + swSpan.setType("Entry"); + } else if ("client".equals(kind) || "producer".equals(kind)) { + swSpan.setType("Exit"); + } + } + }); + jaegerSpan.getLogsList().forEach(log -> { + LogEntity entity = new LogEntity(); + boolean hasTimestamp = log.hasTimestamp(); + if (hasTimestamp) { + long time = Instant.ofEpochSecond(log.getTimestamp().getSeconds(), log.getTimestamp().getNanos()).toEpochMilli(); + entity.setTime(time); + } + log.getFieldsList().forEach(field -> { + String key = field.getKey(); + Model.ValueType valueVType = field.getVType(); + switch (valueVType) { + case STRING: + entity.getData().add(new KeyValue(key, field.getVStr())); + break; + case INT64: + entity.getData().add(new KeyValue(key, field.getVInt64() + "")); + break; + case BOOL: + entity.getData().add(new KeyValue(key, field.getVBool() + "")); + break; + case FLOAT64: + entity.getData().add(new KeyValue(key, field.getVFloat64() + "")); + break; + } + }); + + swSpan.getLogs().add(entity); + }); + + if (serviceId != Const.NONE) { + swSpan.setServiceCode(serviceInventoryCache.get(serviceId).getName()); + } else { + swSpan.setServiceCode("UNKNOWN"); + } + swSpan.setSpanId(0); + swSpan.setParentSpanId(-1); + String spanId = id(format(jaegerSpan.getTraceId()), format(jaegerSpan.getSpanId())); + swSpan.setSegmentSpanId(spanId); + swSpan.setSegmentId(spanId); + + List spanReferencesList = jaegerSpan.getReferencesList(); + if (spanReferencesList.size() > 0) { + spanReferencesList.forEach(jaegerRef -> { + Ref ref = new Ref(); + ref.setTraceId(format(jaegerRef.getTraceId())); + String parentId = id(format(jaegerRef.getTraceId()), format(jaegerRef.getSpanId())); + ref.setParentSegmentId(parentId); + ref.setType(RefType.CROSS_PROCESS); + ref.setParentSpanId(0); + + swSpan.getRefs().add(ref); + swSpan.setSegmentParentSpanId(parentId); + }); + } else { + swSpan.setRoot(true); + swSpan.setSegmentParentSpanId(""); + } + spanList.add(swSpan); + } + return spanList; + } + + private String id(String traceId, String spanId) { + return traceId + "_" + spanId; + } + + private String format(ByteString bytes) { + Base64.Encoder encoder = Base64.getEncoder(); + return encoder.encodeToString(bytes.toByteArray()); + } +} diff --git a/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider new file mode 100644 index 000000000..d982d7fdd --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jaeger-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider @@ -0,0 +1,37 @@ +# +# 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. +# +# + +# +# 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. +# +# + +org.apache.skywalking.oap.server.storage.plugin.jaeger.elasticsearch.JaegerStorageModuleElasticsearchProvider \ No newline at end of file diff --git a/oap-server/server-storage-plugin/storage-zipkin-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/zipkin/elasticsearch/ZipkinTraceQueryEsDAO.java b/oap-server/server-storage-plugin/storage-zipkin-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/zipkin/elasticsearch/ZipkinTraceQueryEsDAO.java index 448e74573..d7ae20c57 100644 --- a/oap-server/server-storage-plugin/storage-zipkin-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/zipkin/elasticsearch/ZipkinTraceQueryEsDAO.java +++ b/oap-server/server-storage-plugin/storage-zipkin-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/zipkin/elasticsearch/ZipkinTraceQueryEsDAO.java @@ -22,6 +22,7 @@ import com.google.common.base.Strings; import java.io.IOException; import java.util.*; import lombok.Setter; +import org.apache.skywalking.apm.util.StringUtil; import org.apache.skywalking.oap.server.core.Const; import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord; import org.apache.skywalking.oap.server.core.cache.ServiceInventoryCache; @@ -158,7 +159,6 @@ public class ZipkinTraceQueryEsDAO extends EsDAO implements ITraceQueryDAO { List spanList = new ArrayList<>(); - boolean isFirst = true; for (SearchHit searchHit : response.getHits().getHits()) { int serviceId = ((Number)searchHit.getSourceAsMap().get(SERVICE_ID)).intValue(); String dataBinaryBase64 = (String)searchHit.getSourceAsMap().get(SegmentRecord.DATA_BINARY); @@ -201,10 +201,9 @@ public class ZipkinTraceQueryEsDAO extends EsDAO implements ITraceQueryDAO { } - if (isFirst) { + if (StringUtil.isEmpty(span.parentId())) { swSpan.setRoot(true); swSpan.setSegmentParentSpanId(""); - isFirst = false; } else { Ref ref = new Ref(); ref.setTraceId(span.traceId());