You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2023/01/04 02:41:27 UTC
[skywalking] branch banyandb updated: Remove measure id and introduce Element
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch banyandb
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/banyandb by this push:
new 8ea8dbc0c8 Remove measure id and introduce Element
8ea8dbc0c8 is described below
commit 8ea8dbc0c86cdf78879c5c3e4d6c5f974df72ca7
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Jan 4 10:40:48 2023 +0800
Remove measure id and introduce Element
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
.../server/core/storage/model/ModelColumnTest.java | 10 +-
.../BanyanDBEBPFProfilingScheduleQueryDAO.java | 2 +-
.../banyandb/measure/BanyanDBMetadataQueryDAO.java | 112 +++++++++++----------
.../banyandb/measure/BanyanDBMetricsQueryDAO.java | 90 ++++++++---------
.../stream/BanyanDBProfileTaskLogQueryDAO.java | 8 +-
.../BanyanDBProfileThreadSnapshotQueryDAO.java | 3 +-
.../banyandb/stream/BanyanDBTraceQueryDAO.java | 3 +-
7 files changed, 118 insertions(+), 110 deletions(-)
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
index 51c6cb9e62..5cd16665fd 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
@@ -32,7 +32,7 @@ public class ModelColumnTest {
new SQLDatabaseExtension(),
new ElasticSearchExtension(
ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc", false),
- new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED, false)
);
Assert.assertEquals(true, column.isStorageOnly());
Assert.assertEquals("abc", column.getColumnName().getName());
@@ -41,7 +41,7 @@ public class ModelColumnTest {
false, false, true, 200,
new SQLDatabaseExtension(),
new ElasticSearchExtension(ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc", false),
- new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED, false)
);
Assert.assertEquals(true, column.isStorageOnly());
Assert.assertEquals("abc", column.getColumnName().getName());
@@ -51,7 +51,7 @@ public class ModelColumnTest {
false, false, true, 200,
new SQLDatabaseExtension(),
new ElasticSearchExtension(ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc", false),
- new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED, false)
);
Assert.assertEquals(false, column.isStorageOnly());
Assert.assertEquals("abc", column.getColumnName().getName());
@@ -64,7 +64,7 @@ public class ModelColumnTest {
new SQLDatabaseExtension(),
new ElasticSearchExtension(
ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc", false),
- new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED, false)
);
}
@@ -75,7 +75,7 @@ public class ModelColumnTest {
new SQLDatabaseExtension(),
new ElasticSearchExtension(
ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc", false),
- new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED, false)
);
}
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEBPFProfilingScheduleQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEBPFProfilingScheduleQueryDAO.java
index 8e3b41741a..f83567adff 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEBPFProfilingScheduleQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEBPFProfilingScheduleQueryDAO.java
@@ -62,7 +62,7 @@ public class BanyanDBEBPFProfilingScheduleQueryDAO extends AbstractBanyanDBDAO i
private EBPFProfilingSchedule buildEBPFProfilingSchedule(DataPoint dataPoint) {
final EBPFProfilingSchedule schedule = new EBPFProfilingSchedule();
- schedule.setScheduleId(dataPoint.getId());
+ schedule.setScheduleId(dataPoint.getTagValue(EBPFProfilingScheduleRecord.EBPF_PROFILING_SCHEDULE_ID));
schedule.setTaskId(dataPoint.getTagValue(EBPFProfilingScheduleRecord.TASK_ID));
schedule.setProcessId(dataPoint.getTagValue(EBPFProfilingScheduleRecord.PROCESS_ID));
schedule.setStartTime(((Number) dataPoint.getTagValue(EBPFProfilingScheduleRecord.START_TIME)).longValue());
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
index d10b92a22f..5d5c18fc55 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
@@ -18,14 +18,15 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.measure;
-import com.google.common.base.Strings;
import com.google.common.collect.ImmutableSet;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
+import org.apache.commons.lang3.StringUtils;
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.oap.server.core.analysis.DownSampling;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
@@ -44,7 +45,9 @@ import org.apache.skywalking.oap.server.core.query.type.Service;
import org.apache.skywalking.oap.server.core.query.type.ServiceInstance;
import org.apache.skywalking.oap.server.core.storage.query.IMetadataQueryDAO;
import org.apache.skywalking.oap.server.library.util.StringUtil;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBConverter;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
import java.io.IOException;
@@ -103,9 +106,10 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
});
final List<Service> services = new ArrayList<>();
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(ServiceTraffic.INDEX_NAME, DownSampling.Minute);
for (final DataPoint dataPoint : resp.getDataPoints()) {
- services.add(buildService(dataPoint));
+ services.add(buildService(dataPoint, schema));
}
return services;
@@ -125,9 +129,10 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
});
final List<Service> services = new ArrayList<>();
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(ServiceTraffic.INDEX_NAME, DownSampling.Minute);
for (final DataPoint dataPoint : resp.getDataPoints()) {
- services.add(buildService(dataPoint));
+ services.add(buildService(dataPoint, schema));
}
return services;
@@ -150,9 +155,9 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
});
final List<ServiceInstance> instances = new ArrayList<>();
-
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(InstanceTraffic.INDEX_NAME, DownSampling.Minute);
for (final DataPoint dataPoint : resp.getDataPoints()) {
- instances.add(buildInstance(dataPoint));
+ instances.add(buildInstance(dataPoint, schema));
}
return instances;
@@ -171,8 +176,8 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
}
}
});
-
- return resp.size() > 0 ? buildInstance(resp.getDataPoints().get(0)) : null;
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(InstanceTraffic.INDEX_NAME, DownSampling.Minute);
+ return resp.size() > 0 ? buildInstance(resp.getDataPoints().get(0), schema) : null;
}
@Override
@@ -190,9 +195,9 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
});
final List<Endpoint> endpoints = new ArrayList<>();
-
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(EndpointTraffic.INDEX_NAME, DownSampling.Minute);
for (final DataPoint dataPoint : resp.getDataPoints()) {
- endpoints.add(buildEndpoint(dataPoint));
+ endpoints.add(buildEndpoint(dataPoint, schema));
}
if (StringUtil.isNotEmpty(serviceId)) {
@@ -217,9 +222,9 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
});
final List<Process> processes = new ArrayList<>();
-
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(ProcessTraffic.INDEX_NAME, DownSampling.Minute);
for (final DataPoint dataPoint : resp.getDataPoints()) {
- processes.add(buildProcess(dataPoint));
+ processes.add(buildProcess(dataPoint, schema));
}
return processes;
@@ -243,9 +248,9 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
});
final List<Process> processes = new ArrayList<>();
-
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(ProcessTraffic.INDEX_NAME, DownSampling.Minute);
for (final DataPoint dataPoint : resp.getDataPoints()) {
- processes.add(buildProcess(dataPoint));
+ processes.add(buildProcess(dataPoint, schema));
}
return processes;
@@ -265,9 +270,9 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
});
final List<Process> processes = new ArrayList<>();
-
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(ProcessTraffic.INDEX_NAME, DownSampling.Minute);
for (final DataPoint dataPoint : resp.getDataPoints()) {
- processes.add(buildProcess(dataPoint));
+ processes.add(buildProcess(dataPoint, schema));
}
return processes;
@@ -326,31 +331,34 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
}
}
});
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(ProcessTraffic.INDEX_NAME, DownSampling.Minute);
- return resp.size() > 0 ? buildProcess(resp.getDataPoints().get(0)) : null;
+ return resp.size() > 0 ? buildProcess(resp.getDataPoints().get(0), schema) : null;
}
- private Service buildService(DataPoint dataPoint) {
+ private Service buildService(DataPoint dataPoint, MetadataRegistry.Schema schema) {
+ final ServiceTraffic.Builder builder = new ServiceTraffic.Builder();
+ final ServiceTraffic serviceTraffic = builder.storage2Entity(new BanyanDBConverter.StorageToMeasure(schema, dataPoint));
+ String serviceName = serviceTraffic.getName();
Service service = new Service();
- service.setId(dataPoint.getTagValue(ServiceTraffic.SERVICE_ID));
- service.setName(dataPoint.getTagValue(ServiceTraffic.NAME));
- service.setShortName(dataPoint.getTagValue(ServiceTraffic.SHORT_NAME));
- service.setGroup(dataPoint.getTagValue(ServiceTraffic.GROUP));
- service.getLayers().add(Layer.valueOf(((Number) dataPoint.getTagValue(ServiceTraffic.LAYER)).intValue()).name());
+ service.setId(serviceTraffic.getServiceId());
+ service.setName(serviceName);
+ service.setShortName(serviceTraffic.getShortName());
+ service.setGroup(serviceTraffic.getGroup());
+ service.getLayers().add(serviceTraffic.getLayer().name());
return service;
}
- private ServiceInstance buildInstance(DataPoint dataPoint) {
+ private ServiceInstance buildInstance(DataPoint dataPoint, MetadataRegistry.Schema schema) {
+ final InstanceTraffic instanceTraffic =
+ new InstanceTraffic.Builder().storage2Entity(new BanyanDBConverter.StorageToMeasure(schema, dataPoint));
+
ServiceInstance serviceInstance = new ServiceInstance();
- serviceInstance.setId(dataPoint.getId());
- serviceInstance.setName(dataPoint.getTagValue(InstanceTraffic.NAME));
- serviceInstance.setInstanceUUID(dataPoint.getId());
-
- final String propString = dataPoint.getTagValue(InstanceTraffic.PROPERTIES);
- JsonObject properties = null;
- if (StringUtil.isNotEmpty(propString)) {
- properties = GSON.fromJson(propString, JsonObject.class);
- }
+ serviceInstance.setId(instanceTraffic.id().build());
+ serviceInstance.setName(instanceTraffic.getName());
+ serviceInstance.setInstanceUUID(serviceInstance.getId());
+
+ JsonObject properties = instanceTraffic.getProperties();
if (properties != null) {
for (Map.Entry<String, JsonElement> property : properties.entrySet()) {
String key = property.getKey();
@@ -364,44 +372,46 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
} else {
serviceInstance.setLanguage(Language.UNKNOWN);
}
-
return serviceInstance;
}
- private Endpoint buildEndpoint(DataPoint dataPoint) {
+ private Endpoint buildEndpoint(DataPoint dataPoint, MetadataRegistry.Schema schema) {
+ final EndpointTraffic endpointTraffic =
+ new EndpointTraffic.Builder().storage2Entity(new BanyanDBConverter.StorageToMeasure(schema, dataPoint));
Endpoint endpoint = new Endpoint();
- endpoint.setId(dataPoint.getId());
- endpoint.setName(dataPoint.getTagValue(EndpointTraffic.NAME));
+ endpoint.setId(endpointTraffic.id().build());
+ endpoint.setName(endpointTraffic.getName());
return endpoint;
}
- private Process buildProcess(DataPoint dataPoint) {
- Process process = new Process();
+ private Process buildProcess(DataPoint dataPoint, MetadataRegistry.Schema schema) {
+ final ProcessTraffic processTraffic =
+ new ProcessTraffic.Builder().storage2Entity(new BanyanDBConverter.StorageToMeasure(schema, dataPoint));
- process.setId(dataPoint.getId());
- process.setName(dataPoint.getTagValue(ProcessTraffic.NAME));
- String serviceId = dataPoint.getTagValue(ProcessTraffic.SERVICE_ID);
+ Process process = new Process();
+ process.setId(processTraffic.id().build());
+ process.setName(processTraffic.getName());
+ final String serviceId = processTraffic.getServiceId();
process.setServiceId(serviceId);
process.setServiceName(IDManager.ServiceID.analysisId(serviceId).getName());
- String instanceId = dataPoint.getTagValue(ProcessTraffic.INSTANCE_ID);
+ final String instanceId = processTraffic.getInstanceId();
process.setInstanceId(instanceId);
process.setInstanceName(IDManager.ServiceInstanceID.analysisId(instanceId).getName());
- process.setAgentId(dataPoint.getTagValue(ProcessTraffic.AGENT_ID));
- process.setDetectType(ProcessDetectType.valueOf(((Number) dataPoint.getTagValue(ProcessTraffic.DETECT_TYPE)).intValue()).name());
- process.setProfilingSupportStatus(ProfilingSupportStatus.valueOf(((Number) dataPoint.getTagValue(ProcessTraffic.PROFILING_SUPPORT_STATUS)).intValue()).name());
+ process.setAgentId(processTraffic.getAgentId());
+ process.setDetectType(ProcessDetectType.valueOf(processTraffic.getDetectType()).name());
+ process.setProfilingSupportStatus(ProfilingSupportStatus.valueOf(processTraffic.getProfilingSupportStatus()).name());
- String propString = dataPoint.getTagValue(ProcessTraffic.PROPERTIES);
- if (!Strings.isNullOrEmpty(propString)) {
- JsonObject properties = GSON.fromJson(propString, JsonObject.class);
+ JsonObject properties = processTraffic.getProperties();
+ if (properties != null) {
for (Map.Entry<String, JsonElement> property : properties.entrySet()) {
String key = property.getKey();
String value = property.getValue().getAsString();
process.getAttributes().add(new Attribute(key, value));
}
}
- String labelJson = dataPoint.getTagValue(ProcessTraffic.LABELS_JSON);
- if (!Strings.isNullOrEmpty(labelJson)) {
- List<String> labels = GSON.<List<String>>fromJson(labelJson, ArrayList.class);
+ final String labelsJson = processTraffic.getLabelsJson();
+ if (StringUtils.isNotEmpty(labelsJson)) {
+ final List<String> labels = GSON.<List<String>>fromJson(labelsJson, ArrayList.class);
process.getLabels().addAll(labels);
}
return process;
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
index 6741c517ff..047f89a9c9 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
@@ -32,6 +32,7 @@ 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.TimestampRange;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
import org.apache.skywalking.oap.server.core.analysis.metrics.HistogramMetrics;
import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
@@ -113,33 +114,27 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
}
final String entityID = condition.getEntity().buildId();
- Map<String, DataPoint> idMap = queryByEntityID(schema, valueColumnName, duration, entityID);
+ Map<Long, DataPoint> idMap = queryByEntityID(schema, valueColumnName, duration, entityID);
- List<String> ids = extractMeasureIDs(duration, entityID);
+ List<PointOfTime> tsPoints = duration.assembleDurationPoints();
MetricsValues metricsValues = new MetricsValues();
- if (!idMap.isEmpty()) {
- // Label is null, because in readMetricsValues, no label parameter.
- IntValues intValues = metricsValues.getValues();
- for (String id : ids) {
- KVInt kvInt = new KVInt();
- kvInt.setId(id);
- kvInt.setValue(0);
- if (idMap.containsKey(id)) {
- DataPoint dataPoint = idMap.get(id);
- kvInt.setValue(extractFieldValue(schema, valueColumnName, dataPoint));
- } else {
- kvInt.setValue(ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()));
- }
- intValues.addKVInt(kvInt);
+ // Label is null, because in readMetricsValues, no label parameter.
+ IntValues intValues = metricsValues.getValues();
+ for (PointOfTime ts : tsPoints) {
+ String id = ts.id(entityID);
+ KVInt kvInt = new KVInt();
+ kvInt.setId(id);
+ kvInt.setValue(0);
+ if (idMap.containsKey(ts.getPoint())) {
+ DataPoint dataPoint = idMap.get(ts.getPoint());
+ kvInt.setValue(extractFieldValue(schema, valueColumnName, dataPoint));
+ } else {
+ kvInt.setValue(ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()));
}
+ intValues.addKVInt(kvInt);
}
- metricsValues.setValues(
- Util.sortValues(
- metricsValues.getValues(), ids, ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()))
- );
-
return metricsValues;
}
@@ -157,16 +152,22 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
@Override
public List<MetricsValues> readLabeledMetricsValues(MetricsCondition condition, String valueColumnName, List<String> labels, Duration duration) throws IOException {
- Map<String, DataPoint> idMap = queryByEntityID(condition, valueColumnName, duration);
+ Map<Long, DataPoint> idMap = queryByEntityID(condition, valueColumnName, duration);
- List<String> ids = extractMeasureIDs(duration, condition.getEntity().buildId());
+ List<PointOfTime> tsPoints = duration.assembleDurationPoints();
+ String entityID = condition.getEntity().buildId();
+ List<String> ids = new ArrayList<>(tsPoints.size());
Map<String, DataTable> dataTableMap = new HashMap<>(idMap.size());
- for (final Map.Entry<String, DataPoint> entry : idMap.entrySet()) {
- dataTableMap.put(
- entry.getKey(),
- new DataTable(entry.getValue().getFieldValue(valueColumnName))
- );
+ for (PointOfTime ts : tsPoints) {
+ String id = ts.id(entityID);
+ ids.add(id);
+ if (idMap.containsKey(ts.getPoint())) {
+ dataTableMap.put(
+ id,
+ new DataTable(idMap.get(ts.getPoint()).getFieldValue(valueColumnName))
+ );
+ }
}
return Util.sortValues(
@@ -178,18 +179,22 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
@Override
public HeatMap readHeatMap(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
- Map<String, DataPoint> idMap = queryByEntityID(condition, valueColumnName, duration);
+ Map<Long, DataPoint> idMap = queryByEntityID(condition, valueColumnName, duration);
HeatMap heatMap = new HeatMap();
if (idMap.isEmpty()) {
return heatMap;
}
- List<String> ids = extractMeasureIDs(duration, condition.getEntity().buildId());
+ List<PointOfTime> tsPoints = duration.assembleDurationPoints();
+ String entityID = condition.getEntity().buildId();
+ List<String> ids = new ArrayList<>(tsPoints.size());
final int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName());
- for (String id : ids) {
- DataPoint dataPoint = idMap.get(id);
+ for (PointOfTime ts : tsPoints) {
+ String id = ts.id(entityID);
+ ids.add(id);
+ DataPoint dataPoint = idMap.get(ts.getPoint());
if (dataPoint != null) {
String value = dataPoint.getFieldValue(HistogramMetrics.DATASET);
heatMap.buildColumn(id, value, defaultValue);
@@ -201,17 +206,7 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
return heatMap;
}
- private List<String> extractMeasureIDs(Duration duration, String entityID) {
- final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
- List<String> ids = new ArrayList<>(pointOfTimes.size());
- pointOfTimes.forEach(pointOfTime -> {
- String id = pointOfTime.id(entityID);
- ids.add(id);
- });
- return ids;
- }
-
- private Map<String, DataPoint> queryByEntityID(final MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
+ private Map<Long, DataPoint> queryByEntityID(final MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
final MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(condition.getName(), duration.getStep());
if (schema == null) {
throw new IOException("schema is not registered");
@@ -219,10 +214,10 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
return queryByEntityID(schema, valueColumnName, duration, condition.getEntity().buildId());
}
- private Map<String, DataPoint> queryByEntityID(MetadataRegistry.Schema schema, String valueColumnName, Duration duration, String entityID) throws IOException {
+ private Map<Long, DataPoint> queryByEntityID(MetadataRegistry.Schema schema, String valueColumnName, Duration duration, String entityID) throws IOException {
TimestampRange timestampRange = new TimestampRange(duration.getStartTimestamp(), duration.getEndTimestamp());
- Map<String, DataPoint> map = new HashMap<>();
+ Map<Long, DataPoint> map = new HashMap<>();
MeasureQueryResponse resp = query(schema, ImmutableSet.of(Metrics.ENTITY_ID), ImmutableSet.of(valueColumnName), timestampRange, new QueryBuilder<MeasureQuery>() {
@Override
protected void apply(MeasureQuery query) {
@@ -230,8 +225,9 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
}
});
for (final DataPoint dp : resp.getDataPoints()) {
- if (map.putIfAbsent(dp.getId(), dp) != null) {
- log.warn("duplicated data point");
+ long timeBucket = TimeBucket.getTimeBucket(dp.getTimestamp(), schema.getMetadata().getDownSampling());
+ if (map.putIfAbsent(timeBucket, dp) != null) {
+ log.warn("duplicated data point at " + timeBucket);
}
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskLogQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskLogQueryDAO.java
index 74038d87bc..709ce72009 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskLogQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskLogQueryDAO.java
@@ -19,7 +19,7 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
import com.google.common.collect.ImmutableSet;
-import org.apache.skywalking.banyandb.v1.client.RowEntity;
+import org.apache.skywalking.banyandb.v1.client.Element;
import org.apache.skywalking.banyandb.v1.client.StreamQuery;
import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
import org.apache.skywalking.oap.server.core.profiling.trace.ProfileTaskLogRecord;
@@ -59,14 +59,14 @@ public class BanyanDBProfileTaskLogQueryDAO extends AbstractBanyanDBDAO implemen
});
final LinkedList<ProfileTaskLog> tasks = new LinkedList<>();
- for (final RowEntity rowEntity : resp.getElements()) {
- tasks.add(buildProfileTaskLog(rowEntity));
+ for (final Element element : resp.getElements()) {
+ tasks.add(buildProfileTaskLog(element));
}
return tasks;
}
- private ProfileTaskLog buildProfileTaskLog(RowEntity data) {
+ private ProfileTaskLog buildProfileTaskLog(Element data) {
return ProfileTaskLog.builder()
.id(data.getId())
.taskId(data.getTagValue(ProfileTaskLogRecord.TASK_ID))
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
index fe4b815122..3072d6337f 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
@@ -19,6 +19,7 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
import com.google.common.collect.ImmutableSet;
+import org.apache.skywalking.banyandb.v1.client.Element;
import org.apache.skywalking.banyandb.v1.client.RowEntity;
import org.apache.skywalking.banyandb.v1.client.StreamQuery;
import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
@@ -120,7 +121,7 @@ public class BanyanDBProfileThreadSnapshotQueryDAO extends AbstractBanyanDBDAO i
});
List<BasicTrace> basicTraces = new ArrayList<>();
- for (final RowEntity row : segmentRecordResp.getElements()) {
+ for (final Element row : segmentRecordResp.getElements()) {
BasicTrace basicTrace = new BasicTrace();
basicTrace.setSegmentId(row.getId());
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java
index f6491cd43d..641204bc3d 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java
@@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
import com.google.common.collect.ImmutableSet;
import org.apache.skywalking.banyandb.v1.client.AbstractQuery;
+import org.apache.skywalking.banyandb.v1.client.Element;
import org.apache.skywalking.banyandb.v1.client.RowEntity;
import org.apache.skywalking.banyandb.v1.client.StreamQuery;
import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
@@ -152,7 +153,7 @@ public class BanyanDBTraceQueryDAO extends AbstractBanyanDBDAO implements ITrace
return traceBrief;
}
- for (final RowEntity row : resp.getElements()) {
+ for (final Element row : resp.getElements()) {
BasicTrace basicTrace = new BasicTrace();
basicTrace.setSegmentId(row.getId());