kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[Feature] Zipkin module support BanyanDB storage. (#10004)

This commit is contained in:
Wan Kai 2022-11-22 22:38:38 +08:00 committed by GitHub
parent 87f3fb1890
commit ffa4990938
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 475 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/skywalking.yaml vendored
View File

@ -606,6 +606,8 @@ jobs:
config: test/e2e-v2/cases/zipkin/postgres/e2e.yaml
- name: Zipkin Kafka
config: test/e2e-v2/cases/zipkin/kafka/e2e.yaml
- name: Zipkin BanyanDB
config: test/e2e-v2/cases/zipkin/banyandb/e2e.yaml
- name: MySQL-Sharding storage
config: test/e2e-v2/cases/storage/mysql/sharding/e2e.yaml

2
docs/en/changes/changes.md
View File

@ -116,6 +116,8 @@
* Changed system variable `SW_SUPERDATASET_STORAGE_DAY_STEP` to `SW_STORAGE_ES_SUPER_DATASET_DAY_STEP` to be consistent with other ES storage related variables.
* Fix ESEventQueryDAO missing metric_table boolQuery criteria.
* Add default entity name(`_blank`) if absent to avoid NPE in the decoding. This caused `Can't split xxx id into 2 parts`.
* Zipkin module support BanyanDB storage.
* Zipkin traces query API, sort the result set by start time by default.
#### UI

1
docs/en/setup/backend/zipkin-trace.md
View File

@ -39,7 +39,6 @@ receiver-zipkin:
## Zipkin query
The Zipkin receiver makes the OAP server work as an alternative Zipkin server implementation for query traces.
It implemented `ZipkinQueryApiV2` through the HTTP service, supporting Zipkin-lens UI.
**Notice: Zipkin query API implementation does not support BanyanDB yet.**
Use the following config to activate it.

2
oap-server-bom/pom.xml
View File

@ -73,7 +73,7 @@
<awaitility.version>3.0.0</awaitility.version>
<httpcore.version>4.4.13</httpcore.version>
<commons-compress.version>1.21</commons-compress.version>
<banyandb-java-client.version>0.2.0</banyandb-java-client.version>
<banyandb-java-client.version>0.3.0-SNAPSHOT</banyandb-java-client.version>
<kafka-clients.version>2.8.1</kafka-clients.version>
<spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
</properties>

51
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/zipkin/ZipkinSpanRecord.java
View File

@ -19,8 +19,10 @@
package org.apache.skywalking.oap.server.core.zipkin;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import java.util.List;
import java.util.Map;
import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.core.Const;
@ -38,6 +40,8 @@ import org.apache.skywalking.oap.server.core.storage.type.Convert2Storage;
import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
import org.apache.skywalking.oap.server.library.util.BooleanUtils;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import zipkin2.Endpoint;
import zipkin2.Span;
import static org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord.TRACE_ID;
import static org.apache.skywalking.oap.server.core.analysis.record.Record.TIME_BUCKET;
@ -249,4 +253,51 @@ public class ZipkinSpanRecord extends Record {
}
}
}
public static Span buildSpanFromRecord(ZipkinSpanRecord record) {
Span.Builder span = Span.newBuilder();
span.traceId(record.getTraceId());
span.id(record.getSpanId());
span.parentId(record.getParentId());
span.kind(Span.Kind.valueOf(record.getKind()));
span.timestamp(record.getTimestamp());
span.duration(record.getDuration());
span.name(record.getName());
//Build localEndpoint
Endpoint.Builder localEndpoint = Endpoint.newBuilder();
localEndpoint.serviceName(record.getLocalEndpointServiceName());
if (!StringUtil.isEmpty(record.getLocalEndpointIPV4())) {
localEndpoint.parseIp(record.getLocalEndpointIPV4());
} else {
localEndpoint.parseIp(record.getLocalEndpointIPV6());
}
localEndpoint.port(record.getLocalEndpointPort());
span.localEndpoint(localEndpoint.build());
//Build remoteEndpoint
Endpoint.Builder remoteEndpoint = Endpoint.newBuilder();
remoteEndpoint.serviceName(record.getRemoteEndpointServiceName());
if (!StringUtil.isEmpty(record.getLocalEndpointIPV4())) {
remoteEndpoint.parseIp(record.getRemoteEndpointIPV4());
} else {
remoteEndpoint.parseIp(record.getRemoteEndpointIPV6());
}
remoteEndpoint.port(record.getRemoteEndpointPort());
span.remoteEndpoint(remoteEndpoint.build());
//Build tags
JsonObject tagsJson = record.getTags();
if (tagsJson != null) {
for (Map.Entry<String, JsonElement> tag : tagsJson.entrySet()) {
span.putTag(tag.getKey(), tag.getValue().getAsString());
}
}
//Build annotation
JsonObject annotationJson = record.getAnnotations();
if (annotationJson != null) {
for (Map.Entry<String, JsonElement> annotation : annotationJson.entrySet()) {
span.addAnnotation(Long.parseLong(annotation.getKey()), annotation.getValue().getAsString());
}
}
return span.build();
}
}

1
oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java
View File

@ -72,7 +72,6 @@ import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBP
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBSpanAttachedEventQueryDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBStorageDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBTraceQueryDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBZipkinQueryDAO;
import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;

297
oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBZipkinQueryDAO.java Normal file
View File

@ -0,0 +1,297 @@
/*
* 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.banyandb;
import com.google.common.collect.ImmutableSet;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.skywalking.banyandb.v1.client.AbstractCriteria;
import org.apache.skywalking.banyandb.v1.client.AbstractQuery;
import org.apache.skywalking.banyandb.v1.client.DataPoint;
import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
import org.apache.skywalking.banyandb.v1.client.RowEntity;
import org.apache.skywalking.banyandb.v1.client.StreamQuery;
import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
import org.apache.skywalking.banyandb.v1.client.TimestampRange;
import org.apache.skywalking.oap.server.core.query.input.Duration;
import org.apache.skywalking.oap.server.core.storage.query.IZipkinQueryDAO;
import org.apache.skywalking.oap.server.core.zipkin.ZipkinServiceRelationTraffic;
import org.apache.skywalking.oap.server.core.zipkin.ZipkinServiceSpanTraffic;
import org.apache.skywalking.oap.server.core.zipkin.ZipkinServiceTraffic;
import org.apache.skywalking.oap.server.core.zipkin.ZipkinSpanRecord;
import org.apache.skywalking.oap.server.library.util.CollectionUtils;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
import zipkin2.Span;
import zipkin2.storage.QueryRequest;
public class BanyanDBZipkinQueryDAO extends AbstractBanyanDBDAO implements IZipkinQueryDAO {
private final static int QUERY_MAX_SIZE = Integer.MAX_VALUE;
private static final Set<String> SERVICE_TRAFFIC_TAGS = ImmutableSet.of(ZipkinServiceTraffic.SERVICE_NAME);
private static final Set<String> REMOTE_SERVICE_TRAFFIC_TAGS = ImmutableSet.of(
ZipkinServiceRelationTraffic.REMOTE_SERVICE_NAME);
private static final Set<String> SPAN_TRAFFIC_TAGS = ImmutableSet.of(ZipkinServiceSpanTraffic.SPAN_NAME);
private static final Set<String> TRACE_ID = ImmutableSet.of(ZipkinSpanRecord.TRACE_ID);
private static final Set<String> TRACE_TAGS = ImmutableSet.of(
ZipkinSpanRecord.TRACE_ID,
ZipkinSpanRecord.SPAN_ID,
ZipkinSpanRecord.PARENT_ID,
ZipkinSpanRecord.KIND,
ZipkinSpanRecord.TIMESTAMP,
ZipkinSpanRecord.TIMESTAMP_MILLIS,
ZipkinSpanRecord.DURATION,
ZipkinSpanRecord.NAME,
ZipkinSpanRecord.DEBUG,
ZipkinSpanRecord.SHARED,
ZipkinSpanRecord.LOCAL_ENDPOINT_SERVICE_NAME,
ZipkinSpanRecord.LOCAL_ENDPOINT_IPV4,
ZipkinSpanRecord.LOCAL_ENDPOINT_IPV6,
ZipkinSpanRecord.LOCAL_ENDPOINT_PORT,
ZipkinSpanRecord.REMOTE_ENDPOINT_SERVICE_NAME,
ZipkinSpanRecord.REMOTE_ENDPOINT_IPV4,
ZipkinSpanRecord.REMOTE_ENDPOINT_IPV6,
ZipkinSpanRecord.REMOTE_ENDPOINT_PORT,
ZipkinSpanRecord.TAGS,
ZipkinSpanRecord.ANNOTATIONS
);
public BanyanDBZipkinQueryDAO(BanyanDBStorageClient client) {
super(client);
}
@Override
public List<String> getServiceNames() throws IOException {
MeasureQueryResponse resp =
query(ZipkinServiceTraffic.INDEX_NAME,
SERVICE_TRAFFIC_TAGS,
Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
@Override
protected void apply(MeasureQuery query) {
query.setLimit(QUERY_MAX_SIZE);
}
}
);
final List<String> services = new ArrayList<>();
for (final DataPoint dataPoint : resp.getDataPoints()) {
services.add(dataPoint.getTagValue(ZipkinServiceTraffic.SERVICE_NAME));
}
return services;
}
@Override
public List<String> getRemoteServiceNames(final String serviceName) throws IOException {
MeasureQueryResponse resp =
query(ZipkinServiceRelationTraffic.INDEX_NAME,
REMOTE_SERVICE_TRAFFIC_TAGS,
Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
@Override
protected void apply(MeasureQuery query) {
if (StringUtil.isNotEmpty(serviceName)) {
query.and(eq(ZipkinServiceRelationTraffic.SERVICE_NAME, serviceName));
}
query.setLimit(QUERY_MAX_SIZE);
}
}
);
final List<String> remoteServices = new ArrayList<>();
for (final DataPoint dataPoint : resp.getDataPoints()) {
remoteServices.add(dataPoint.getTagValue(ZipkinServiceRelationTraffic.REMOTE_SERVICE_NAME));
}
return remoteServices;
}
@Override
public List<String> getSpanNames(final String serviceName) throws IOException {
MeasureQueryResponse resp =
query(ZipkinServiceSpanTraffic.INDEX_NAME,
SPAN_TRAFFIC_TAGS,
Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
@Override
protected void apply(MeasureQuery query) {
if (StringUtil.isNotEmpty(serviceName)) {
query.and(eq(ZipkinServiceSpanTraffic.SERVICE_NAME, serviceName));
}
query.setLimit(QUERY_MAX_SIZE);
}
}
);
final List<String> spanNames = new ArrayList<>();
for (final DataPoint dataPoint : resp.getDataPoints()) {
spanNames.add(dataPoint.getTagValue(ZipkinServiceSpanTraffic.SPAN_NAME));
}
return spanNames;
}
@Override
public List<Span> getTrace(final String traceId) throws IOException {
StreamQueryResponse resp =
query(ZipkinSpanRecord.INDEX_NAME, TRACE_TAGS,
new QueryBuilder<StreamQuery>() {
@Override
protected void apply(StreamQuery query) {
query.and(eq(ZipkinSpanRecord.TRACE_ID, traceId));
query.setLimit(QUERY_MAX_SIZE);
}
}
);
List<Span> trace = new ArrayList<>(resp.getElements().size());
for (final RowEntity rowEntity : resp.getElements()) {
ZipkinSpanRecord spanRecord = new ZipkinSpanRecord.Builder().storage2Entity(
new BanyanDBConverter.StorageToStream(ZipkinSpanRecord.INDEX_NAME, rowEntity));
trace.add(ZipkinSpanRecord.buildSpanFromRecord(spanRecord));
}
return trace;
}
@Override
public List<List<Span>> getTraces(final QueryRequest request, Duration duration) throws IOException {
final int tracesLimit = request.limit();
int scrollLimit = 1000;
int scrollFrom = 0;
Set<String> traceIds = new HashSet<>();
while (traceIds.size() < tracesLimit) {
Set<String> resp = getTraceIds(request, duration, scrollFrom, scrollLimit);
if (resp.size() == 0) {
break;
}
for (String traceId : resp) {
traceIds.add(traceId);
if (traceIds.size() >= tracesLimit) {
break;
}
}
scrollFrom = scrollFrom + scrollLimit;
}
return getTraces(traceIds);
}
private Set<String> getTraceIds(final QueryRequest request,
Duration duration,
int from,
int limit) throws IOException {
final long startTimeMillis = duration.getStartTimestamp();
final long endTimeMillis = duration.getEndTimestamp();
TimestampRange tsRange = null;
if (startTimeMillis > 0 && endTimeMillis > 0) {
tsRange = new TimestampRange(startTimeMillis, endTimeMillis);
}
final QueryBuilder<StreamQuery> queryBuilder = new QueryBuilder<StreamQuery>() {
@Override
public void apply(final StreamQuery query) {
if (!StringUtil.isEmpty(request.serviceName())) {
query.and(eq(ZipkinSpanRecord.LOCAL_ENDPOINT_SERVICE_NAME, request.serviceName()));
}
if (!StringUtil.isEmpty(request.remoteServiceName())) {
query.and(eq(ZipkinSpanRecord.REMOTE_ENDPOINT_SERVICE_NAME, request.remoteServiceName()));
}
if (!StringUtil.isEmpty(request.spanName())) {
query.and(eq(ZipkinSpanRecord.NAME, request.spanName()));
}
if (!CollectionUtils.isEmpty(request.annotationQuery())) {
for (Map.Entry<String, String> annotation : request.annotationQuery().entrySet()) {
if (annotation.getValue().isEmpty()) {
query.and(eq(ZipkinSpanRecord.QUERY, annotation.getKey()));
} else {
query.and(eq(ZipkinSpanRecord.QUERY, annotation.getKey() + "=" + annotation.getValue()));
}
}
}
if (request.minDuration() != null) {
query.and(gte(ZipkinSpanRecord.DURATION, request.minDuration()));
}
if (request.maxDuration() != null) {
query.and(lte(ZipkinSpanRecord.DURATION, request.maxDuration()));
}
query.setOrderBy(new StreamQuery.OrderBy(ZipkinSpanRecord.TIMESTAMP_MILLIS, AbstractQuery.Sort.DESC));
query.setLimit(limit);
query.setOffset(from);
}
};
StreamQueryResponse resp = query(ZipkinSpanRecord.INDEX_NAME, TRACE_TAGS, tsRange, queryBuilder);
Set<String> traceIds = new LinkedHashSet<>(); //needs to keep order here
for (final RowEntity rowEntity : resp.getElements()) {
traceIds.add(rowEntity.getTagValue(ZipkinSpanRecord.TRACE_ID));
}
return traceIds;
}
@Override
public List<List<Span>> getTraces(final Set<String> traceIds) throws IOException {
if (CollectionUtils.isEmpty(traceIds)) {
return Collections.EMPTY_LIST;
}
List<AbstractCriteria> conditions = new ArrayList<>(traceIds.size());
StreamQueryResponse resp =
query(ZipkinSpanRecord.INDEX_NAME, TRACE_TAGS,
new QueryBuilder<StreamQuery>() {
@Override
protected void apply(StreamQuery query) {
for (String traceId : traceIds) {
conditions.add(eq(ZipkinSpanRecord.TRACE_ID, traceId));
}
if (conditions.size() == 1) {
query.criteria(conditions.get(0));
} else if (conditions.size() > 1) {
query.criteria(or(conditions));
}
query.setOrderBy(
new StreamQuery.OrderBy(ZipkinSpanRecord.TIMESTAMP_MILLIS, AbstractQuery.Sort.DESC));
query.setLimit(QUERY_MAX_SIZE);
}
}
);
return buildTraces(resp);
}
private List<List<Span>> buildTraces(StreamQueryResponse resp) {
Map<String, List<Span>> groupedByTraceId = new LinkedHashMap<String, List<Span>>();
for (final RowEntity rowEntity : resp.getElements()) {
ZipkinSpanRecord spanRecord = new ZipkinSpanRecord.Builder().storage2Entity(
new BanyanDBConverter.StorageToStream(ZipkinSpanRecord.INDEX_NAME, rowEntity));
Span span = ZipkinSpanRecord.buildSpanFromRecord(spanRecord);
String traceId = span.traceId();
groupedByTraceId.putIfAbsent(traceId, new ArrayList<>());
groupedByTraceId.get(traceId).add(span);
}
return new ArrayList<>(groupedByTraceId.values());
}
}

67
oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBZipkinQueryDAO.java
View File

@ -1,67 +0,0 @@
/*
* 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.banyandb.stream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import org.apache.skywalking.oap.server.core.query.input.Duration;
import org.apache.skywalking.oap.server.core.storage.query.IZipkinQueryDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
import zipkin2.Span;
import zipkin2.storage.QueryRequest;
//TODO: Not support BanyanDB for query yet.
public class BanyanDBZipkinQueryDAO extends AbstractBanyanDBDAO implements IZipkinQueryDAO {
public BanyanDBZipkinQueryDAO(BanyanDBStorageClient client) {
super(client);
}
@Override
public List<String> getServiceNames() throws IOException {
return new ArrayList<>();
}
@Override
public List<String> getRemoteServiceNames(final String serviceName) throws IOException {
return new ArrayList<>();
}
@Override
public List<String> getSpanNames(final String serviceName) throws IOException {
return new ArrayList<>();
}
@Override
public List<Span> getTrace(final String traceId) throws IOException {
return new ArrayList<>();
}
@Override
public List<List<Span>> getTraces(final QueryRequest request, Duration duration) throws IOException {
return new ArrayList<>();
}
@Override
public List<List<Span>> getTraces(final Set<String> traceIds) throws IOException {
return new ArrayList<>();
}
}

58
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java
View File

@ -18,8 +18,6 @@
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.zipkin;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
@ -31,6 +29,7 @@ import org.apache.skywalking.library.elasticsearch.requests.search.Query;
import org.apache.skywalking.library.elasticsearch.requests.search.Search;
import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
import org.apache.skywalking.library.elasticsearch.requests.search.SearchParams;
import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.Aggregation;
import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.BucketOrder;
import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.TermsAggregationBuilder;
@ -50,7 +49,6 @@ import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.Elasti
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeRangeIndexNameGenerator;
import zipkin2.Endpoint;
import zipkin2.Span;
import zipkin2.storage.QueryRequest;
@ -153,7 +151,7 @@ public class ZipkinQueryEsDAO extends EsDAO implements IZipkinQueryDAO {
Map<String, Object> sourceAsMap = searchHit.getSource();
ZipkinSpanRecord record = new ZipkinSpanRecord.Builder().storage2Entity(
new ElasticSearchConverter.ToEntity(ZipkinSpanRecord.INDEX_NAME, sourceAsMap));
trace.add(buildSpanFromRecord(record));
trace.add(ZipkinSpanRecord.buildSpanFromRecord(record));
}
if (response.getHits().getHits().size() < SCROLLING_BATCH_SIZE) {
break;
@ -236,7 +234,8 @@ public class ZipkinQueryEsDAO extends EsDAO implements IZipkinQueryDAO {
public List<List<Span>> getTraces(final Set<String> traceIds) {
String index = IndexController.LogicIndicesRegister.getPhysicalTableName(ZipkinSpanRecord.INDEX_NAME);
BoolQueryBuilder query = Query.bool().must(Query.terms(ZipkinSpanRecord.TRACE_ID, new ArrayList<>(traceIds)));
SearchBuilder search = Search.builder().query(query).size(SCROLLING_BATCH_SIZE); //max span size for 1 scroll
SearchBuilder search = Search.builder().query(query).sort(ZipkinSpanRecord.TIMESTAMP_MILLIS, Sort.Order.DESC)
.size(SCROLLING_BATCH_SIZE); //max span size for 1 scroll
final SearchParams params = new SearchParams().scroll(SCROLL_CONTEXT_RETENTION);
SearchResponse response = getClient().search(index, search.build(), params);
@ -263,57 +262,10 @@ public class ZipkinQueryEsDAO extends EsDAO implements IZipkinQueryDAO {
Map<String, Object> sourceAsMap = searchHit.getSource();
ZipkinSpanRecord record = new ZipkinSpanRecord.Builder().storage2Entity(
new ElasticSearchConverter.ToEntity(ZipkinSpanRecord.INDEX_NAME, sourceAsMap));
Span span = buildSpanFromRecord(record);
Span span = ZipkinSpanRecord.buildSpanFromRecord(record);
String traceId = span.traceId();
groupedByTraceId.putIfAbsent(traceId, new ArrayList<>());
groupedByTraceId.get(traceId).add(span);
}
}
private Span buildSpanFromRecord(ZipkinSpanRecord record) {
Span.Builder span = Span.newBuilder();
span.traceId(record.getTraceId());
span.id(record.getSpanId());
span.parentId(record.getParentId());
span.kind(Span.Kind.valueOf(record.getKind()));
span.timestamp(record.getTimestamp());
span.duration(record.getDuration());
span.name(record.getName());
//Build localEndpoint
Endpoint.Builder localEndpoint = Endpoint.newBuilder();
localEndpoint.serviceName(record.getLocalEndpointServiceName());
if (!StringUtil.isEmpty(record.getLocalEndpointIPV4())) {
localEndpoint.parseIp(record.getLocalEndpointIPV4());
} else {
localEndpoint.parseIp(record.getLocalEndpointIPV6());
}
localEndpoint.port(record.getLocalEndpointPort());
span.localEndpoint(localEndpoint.build());
//Build remoteEndpoint
Endpoint.Builder remoteEndpoint = Endpoint.newBuilder();
remoteEndpoint.serviceName(record.getRemoteEndpointServiceName());
if (!StringUtil.isEmpty(record.getLocalEndpointIPV4())) {
remoteEndpoint.parseIp(record.getRemoteEndpointIPV4());
} else {
remoteEndpoint.parseIp(record.getRemoteEndpointIPV6());
}
remoteEndpoint.port(record.getRemoteEndpointPort());
span.remoteEndpoint(remoteEndpoint.build());
//Build tags
JsonObject tagsJson = record.getTags();
if (tagsJson != null) {
for (Map.Entry<String, JsonElement> tag : tagsJson.entrySet()) {
span.putTag(tag.getKey(), tag.getValue().getAsString());
}
}
//Build annotation
JsonObject annotationJson = record.getAnnotations();
if (annotationJson != null) {
for (Map.Entry<String, JsonElement> annotation : annotationJson.entrySet()) {
span.addAnnotation(Long.parseLong(annotation.getKey()), annotation.getValue().getAsString());
}
}
return span.build();
}
}

6
oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCZipkinQueryDAO.java
View File

@ -143,7 +143,7 @@ public class JDBCZipkinQueryDAO implements IZipkinQueryDAO {
List<Object> condition = new ArrayList<>(5);
List<Map.Entry<String, String>> annotations = new ArrayList<>(request.annotationQuery().entrySet());
sql.append("select ").append(ZipkinSpanRecord.INDEX_NAME).append(".").append(ZipkinSpanRecord.TRACE_ID).append(", ")
.append("max(").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(")").append(" from ");
.append("min(").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(")").append(" from ");
sql.append(ZipkinSpanRecord.INDEX_NAME);
/**
* This is an AdditionalEntity feature, see:
@ -208,7 +208,7 @@ public class JDBCZipkinQueryDAO implements IZipkinQueryDAO {
}
}
sql.append(" group by ").append(ZipkinSpanRecord.TRACE_ID);
sql.append(" order by max(").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(") desc");
sql.append(" order by min(").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(") desc");
sql.append(" limit ").append(request.limit());
Set<String> traceIds = new HashSet<>();
try (Connection connection = h2Client.getConnection()) {
@ -244,6 +244,8 @@ public class JDBCZipkinQueryDAO implements IZipkinQueryDAO {
i++;
}
sql.append(" order by ").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(" desc");
try (Connection connection = h2Client.getConnection()) {
ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]));
return buildTraces(resultSet);

66
test/e2e-v2/cases/zipkin/banyandb/docker-compose.yml Normal file
View File

@ -0,0 +1,66 @@
# 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.
version: '2.1'
services:
banyandb:
extends:
file: ../../../script/docker-compose/base-compose.yml
service: banyandb
networks:
- e2e
oap:
extends:
file: ../../../script/docker-compose/base-compose.yml
service: oap
environment:
SW_STORAGE: banyandb
SW_QUERY_ZIPKIN: default
SW_RECEIVER_ZIPKIN: default
expose:
- 9411
ports:
- 9412
entrypoint: ['sh', '-c', '/download-mysql.sh /skywalking/oap-libs && /skywalking/docker-entrypoint.sh']
networks:
- e2e
depends_on:
banyandb:
condition: service_healthy
frontend:
extends:
file: ../docker-compose-brave.yml
service: frontend
depends_on:
backend:
condition: service_healthy
oap:
condition: service_healthy
ports:
- 8081
backend:
extends:
file: ../docker-compose-brave.yml
service: backend
depends_on:
oap:
condition: service_healthy
networks:
e2e:

47
test/e2e-v2/cases/zipkin/banyandb/e2e.yaml Normal file
View File

@ -0,0 +1,47 @@
# 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.
# This file is used to show how to write configuration files and can be used to test.
setup:
env: compose
file: docker-compose.yml
timeout: 20m
init-system-environment: ../../../script/env
steps:
- name: set PATH
command: export PATH=/tmp/skywalking-infra-e2e/bin:$PATH
- name: install yq
command: bash test/e2e-v2/script/prepare/setup-e2e-shell/install.sh yq
- name: install swctl
command: bash test/e2e-v2/script/prepare/setup-e2e-shell/install.sh swctl
trigger:
action: http
interval: 3s
times: 10
url: http://${frontend_host}:${frontend_8081}
method: POST
verify:
# verify with retry strategy
retry:
# max retry count
count: 20
# the interval between two retries, in millisecond.
interval: 10s
cases:
- includes:
- ../zipkin-cases.yaml