diff --git a/oap-server/server-bootstrap/src/main/resources/application.yml b/oap-server/server-bootstrap/src/main/resources/application.yml index 29c6754d4..00bcceabc 100755 --- a/oap-server/server-bootstrap/src/main/resources/application.yml +++ b/oap-server/server-bootstrap/src/main/resources/application.yml @@ -370,6 +370,7 @@ receiver_zipkin: jettyIdleTimeOut: ${SW_RECEIVER_ZIPKIN_JETTY_IDLE_TIMEOUT:30000} jettyAcceptorPriorityDelta: ${SW_RECEIVER_ZIPKIN_JETTY_DELTA:0} jettyAcceptQueueSize: ${SW_RECEIVER_ZIPKIN_QUEUE_SIZE:0} + instanceNameRule: ${SW_RECEIVER_ZIPKIN_INSTANCE_NAME_RULE:[spring.instance_id,node_id]} receiver_jaeger: selector: ${SW_RECEIVER_JAEGER:-} diff --git a/oap-server/server-bootstrap/src/test/java/org/apache/skywalking/oap/server/starter/config/ApplicationConfigLoaderTestCase.java b/oap-server/server-bootstrap/src/test/java/org/apache/skywalking/oap/server/starter/config/ApplicationConfigLoaderTestCase.java index fb0929ea0..ccc7bf2b9 100644 --- a/oap-server/server-bootstrap/src/test/java/org/apache/skywalking/oap/server/starter/config/ApplicationConfigLoaderTestCase.java +++ b/oap-server/server-bootstrap/src/test/java/org/apache/skywalking/oap/server/starter/config/ApplicationConfigLoaderTestCase.java @@ -18,13 +18,16 @@ package org.apache.skywalking.oap.server.starter.config; -import java.util.Properties; import org.apache.skywalking.oap.server.library.module.ApplicationConfiguration; import org.junit.Before; import org.junit.Test; +import java.util.List; +import java.util.Properties; + import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThat; public class ApplicationConfigLoaderTestCase { @@ -34,6 +37,7 @@ public class ApplicationConfigLoaderTestCase { @Before public void setUp() throws ConfigFileNotFoundException { System.setProperty("SW_STORAGE", "mysql"); + System.setProperty("SW_RECEIVER_ZIPKIN", "default"); ApplicationConfigLoader configLoader = new ApplicationConfigLoader(); applicationConfiguration = configLoader.load(); } @@ -47,4 +51,13 @@ public class ApplicationConfigLoaderTestCase { Properties properties = (Properties) providerConfig.get("properties"); assertThat(properties.get("jdbcUrl"), is("jdbc:mysql://localhost:3306/swtest")); } + + @Test + public void testLoadListTypeConfig() { + Properties providerConfig = applicationConfiguration.getModuleConfiguration("receiver_zipkin") + .getProviderConfiguration("default"); + List instanceNameRule = (List) providerConfig.get("instanceNameRule"); + assertEquals(2, instanceNameRule.size()); + } + } diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java index ded3932dd..113913078 100644 --- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java +++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java @@ -22,6 +22,9 @@ import lombok.Getter; import lombok.Setter; import org.apache.skywalking.oap.server.library.module.ModuleConfig; +import java.util.ArrayList; +import java.util.List; + @Setter @Getter public class ZipkinReceiverConfig extends ModuleConfig { @@ -33,4 +36,5 @@ public class ZipkinReceiverConfig extends ModuleConfig { private long jettyIdleTimeOut = 30000; private int jettyAcceptorPriorityDelta = 0; private int jettyAcceptQueueSize = 0; + private List instanceNameRule = new ArrayList<>(); } diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanProcessor.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanProcessor.java index 8ae612195..f03589577 100644 --- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanProcessor.java +++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanProcessor.java @@ -24,6 +24,7 @@ import java.io.InputStream; import java.util.List; import java.util.zip.GZIPInputStream; import javax.servlet.http.HttpServletRequest; + import lombok.RequiredArgsConstructor; import org.apache.skywalking.oap.server.core.config.NamingControl; import org.apache.skywalking.oap.server.core.source.SourceReceiver; @@ -49,7 +50,7 @@ public class SpanProcessor { List spanList = decoder.decodeList(out.toByteArray()); - SpanForward forward = new SpanForward(namingControl, receiver); + SpanForward forward = new SpanForward(namingControl, receiver, config); forward.send(spanList); } } diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java index 3cde23aa3..9482e278d 100644 --- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java +++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java @@ -19,6 +19,7 @@ package org.apache.skywalking.oap.server.receiver.zipkin.trace; import java.util.List; + import lombok.RequiredArgsConstructor; import org.apache.skywalking.apm.util.StringUtil; import org.apache.skywalking.oap.server.core.analysis.IDManager; @@ -29,6 +30,7 @@ import org.apache.skywalking.oap.server.core.source.EndpointMeta; import org.apache.skywalking.oap.server.core.source.ServiceMeta; import org.apache.skywalking.oap.server.core.source.SourceReceiver; import org.apache.skywalking.oap.server.library.util.BooleanUtils; +import org.apache.skywalking.oap.server.receiver.zipkin.ZipkinReceiverConfig; import org.apache.skywalking.oap.server.receiver.zipkin.handler.SpanEncode; import org.apache.skywalking.oap.server.storage.plugin.zipkin.ZipkinSpan; import zipkin2.Span; @@ -36,8 +38,10 @@ import zipkin2.codec.SpanBytesEncoder; @RequiredArgsConstructor public class SpanForward { + private static final String DEFAULT_SERVICE_INSTANCE_NAME = " unknown_instance"; private final NamingControl namingControl; private final SourceReceiver receiver; + private final ZipkinReceiverConfig config; public void send(List spanList) { spanList.forEach(span -> { @@ -49,7 +53,11 @@ public class SpanForward { serviceName = "Unknown"; } serviceName = namingControl.formatServiceName(serviceName); - zipkinSpan.setServiceId(IDManager.ServiceID.buildId(serviceName, NodeType.Normal)); + String serviceId = IDManager.ServiceID.buildId(serviceName, NodeType.Normal); + zipkinSpan.setServiceId(serviceId); + String serviceInstanceName = this.getServiceInstanceName(span); + serviceInstanceName = namingControl.formatInstanceName(serviceInstanceName); + zipkinSpan.setServiceInstanceId(IDManager.ServiceInstanceID.buildId(serviceId, serviceInstanceName)); long startTime = span.timestampAsLong() / 1000; zipkinSpan.setStartTime(startTime); @@ -93,4 +101,14 @@ public class SpanForward { receiver.receive(serviceMeta); }); } + + private String getServiceInstanceName(Span span) { + for (String tagName : config.getInstanceNameRule()) { + String serviceInstanceName = span.tags().get(tagName); + if (StringUtil.isNotEmpty(serviceInstanceName)) { + return serviceInstanceName; + } + } + return DEFAULT_SERVICE_INSTANCE_NAME; + } } diff --git a/oap-server/server-storage-plugin/storage-zipkin-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/zipkin/elasticsearch/ZipkinTraceQueryEs7DAO.java b/oap-server/server-storage-plugin/storage-zipkin-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/zipkin/elasticsearch/ZipkinTraceQueryEs7DAO.java index b930c4368..b99f00513 100644 --- a/oap-server/server-storage-plugin/storage-zipkin-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/zipkin/elasticsearch/ZipkinTraceQueryEs7DAO.java +++ b/oap-server/server-storage-plugin/storage-zipkin-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/zipkin/elasticsearch/ZipkinTraceQueryEs7DAO.java @@ -19,11 +19,13 @@ package org.apache.skywalking.oap.server.storage.plugin.zipkin.elasticsearch; import com.google.common.base.Strings; + import java.io.IOException; import java.util.ArrayList; import java.util.Base64; import java.util.Collections; import java.util.List; + import org.apache.skywalking.apm.util.StringUtil; import org.apache.skywalking.oap.server.core.analysis.IDManager; import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.Tag; @@ -157,7 +159,7 @@ public class ZipkinTraceQueryEs7DAO extends EsDAO implements ITraceQueryDAO { BasicTrace basicTrace = new BasicTrace(); final ZipkinSpanRecord zipkinSpanRecord = new ZipkinSpanRecord.Builder().storage2Entity( - searchHit.getSourceAsMap()); + searchHit.getSourceAsMap()); basicTrace.setSegmentId(zipkinSpanRecord.getSpanId()); basicTrace.setStart(String.valueOf((long) zipkinSpanRecord.getStartTime())); @@ -179,7 +181,7 @@ public class ZipkinTraceQueryEs7DAO extends EsDAO implements ITraceQueryDAO { @Override public List doFlexibleTraceQuery( - String traceId) throws IOException { + String traceId) throws IOException { SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); sourceBuilder.query(QueryBuilders.termQuery(TRACE_ID, traceId)); sourceBuilder.sort(START_TIME, SortOrder.ASC); @@ -191,6 +193,7 @@ public class ZipkinTraceQueryEs7DAO extends EsDAO implements ITraceQueryDAO { for (SearchHit searchHit : response.getHits().getHits()) { String serviceId = (String) searchHit.getSourceAsMap().get(SERVICE_ID); + String serviceInstanceId = (String) searchHit.getSourceAsMap().get(SERVICE_INSTANCE_ID); String dataBinaryBase64 = (String) searchHit.getSourceAsMap().get(SegmentRecord.DATA_BINARY); Span span = SpanBytesDecoder.PROTO3.decodeOne(Base64.getDecoder().decode(dataBinaryBase64)); @@ -211,9 +214,14 @@ public class ZipkinTraceQueryEs7DAO extends EsDAO implements ITraceQueryDAO { }); if (StringUtil.isNotEmpty(serviceId)) { final IDManager.ServiceID.ServiceIDDefinition serviceIDDefinition = IDManager.ServiceID.analysisId( - serviceId); + serviceId); swSpan.setServiceCode(serviceIDDefinition.getName()); } + if (StringUtil.isNotEmpty(serviceInstanceId)) { + final IDManager.ServiceInstanceID.InstanceIDDefinition instanceIDDefinition = IDManager.ServiceInstanceID.analysisId( + serviceInstanceId); + swSpan.setServiceInstanceName(instanceIDDefinition.getName()); + } swSpan.setSpanId(0); swSpan.setParentSpanId(-1); swSpan.setSegmentSpanId(span.id());