From 8d33f0275ed958ddb4d45a1846a2ae1bb0900ddb Mon Sep 17 00:00:00 2001 From: Daming Date: Thu, 17 Dec 2020 22:54:36 +0800 Subject: [PATCH] chore: code polish (#6025) --- .../storage/plugin/influxdb/InfluxClient.java | 10 +- .../influxdb/InfluxStorageProvider.java | 11 +- .../plugin/influxdb/InfluxTableInstaller.java | 5 +- .../plugin/influxdb/TableMetaInfo.java | 8 +- .../influxdb/base/InfluxInsertRequest.java | 10 +- .../influxdb/base/InfluxStorageDAO.java | 2 +- .../plugin/influxdb/base/ManagementDAO.java | 4 +- .../plugin/influxdb/base/MetricsDAO.java | 18 ++- .../plugin/influxdb/base/NoneStreamDAO.java | 11 +- .../plugin/influxdb/base/RecordDAO.java | 29 ++-- .../influxdb/query/AggregationQuery.java | 22 +-- .../plugin/influxdb/query/AlarmQuery.java | 4 +- .../plugin/influxdb/query/MetadataQuery.java | 24 ++-- .../plugin/influxdb/query/MetricsQuery.java | 54 +++----- .../query/NetworkAddressAliasDAO.java | 2 +- .../influxdb/query/ProfileTaskLogQuery.java | 22 ++- .../influxdb/query/ProfileTaskQuery.java | 14 +- .../query/ProfileThreadSnapshotQuery.java | 92 +++++++------ .../influxdb/query/TopNRecordsQuery.java | 14 +- .../plugin/influxdb/query/TopologyQuery.java | 125 +++++++++--------- .../plugin/influxdb/query/TraceQuery.java | 35 ++--- .../query/UITemplateManagementDAOImpl.java | 36 ++--- 22 files changed, 261 insertions(+), 291 deletions(-) diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java index a2be5695f..8506431d1 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java @@ -46,17 +46,13 @@ import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.ti; */ @Slf4j public class InfluxClient implements Client, HealthCheckable { - private InfluxStorageConfig config; + private final DelegatedHealthChecker healthChecker = new DelegatedHealthChecker(); + private final InfluxStorageConfig config; private InfluxDB influx; - private DelegatedHealthChecker healthChecker = new DelegatedHealthChecker(); /** * A constant, the name of time field in Time-series database. */ public static final String TIME = "time"; - /** - * A constant, the name of tag of time_bucket. - */ - public static final String TAG_TIME_BUCKET = "_time_bucket"; private final String database; @@ -217,7 +213,7 @@ public class InfluxClient implements Client, HealthCheckable { this.healthChecker.health(); } catch (Throwable e) { healthChecker.unHealth(e); - throw e; + throw new IOException(e); } } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageProvider.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageProvider.java index f062050bb..7b8d5ad45 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageProvider.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageProvider.java @@ -69,7 +69,7 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsTag; @Slf4j public class InfluxStorageProvider extends ModuleProvider { - private InfluxStorageConfig config; + private final InfluxStorageConfig config; private InfluxClient client; public InfluxStorageProvider() { @@ -123,8 +123,11 @@ public class InfluxStorageProvider extends ModuleProvider { @Override public void start() throws ServiceNotProvidedException, ModuleStartException { - MetricsCreator metricCreator = getManager().find(TelemetryModule.NAME).provider().getService(MetricsCreator.class); - HealthCheckMetrics healthChecker = metricCreator.createHealthCheckerGauge("storage_influxdb", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE); + MetricsCreator metricCreator = getManager().find(TelemetryModule.NAME) + .provider() + .getService(MetricsCreator.class); + HealthCheckMetrics healthChecker = metricCreator.createHealthCheckerGauge( + "storage_influxdb", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE); client.registerChecker(healthChecker); try { client.connect(); @@ -137,7 +140,7 @@ public class InfluxStorageProvider extends ModuleProvider { } @Override - public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException { + public void notifyAfterCompleted() throws ServiceNotProvidedException { } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxTableInstaller.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxTableInstaller.java index 0246128a2..8642592f0 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxTableInstaller.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxTableInstaller.java @@ -18,7 +18,6 @@ package org.apache.skywalking.oap.server.storage.plugin.influxdb; -import org.apache.skywalking.oap.server.core.storage.StorageException; import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller; import org.apache.skywalking.oap.server.library.client.Client; @@ -31,13 +30,13 @@ public class InfluxTableInstaller extends ModelInstaller { } @Override - protected boolean isExists(final Model model) throws StorageException { + protected boolean isExists(final Model model) { TableMetaInfo.addModel(model); return true; } @Override - protected void createTable(final Model model) throws StorageException { + protected void createTable(final Model model) { // Automatically create table } } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/TableMetaInfo.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/TableMetaInfo.java index ce4e805f4..698ac9705 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/TableMetaInfo.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/TableMetaInfo.java @@ -41,9 +41,9 @@ import org.apache.skywalking.oap.server.core.storage.model.ModelColumn; public class TableMetaInfo { private static final Map TABLES = new HashMap<>(); - private Map storageAndColumnMap; - private Map storageAndTagMap; - private Model model; + private final Map storageAndColumnMap; + private final Map storageAndTagMap; + private final Model model; public static void addModel(Model model) { final List columns = model.getColumns(); @@ -88,7 +88,7 @@ public class TableMetaInfo { } } - TableMetaInfo info = TableMetaInfo.builder() + final TableMetaInfo info = TableMetaInfo.builder() .model(model) .storageAndTagMap(storageAndTagMap) .storageAndColumnMap(storageAndColumnMap) diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxInsertRequest.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxInsertRequest.java index c510d2e81..9818e9ca7 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxInsertRequest.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxInsertRequest.java @@ -36,17 +36,17 @@ import org.influxdb.dto.Point; * InfluxDB Point wrapper. */ public class InfluxInsertRequest implements InsertRequest, UpdateRequest { - private Point.Builder builder; - private Map fields = Maps.newHashMap(); + private final Point.Builder builder; + private final Map fields = Maps.newHashMap(); - public InfluxInsertRequest(Model model, StorageData storageData, StorageBuilder storageBuilder) { - Map objectMap = storageBuilder.data2Map(storageData); + public InfluxInsertRequest(Model model, T storageData, StorageBuilder storageBuilder) { + final Map objectMap = storageBuilder.data2Map(storageData); if (SegmentRecord.INDEX_NAME.equals(model.getName())) { objectMap.remove(SegmentRecord.TAGS); } for (ModelColumn column : model.getColumns()) { - Object value = objectMap.get(column.getColumnName().getName()); + final Object value = objectMap.get(column.getColumnName().getName()); if (value instanceof StorageDataComplexObject) { fields.put( diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxStorageDAO.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxStorageDAO.java index 45553d09c..a51204343 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxStorageDAO.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxStorageDAO.java @@ -44,7 +44,7 @@ public class InfluxStorageDAO implements StorageDAO { @Override public IRecordDAO newRecordDao(StorageBuilder storageBuilder) { - return new RecordDAO(influxClient, storageBuilder); + return new RecordDAO(storageBuilder); } @Override diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/ManagementDAO.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/ManagementDAO.java index e90a38b7d..59f2a052a 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/ManagementDAO.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/ManagementDAO.java @@ -40,8 +40,8 @@ import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.select; @Slf4j public class ManagementDAO implements IManagementDAO { - private InfluxClient client; - private StorageBuilder storageBuilder; + private final InfluxClient client; + private final StorageBuilder storageBuilder; public ManagementDAO(InfluxClient client, StorageBuilder storageBuilder) { this.client = client; diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/MetricsDAO.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/MetricsDAO.java index 86f5c8c0d..13121e56f 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/MetricsDAO.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/MetricsDAO.java @@ -58,7 +58,7 @@ public class MetricsDAO implements IMetricsDAO { @Override public List multiGet(Model model, List ids) throws IOException { - WhereQueryImpl query = select() + final WhereQueryImpl query = select() .raw(ALL_FIELDS) .from(client.getDatabase(), model.getName()) .where(contains("id", Joiner.on("|").join(ids))); @@ -72,10 +72,10 @@ public class MetricsDAO implements IMetricsDAO { } final List metrics = Lists.newArrayList(); - List columns = series.getColumns(); + final List columns = series.getColumns(); - TableMetaInfo metaInfo = TableMetaInfo.get(model.getName()); - Map storageAndColumnMap = metaInfo.getStorageAndColumnMap(); + final TableMetaInfo metaInfo = TableMetaInfo.get(model.getName()); + final Map storageAndColumnMap = metaInfo.getStorageAndColumnMap(); series.getValues().forEach(values -> { Map data = Maps.newHashMap(); @@ -96,21 +96,19 @@ public class MetricsDAO implements IMetricsDAO { } @Override - public InsertRequest prepareBatchInsert(Model model, Metrics metrics) throws IOException { + public InsertRequest prepareBatchInsert(Model model, Metrics metrics) { final long timestamp = TimeBucket.getTimestamp(metrics.getTimeBucket(), model.getDownsampling()); - TableMetaInfo tableMetaInfo = TableMetaInfo.get(model.getName()); + final TableMetaInfo tableMetaInfo = TableMetaInfo.get(model.getName()); final InfluxInsertRequest request = new InfluxInsertRequest(model, metrics, storageBuilder) .time(timestamp, TimeUnit.MILLISECONDS); - tableMetaInfo.getStorageAndTagMap().forEach((field, tag) -> { - request.addFieldAsTag(field, tag); - }); + tableMetaInfo.getStorageAndTagMap().forEach(request::addFieldAsTag); return request; } @Override - public UpdateRequest prepareBatchUpdate(Model model, Metrics metrics) throws IOException { + public UpdateRequest prepareBatchUpdate(Model model, Metrics metrics) { return (UpdateRequest) this.prepareBatchInsert(model, metrics); } } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/NoneStreamDAO.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/NoneStreamDAO.java index 3d89e419f..b9f684c0c 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/NoneStreamDAO.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/NoneStreamDAO.java @@ -18,7 +18,6 @@ package org.apache.skywalking.oap.server.storage.plugin.influxdb.base; -import java.io.IOException; import java.util.concurrent.TimeUnit; import org.apache.skywalking.apm.commons.datacarrier.common.AtomicRangeInteger; import org.apache.skywalking.oap.server.core.analysis.TimeBucket; @@ -33,8 +32,8 @@ public class NoneStreamDAO implements INoneStreamDAO { private static final int PADDING_SIZE = 1_000_000; private static final AtomicRangeInteger SUFFIX = new AtomicRangeInteger(0, PADDING_SIZE); - private InfluxClient client; - private StorageBuilder storageBuilder; + private final InfluxClient client; + private final StorageBuilder storageBuilder; public NoneStreamDAO(InfluxClient client, StorageBuilder storageBuilder) { this.client = client; @@ -42,15 +41,13 @@ public class NoneStreamDAO implements INoneStreamDAO { } @Override - public void insert(final Model model, final NoneStream noneStream) throws IOException { + public void insert(final Model model, final NoneStream noneStream) { final long timestamp = TimeBucket.getTimestamp(noneStream.getTimeBucket(), model.getDownsampling()) * PADDING_SIZE + SUFFIX.getAndIncrement(); final InfluxInsertRequest request = new InfluxInsertRequest(model, noneStream, storageBuilder) .time(timestamp, TimeUnit.NANOSECONDS); - TableMetaInfo.get(model.getName()).getStorageAndTagMap().forEach((field, tag) -> { - request.addFieldAsTag(field, tag); - }); + TableMetaInfo.get(model.getName()).getStorageAndTagMap().forEach(request::addFieldAsTag); client.write(request.getPoint()); } } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/RecordDAO.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/RecordDAO.java index a3d1bc7e7..8ccfe874b 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/RecordDAO.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/RecordDAO.java @@ -19,7 +19,6 @@ package org.apache.skywalking.oap.server.storage.plugin.influxdb.base; import com.google.common.base.Joiner; -import java.io.IOException; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -33,44 +32,38 @@ import org.apache.skywalking.oap.server.core.storage.IRecordDAO; import org.apache.skywalking.oap.server.core.storage.StorageBuilder; import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.library.client.request.InsertRequest; -import org.apache.skywalking.oap.server.storage.plugin.influxdb.InfluxClient; import org.apache.skywalking.oap.server.storage.plugin.influxdb.TableMetaInfo; public class RecordDAO implements IRecordDAO { private static final int PADDING_SIZE = 1_000_000; private static final AtomicRangeInteger SUFFIX = new AtomicRangeInteger(0, PADDING_SIZE); - private InfluxClient client; - private StorageBuilder storageBuilder; + private final StorageBuilder storageBuilder; - public RecordDAO(InfluxClient client, StorageBuilder storageBuilder) { - this.client = client; + public RecordDAO(StorageBuilder storageBuilder) { this.storageBuilder = storageBuilder; } @Override - public InsertRequest prepareBatchInsert(Model model, Record record) throws IOException { + public InsertRequest prepareBatchInsert(Model model, Record record) { final long timestamp = TimeBucket.getTimestamp(record.getTimeBucket(), model.getDownsampling()) - * PADDING_SIZE + SUFFIX.getAndIncrement(); + * PADDING_SIZE + + SUFFIX.getAndIncrement(); final InfluxInsertRequest request = new InfluxInsertRequest(model, record, storageBuilder) .time(timestamp, TimeUnit.NANOSECONDS); - TableMetaInfo.get(model.getName()).getStorageAndTagMap().forEach((field, tag) -> { - request.addFieldAsTag(field, tag); - }); + + TableMetaInfo.get(model.getName()).getStorageAndTagMap().forEach(request::addFieldAsTag); if (SegmentRecord.INDEX_NAME.equals(model.getName())) { Map> collect = ((SegmentRecord) record).getTagsRawData() .stream() .collect( Collectors.groupingBy(SpanTag::getKey)); - collect.entrySet().forEach(e -> { - request.tag(e.getKey(), "'" + Joiner.on("'") - .join(e.getValue() - .stream() - .map(SpanTag::getValue) - .collect(Collectors.toSet())) + "'"); - }); + collect.forEach((key, value) -> request.tag( + key, + "'" + Joiner.on("'").join(value.stream().map(SpanTag::getValue).collect(Collectors.toSet())) + "'" + )); } return request; } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AggregationQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AggregationQuery.java index eba8c844d..5d04cb4f9 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AggregationQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AggregationQuery.java @@ -44,7 +44,11 @@ import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.select; @Slf4j public class AggregationQuery implements IAggregationQueryDAO { - private InfluxClient client; + private static final Comparator ASCENDING = + Comparator.comparingLong(a -> Long.parseLong(a.getValue())); + private static final Comparator DESCENDING = (a, b) -> + Long.compare(Long.parseLong(b.getValue()), Long.parseLong(a.getValue())); + private final InfluxClient client; public AggregationQuery(InfluxClient client) { this.client = client; @@ -72,11 +76,12 @@ public class AggregationQuery implements IAggregationQueryDAO { WhereSubQueryImpl, SelectQueryImpl> where = select() .fromSubQuery(client.getDatabase()) .mean(valueColumnName) - .from(condition.getName()).where(); + .from(condition.getName()) + .where(); if (additionalConditions != null) { - additionalConditions.forEach(moreCondition -> { - where.and(eq(moreCondition.getKey(), moreCondition.getValue())); - }); + additionalConditions.forEach(moreCondition -> + where.and(eq(moreCondition.getKey(), moreCondition.getValue())) + ); } final SelectSubQueryImpl subQuery = where .and(gte(InfluxClient.TIME, InfluxClient.timeIntervalTS(duration.getStartTimestamp()))) @@ -102,13 +107,8 @@ public class AggregationQuery implements IAggregationQueryDAO { entities.add(entity); }); - Collections.sort(entities, comparator); // re-sort by self, because of the result order by time. + entities.sort(comparator); // re-sort by self, because of the result order by time. return entities; } - private static final Comparator ASCENDING = (a, b) -> Long.compare( - Long.parseLong(a.getValue()), Long.parseLong(b.getValue())); - - private static final Comparator DESCENDING = (a, b) -> Long.compare( - Long.parseLong(b.getValue()), Long.parseLong(a.getValue())); } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AlarmQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AlarmQuery.java index ca74f0a0b..a9c67e598 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AlarmQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AlarmQuery.java @@ -73,9 +73,7 @@ public class AlarmQuery implements IAlarmQueryDAO { WhereQueryImpl countQuery = select().count(AlarmRecord.ID0) .from(client.getDatabase(), AlarmRecord.INDEX_NAME) .where(); - recallQuery.getClauses().forEach(clause -> { - countQuery.where(clause); - }); + recallQuery.getClauses().forEach(countQuery::where); Query query = new Query(countQuery.getCommand() + recallQuery.getCommand()); List results = client.query(query); diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetadataQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetadataQuery.java index bdfbb26be..2da548e27 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetadataQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetadataQuery.java @@ -80,7 +80,7 @@ public class MetadataQuery implements IMetadataQueryDAO { @Override public List getAllBrowserServices() throws IOException { - WhereQueryImpl query = select(ID_COLUMN, NAME, ServiceTraffic.GROUP) + final WhereQueryImpl query = select(ID_COLUMN, NAME, ServiceTraffic.GROUP) .from(client.getDatabase(), ServiceTraffic.INDEX_NAME) .where(eq(InfluxConstants.TagName.NODE_TYPE, String.valueOf(NodeType.Browser.value()))); return buildServices(query); @@ -88,16 +88,16 @@ public class MetadataQuery implements IMetadataQueryDAO { @Override public List getAllDatabases() throws IOException { - WhereQueryImpl query = select(ID_COLUMN, NAME, ServiceTraffic.GROUP) + final WhereQueryImpl query = select(ID_COLUMN, NAME, ServiceTraffic.GROUP) .from(client.getDatabase(), ServiceTraffic.INDEX_NAME) .where(eq(InfluxConstants.TagName.NODE_TYPE, String.valueOf(NodeType.Database.value()))); - QueryResult.Series series = client.queryForSingleSeries(query); + final QueryResult.Series series = client.queryForSingleSeries(query); if (log.isDebugEnabled()) { log.debug("SQL: {} result: {}", query.getCommand(), series); } - List databases = Lists.newArrayList(); + final List databases = Lists.newArrayList(); if (Objects.nonNull(series)) { for (List values : series.getValues()) { Database database = new Database(); @@ -111,8 +111,7 @@ public class MetadataQuery implements IMetadataQueryDAO { @Override public List searchServices(String keyword) throws IOException { - final WhereQueryImpl where = select( - ID_COLUMN, NAME, ServiceTraffic.GROUP) + final WhereQueryImpl where = select(ID_COLUMN, NAME, ServiceTraffic.GROUP) .from(client.getDatabase(), ServiceTraffic.INDEX_NAME) .where(eq(TagName.NODE_TYPE, String.valueOf(NodeType.Normal.value()))); if (!Strings.isNullOrEmpty(keyword)) { @@ -123,12 +122,11 @@ public class MetadataQuery implements IMetadataQueryDAO { @Override public Service searchService(String serviceCode) throws IOException { - WhereQueryImpl where = select( - ID_COLUMN, NAME, ServiceTraffic.GROUP) + final WhereQueryImpl whereQuery = select(ID_COLUMN, NAME, ServiceTraffic.GROUP) .from(client.getDatabase(), ServiceTraffic.INDEX_NAME) - .where(eq(TagName.NODE_TYPE, String.valueOf(NodeType.Normal.value()))) - .and(eq(ServiceTraffic.NAME, serviceCode)); - return buildServices(where).get(0); + .where(eq(TagName.NODE_TYPE, String.valueOf(NodeType.Normal.value()))); + whereQuery.and(eq(ServiceTraffic.NAME, serviceCode)); + return buildServices(whereQuery).get(0); } @Override @@ -147,7 +145,7 @@ public class MetadataQuery implements IMetadataQueryDAO { final QueryResult.Series series = client.queryForSingleSeries(where); if (log.isDebugEnabled()) { - log.debug("SQL: {} result: {}", where.getCommand(), series); + log.debug("SQL: {} result: {}.", where.getCommand(), series); } List list = new ArrayList<>(limit); @@ -189,7 +187,7 @@ public class MetadataQuery implements IMetadataQueryDAO { } if (Objects.isNull(series)) { - return Collections.EMPTY_LIST; + return Collections.emptyList(); } List> result = series.getValues(); diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetricsQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetricsQuery.java index a78b9dce0..70de0113a 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetricsQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetricsQuery.java @@ -63,22 +63,20 @@ public class MetricsQuery implements IMetricsQueryDAO { public long readMetricsValue(final MetricsCondition condition, final String valueColumnName, final Duration duration) throws IOException { - int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()); + final int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()); final Function function = ValueColumnMetadata.INSTANCE.getValueFunction(condition.getName()); if (function == Function.Latest) { return readMetricsValues(condition, valueColumnName, duration).getValues().latestValue(defaultValue); } final String measurement = condition.getName(); - SelectionQueryImpl query = select(); - switch (function) { - case Avg: - query.mean(valueColumnName); - break; - default: - query.sum(valueColumnName); + final SelectionQueryImpl query = select(); + if (function == Function.Avg) { + query.mean(valueColumnName); + } else { + query.sum(valueColumnName); } - WhereQueryImpl queryWhereQuery = query.from(client.getDatabase(), measurement).where(); + final WhereQueryImpl queryWhereQuery = query.from(client.getDatabase(), measurement).where(); final String entityId = condition.getEntity().buildId(); if (entityId != null) { @@ -90,7 +88,7 @@ public class MetricsQuery implements IMetricsQueryDAO { .and(lte(InfluxClient.TIME, InfluxClient.timeIntervalTS(duration.getEndTimestamp()))) .groupBy(InfluxConstants.TagName.ENTITY_ID); - List seriesList = client.queryForSeries(queryWhereQuery); + final List seriesList = client.queryForSeries(queryWhereQuery); if (log.isDebugEnabled()) { log.debug("SQL: {} result set: {}", queryWhereQuery.getCommand(), seriesList); } @@ -109,12 +107,10 @@ public class MetricsQuery implements IMetricsQueryDAO { final String valueColumnName, final Duration duration) throws IOException { final List pointOfTimes = duration.assembleDurationPoints(); - List ids = new ArrayList<>(pointOfTimes.size()); - pointOfTimes.forEach(pointOfTime -> { - ids.add(pointOfTime.id(condition.getEntity().buildId())); - }); + final List ids = new ArrayList<>(pointOfTimes.size()); + pointOfTimes.forEach(pointOfTime -> ids.add(pointOfTime.id(condition.getEntity().buildId()))); - WhereQueryImpl query = select() + final WhereQueryImpl query = select() .column(ID_COLUMN) .column(valueColumnName) .from(client.getDatabase(), condition.getName()) @@ -156,12 +152,10 @@ public class MetricsQuery implements IMetricsQueryDAO { final List labels, final Duration duration) throws IOException { final List pointOfTimes = duration.assembleDurationPoints(); - List ids = new ArrayList<>(pointOfTimes.size()); - pointOfTimes.forEach(pointOfTime -> { - ids.add(pointOfTime.id(condition.getEntity().buildId())); - }); + final List ids = new ArrayList<>(pointOfTimes.size()); + pointOfTimes.forEach(pointOfTime -> ids.add(pointOfTime.id(condition.getEntity().buildId()))); - WhereQueryImpl query = select() + final WhereQueryImpl query = select() .column(ID_COLUMN) .column(valueColumnName) .from(client.getDatabase(), condition.getName()) @@ -174,12 +168,12 @@ public class MetricsQuery implements IMetricsQueryDAO { query.where(contains(ID_COLUMN, Joiner.on("|").join(ids))); } } - List series = client.queryForSeries(query); + final List series = client.queryForSeries(query); if (log.isDebugEnabled()) { log.debug("SQL: {} result set: {}", query.getCommand(), series); } - Map idMap = new HashMap<>(); + final Map idMap = new HashMap<>(); if (!CollectionUtils.isEmpty(series)) { series.get(0).getValues().forEach(values -> { final String id = (String) values.get(1); @@ -196,26 +190,23 @@ public class MetricsQuery implements IMetricsQueryDAO { final String valueColumnName, final Duration duration) throws IOException { final List pointOfTimes = duration.assembleDurationPoints(); - List ids = new ArrayList<>(pointOfTimes.size()); - pointOfTimes.forEach(pointOfTime -> { - ids.add(pointOfTime.id(condition.getEntity().buildId())); - }); + final List ids = new ArrayList<>(pointOfTimes.size()); + pointOfTimes.forEach(pointOfTime -> ids.add(pointOfTime.id(condition.getEntity().buildId()))); - WhereQueryImpl query = select() + final WhereQueryImpl query = select() .column(ID_COLUMN) .column(valueColumnName) .from(client.getDatabase(), condition.getName()) .where(contains(ID_COLUMN, Joiner.on("|").join(ids))); - Map> thermodynamicValueMatrix = new HashMap<>(); - QueryResult.Series series = client.queryForSingleSeries(query); + final QueryResult.Series series = client.queryForSingleSeries(query); if (log.isDebugEnabled()) { log.debug("SQL: {} result set: {}", query.getCommand(), series); } final int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()); - HeatMap heatMap = new HeatMap(); + final HeatMap heatMap = new HeatMap(); if (series != null) { for (List values : series.getValues()) { heatMap.buildColumn(values.get(1).toString(), values.get(2).toString(), defaultValue); @@ -223,7 +214,6 @@ public class MetricsQuery implements IMetricsQueryDAO { } heatMap.fixMissingColumns(ids, defaultValue); - return heatMap; } -} \ No newline at end of file +} diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/NetworkAddressAliasDAO.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/NetworkAddressAliasDAO.java index ef60af82a..07c359f98 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/NetworkAddressAliasDAO.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/NetworkAddressAliasDAO.java @@ -40,7 +40,7 @@ import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.select; @Slf4j public class NetworkAddressAliasDAO implements INetworkAddressAliasDAO { private final NetworkAddressAlias.Builder builder = new NetworkAddressAlias.Builder(); - private InfluxClient client; + private final InfluxClient client; public NetworkAddressAliasDAO(final InfluxClient client) { this.client = client; diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskLogQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskLogQuery.java index 2fb972c58..99661f642 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskLogQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskLogQuery.java @@ -37,8 +37,8 @@ import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.select; @Slf4j public class ProfileTaskLogQuery implements IProfileTaskLogQueryDAO { - private InfluxClient client; - private int fetchTaskLogMaxSize; + private final InfluxClient client; + private final int fetchTaskLogMaxSize; public ProfileTaskLogQuery(InfluxClient client, int fetchTaskLogMaxSize) { this.client = client; @@ -68,16 +68,14 @@ public class ProfileTaskLogQuery implements IProfileTaskLogQueryDAO { series.getValues().stream() // re-sort by self, because of the result order by time. .sorted((a, b) -> Long.compare(((Number) b.get(1)).longValue(), ((Number) a.get(1)).longValue())) - .forEach(values -> { - taskLogs.add(ProfileTaskLog.builder() - .id((String) values.get(2)) - .taskId((String) values.get(3)) - .instanceId((String) values.get(4)) - .operationTime(((Number) values.get(5)).longValue()) - .operationType(ProfileTaskLogOperationType.parse( - ((Number) values.get(6)).intValue())) - .build()); - }); + .forEach(values -> taskLogs.add(ProfileTaskLog.builder() + .id((String) values.get(2)) + .taskId((String) values.get(3)) + .instanceId((String) values.get(4)) + .operationTime(((Number) values.get(5)).longValue()) + .operationType(ProfileTaskLogOperationType.parse( + ((Number) values.get(6)).intValue())) + .build())); return taskLogs; } } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskQuery.java index 2ec06a5a1..64bc56ff4 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskQuery.java @@ -52,7 +52,7 @@ public class ProfileTaskQuery implements IProfileTaskQueryDAO { final Long startTimeBucket, final Long endTimeBucket, final Integer limit) throws IOException { - WhereQueryImpl query = + final WhereQueryImpl query = select( InfluxConstants.ID_COLUMN, ProfileTaskRecord.SERVICE_ID, @@ -83,15 +83,13 @@ public class ProfileTaskQuery implements IProfileTaskQueryDAO { query.limit(limit); } - List tasks = Lists.newArrayList(); + final List tasks = Lists.newArrayList(); QueryResult.Series series = client.queryForSingleSeries(query); if (log.isDebugEnabled()) { log.debug("SQL: {} result: {}", query.getCommand(), series); } if (series != null) { - series.getValues().forEach(values -> { - tasks.add(profileTaskBuilder(values)); - }); + series.getValues().forEach(values -> tasks.add(profileTaskBuilder(values))); } return tasks; } @@ -101,7 +99,7 @@ public class ProfileTaskQuery implements IProfileTaskQueryDAO { if (StringUtil.isEmpty(id)) { return null; } - SelectQueryImpl query = select( + final SelectQueryImpl query = select( InfluxConstants.ID_COLUMN, ProfileTaskRecord.SERVICE_ID, ProfileTaskRecord.ENDPOINT_NAME, @@ -117,7 +115,7 @@ public class ProfileTaskQuery implements IProfileTaskQueryDAO { .and(eq(InfluxConstants.ID_COLUMN, id)) .limit(1); - QueryResult.Series series = client.queryForSingleSeries(query); + final QueryResult.Series series = client.queryForSingleSeries(query); if (log.isDebugEnabled()) { log.debug("SQL: {} result: {}", query.getCommand(), series); } @@ -127,7 +125,7 @@ public class ProfileTaskQuery implements IProfileTaskQueryDAO { return null; } - private static final ProfileTask profileTaskBuilder(List values) { + private static ProfileTask profileTaskBuilder(List values) { return ProfileTask.builder() .id((String) values.get(1)) .serviceId((String) values.get(2)) diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java index b5d08be77..040074f24 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java @@ -38,6 +38,7 @@ import org.apache.skywalking.oap.server.storage.plugin.influxdb.InfluxClient; import org.apache.skywalking.oap.server.storage.plugin.influxdb.InfluxConstants; import org.elasticsearch.common.Strings; import org.influxdb.dto.QueryResult; +import org.influxdb.querybuilder.SelectQueryImpl; import org.influxdb.querybuilder.WhereQueryImpl; import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.contains; @@ -56,26 +57,25 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA @Override public List queryProfiledSegments(String taskId) throws IOException { - WhereQueryImpl query = select(ProfileThreadSnapshotRecord.SEGMENT_ID) + final WhereQueryImpl countQuery = select(ProfileThreadSnapshotRecord.SEGMENT_ID) .from(client.getDatabase(), ProfileThreadSnapshotRecord.INDEX_NAME) - .where() - .and(eq(ProfileThreadSnapshotRecord.TASK_ID, taskId)) - .and(eq(ProfileThreadSnapshotRecord.SEQUENCE, 0)); + .where(); + + countQuery.and(eq(ProfileThreadSnapshotRecord.TASK_ID, taskId)) + .and(eq(ProfileThreadSnapshotRecord.SEQUENCE, 0)); final LinkedList segments = new LinkedList<>(); - QueryResult.Series series = client.queryForSingleSeries(query); + QueryResult.Series series = client.queryForSingleSeries(countQuery); if (Objects.isNull(series)) { return Collections.emptyList(); } - series.getValues().forEach(values -> { - segments.add((String) values.get(1)); - }); + series.getValues().forEach(values -> segments.add((String) values.get(1))); if (segments.isEmpty()) { return Collections.emptyList(); } - query = select() + final WhereQueryImpl whereQuery = select() .function(InfluxConstants.SORT_ASC, SegmentRecord.START_TIME, segments.size()) .column(SegmentRecord.SEGMENT_ID) .column(SegmentRecord.START_TIME) @@ -84,16 +84,16 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA .column(SegmentRecord.IS_ERROR) .column(SegmentRecord.TRACE_ID) .from(client.getDatabase(), SegmentRecord.INDEX_NAME) - .where() - .and(contains(SegmentRecord.SEGMENT_ID, Joiner.on("|").join(segments))); + .where(); + whereQuery.and(contains(SegmentRecord.SEGMENT_ID, Joiner.on("|").join(segments))); ArrayList result = Lists.newArrayListWithCapacity(segments.size()); - client.queryForSingleSeries(query) + client.queryForSingleSeries(whereQuery) .getValues() .stream() .sorted((a, b) -> Long.compare(((Number) b.get(1)).longValue(), ((Number) a.get(1)).longValue())) .forEach(values -> { - BasicTrace basicTrace = new BasicTrace(); + final BasicTrace basicTrace = new BasicTrace(); basicTrace.setSegmentId((String) values.get(2)); basicTrace.setStart(String.valueOf(((Number) values.get(3)).longValue())); @@ -122,7 +122,7 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA @Override public List queryRecords(String segmentId, int minSequence, int maxSequence) throws IOException { - WhereQueryImpl query = select( + WhereQueryImpl whereQuery = select( ProfileThreadSnapshotRecord.TASK_ID, ProfileThreadSnapshotRecord.SEGMENT_ID, ProfileThreadSnapshotRecord.DUMP_TIME, @@ -130,18 +130,19 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA ProfileThreadSnapshotRecord.STACK_BINARY ) .from(client.getDatabase(), ProfileThreadSnapshotRecord.INDEX_NAME) - .where(eq(ProfileThreadSnapshotRecord.SEGMENT_ID, segmentId)) - .and(gte(ProfileThreadSnapshotRecord.SEQUENCE, minSequence)) - .and(lte(ProfileThreadSnapshotRecord.SEQUENCE, maxSequence)); + .where(eq(ProfileThreadSnapshotRecord.SEGMENT_ID, segmentId)); - QueryResult.Series series = client.queryForSingleSeries(query); + whereQuery.and(gte(ProfileThreadSnapshotRecord.SEQUENCE, minSequence)) + .and(lte(ProfileThreadSnapshotRecord.SEQUENCE, maxSequence)); + + final QueryResult.Series series = client.queryForSingleSeries(whereQuery); if (log.isDebugEnabled()) { - log.debug("SQL: {} result: {}", query.getCommand(), series); + log.debug("SQL: {} result: {}", whereQuery.getCommand(), series); } if (Objects.isNull(series)) { - return Collections.EMPTY_LIST; + return Collections.emptyList(); } - ArrayList result = new ArrayList<>(maxSequence - minSequence); + final ArrayList result = new ArrayList<>(maxSequence - minSequence); series.getValues().forEach(values -> { ProfileThreadSnapshotRecord record = new ProfileThreadSnapshotRecord(); @@ -162,29 +163,31 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA @Override public SegmentRecord getProfiledSegment(String segmentId) throws IOException { - WhereQueryImpl query = select().column(SegmentRecord.SEGMENT_ID) - .column(SegmentRecord.TRACE_ID) - .column(SegmentRecord.SERVICE_ID) - .column(SegmentRecord.ENDPOINT_NAME) - .column(SegmentRecord.START_TIME) - .column(SegmentRecord.END_TIME) - .column(SegmentRecord.LATENCY) - .column(SegmentRecord.IS_ERROR) - .column(SegmentRecord.DATA_BINARY) - .column(SegmentRecord.VERSION) - .from(client.getDatabase(), SegmentRecord.INDEX_NAME) - .where() - .and(eq(SegmentRecord.SEGMENT_ID, segmentId)); - List series = client.queryForSeries(query); + WhereQueryImpl whereQuery = select() + .column(SegmentRecord.SEGMENT_ID) + .column(SegmentRecord.TRACE_ID) + .column(SegmentRecord.SERVICE_ID) + .column(SegmentRecord.ENDPOINT_NAME) + .column(SegmentRecord.START_TIME) + .column(SegmentRecord.END_TIME) + .column(SegmentRecord.LATENCY) + .column(SegmentRecord.IS_ERROR) + .column(SegmentRecord.DATA_BINARY) + .column(SegmentRecord.VERSION) + .from(client.getDatabase(), SegmentRecord.INDEX_NAME) + .where(); + + whereQuery.and(eq(SegmentRecord.SEGMENT_ID, segmentId)); + List series = client.queryForSeries(whereQuery); if (log.isDebugEnabled()) { - log.debug("SQL: {} result set: {}", query.getCommand(), series); + log.debug("SQL: {} result set: {}", whereQuery.getCommand(), series); } if (Objects.isNull(series) || series.isEmpty()) { return null; } - List values = series.get(0).getValues().get(0); - SegmentRecord segmentRecord = new SegmentRecord(); + final List values = series.get(0).getValues().get(0); + final SegmentRecord segmentRecord = new SegmentRecord(); segmentRecord.setSegmentId((String) values.get(1)); segmentRecord.setTraceId((String) values.get(2)); @@ -196,7 +199,7 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA segmentRecord.setIsError(((Number) values.get(8)).intValue()); segmentRecord.setVersion(((Number) values.get(10)).intValue()); - String base64 = (String) values.get(9); + final String base64 = (String) values.get(9); if (!Strings.isNullOrEmpty(base64)) { segmentRecord.setDataBinary(Base64.getDecoder().decode(base64)); } @@ -205,13 +208,14 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA } private int querySequenceWithAgg(String function, String segmentId, long start, long end) throws IOException { - WhereQueryImpl query = select() + WhereQueryImpl query = select() .function(function, ProfileThreadSnapshotRecord.SEQUENCE) .from(client.getDatabase(), ProfileThreadSnapshotRecord.INDEX_NAME) - .where() - .and(eq(ProfileThreadSnapshotRecord.SEGMENT_ID, segmentId)) - .and(gte(ProfileThreadSnapshotRecord.DUMP_TIME, start)) - .and(lte(ProfileThreadSnapshotRecord.DUMP_TIME, end)); + .where(); + + query.and(eq(ProfileThreadSnapshotRecord.SEGMENT_ID, segmentId)) + .and(gte(ProfileThreadSnapshotRecord.DUMP_TIME, start)) + .and(lte(ProfileThreadSnapshotRecord.DUMP_TIME, end)); return client.getCounter(query); } } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopNRecordsQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopNRecordsQuery.java index d44f4f796..7284214a8 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopNRecordsQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopNRecordsQuery.java @@ -35,6 +35,7 @@ import org.apache.skywalking.oap.server.core.storage.query.ITopNRecordsQueryDAO; import org.apache.skywalking.oap.server.storage.plugin.influxdb.InfluxClient; import org.apache.skywalking.oap.server.storage.plugin.influxdb.InfluxConstants; import org.influxdb.dto.QueryResult; +import org.influxdb.querybuilder.SelectQueryImpl; import org.influxdb.querybuilder.WhereQueryImpl; import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.eq; @@ -62,13 +63,14 @@ public class TopNRecordsQuery implements ITopNRecordsQueryDAO { comparator = DESCENDING; } - WhereQueryImpl query = select() + final WhereQueryImpl query = select() .function(function, valueColumnName, condition.getTopN()) .column(TopN.STATEMENT) .column(TopN.TRACE_ID) .from(client.getDatabase(), condition.getName()) - .where() - .and(gte(TopN.TIME_BUCKET, duration.getStartTimeBucketInSec())) + .where(); + + query.and(gte(TopN.TIME_BUCKET, duration.getStartTimeBucketInSec())) .and(lte(TopN.TIME_BUCKET, duration.getEndTimeBucketInSec())); if (StringUtil.isNotEmpty(condition.getParentService())) { @@ -94,12 +96,12 @@ public class TopNRecordsQuery implements ITopNRecordsQueryDAO { records.add(record); }); - Collections.sort(records, comparator); // re-sort by self, because of the result order by time. + records.sort(comparator); // re-sort by self, because of the result order by time. return records; } - private static final Comparator ASCENDING = (a, b) -> Long.compare( - Long.parseLong(a.getValue()), Long.parseLong(b.getValue())); + private static final Comparator ASCENDING = Comparator.comparingLong( + a -> Long.parseLong(a.getValue())); private static final Comparator DESCENDING = (a, b) -> Long.compare( Long.parseLong(b.getValue()), Long.parseLong(a.getValue())); diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopologyQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopologyQuery.java index 5d8d99ac3..da76d5190 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopologyQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopologyQuery.java @@ -54,11 +54,11 @@ public class TopologyQuery implements ITopologyQueryDAO { } @Override - public List loadServiceRelationsDetectedAtServerSide(long startTB, - long endTB, - List serviceIds) throws IOException { - String measurement = ServiceRelationServerSideMetrics.INDEX_NAME; - WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( + public List loadServiceRelationsDetectedAtServerSide(final long startTB, + final long endTB, + final List serviceIds) throws IOException { + final String measurement = ServiceRelationServerSideMetrics.INDEX_NAME; + final WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( measurement, startTB, endTB, @@ -71,12 +71,11 @@ public class TopologyQuery implements ITopologyQueryDAO { } @Override - public List loadServiceRelationDetectedAtClientSide(long startTB, - long endTB, + public List loadServiceRelationDetectedAtClientSide(final long startTB, + final long endTB, List serviceIds) throws IOException { - String measurement = ServiceRelationClientSideMetrics.INDEX_NAME; - WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( - measurement, + final WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( + ServiceRelationClientSideMetrics.INDEX_NAME, startTB, endTB, ServiceRelationServerSideMetrics.SOURCE_SERVICE_ID, @@ -87,11 +86,10 @@ public class TopologyQuery implements ITopologyQueryDAO { } @Override - public List loadServiceRelationsDetectedAtServerSide(long startTB, - long endTB) throws IOException { - String measurement = ServiceRelationServerSideMetrics.INDEX_NAME; - WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( - measurement, + public List loadServiceRelationsDetectedAtServerSide(final long startTB, + final long endTB) throws IOException { + final WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( + ServiceRelationServerSideMetrics.INDEX_NAME, startTB, endTB, ServiceRelationServerSideMetrics.SOURCE_SERVICE_ID, @@ -102,11 +100,10 @@ public class TopologyQuery implements ITopologyQueryDAO { } @Override - public List loadServiceRelationDetectedAtClientSide(long startTB, - long endTB) throws IOException { - String tableName = ServiceRelationClientSideMetrics.INDEX_NAME; + public List loadServiceRelationDetectedAtClientSide(final long startTB, + final long endTB) throws IOException { WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( - tableName, + ServiceRelationClientSideMetrics.INDEX_NAME, startTB, endTB, ServiceRelationServerSideMetrics.SOURCE_SERVICE_ID, @@ -117,13 +114,12 @@ public class TopologyQuery implements ITopologyQueryDAO { } @Override - public List loadInstanceRelationDetectedAtServerSide(String clientServiceId, - String serverServiceId, - long startTB, - long endTB) throws IOException { - String measurement = ServiceInstanceRelationServerSideMetrics.INDEX_NAME; + public List loadInstanceRelationDetectedAtServerSide(final String clientServiceId, + final String serverServiceId, + final long startTB, + final long endTB) throws IOException { WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceInstanceCallsQuery( - measurement, + ServiceInstanceRelationServerSideMetrics.INDEX_NAME, startTB, endTB, ServiceInstanceRelationServerSideMetrics.SOURCE_SERVICE_ID, @@ -134,13 +130,12 @@ public class TopologyQuery implements ITopologyQueryDAO { } @Override - public List loadInstanceRelationDetectedAtClientSide(String clientServiceId, - String serverServiceId, - long startTB, - long endTB) throws IOException { - String measurement = ServiceInstanceRelationClientSideMetrics.INDEX_NAME; + public List loadInstanceRelationDetectedAtClientSide(final String clientServiceId, + final String serverServiceId, + final long startTB, + final long endTB) throws IOException { WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceInstanceCallsQuery( - measurement, + ServiceInstanceRelationClientSideMetrics.INDEX_NAME, startTB, endTB, ServiceInstanceRelationClientSideMetrics.SOURCE_SERVICE_ID, @@ -151,13 +146,11 @@ public class TopologyQuery implements ITopologyQueryDAO { } @Override - public List loadEndpointRelation(long startTB, - long endTB, - String destEndpointId) throws IOException { - String measurement = EndpointRelationServerSideMetrics.INDEX_NAME; - - WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( - measurement, + public List loadEndpointRelation(final long startTB, + final long endTB, + final String destEndpointId) throws IOException { + final WhereSubQueryImpl, SelectQueryImpl> subQuery = buildServiceCallsQuery( + EndpointRelationServerSideMetrics.INDEX_NAME, startTB, endTB, EndpointRelationServerSideMetrics.SOURCE_ENDPOINT, @@ -166,8 +159,8 @@ public class TopologyQuery implements ITopologyQueryDAO { ); subQuery.and(eq(EndpointRelationServerSideMetrics.DEST_ENDPOINT, destEndpointId)); - WhereSubQueryImpl, SelectQueryImpl> subQuery2 = buildServiceCallsQuery( - measurement, + final WhereSubQueryImpl, SelectQueryImpl> subQuery2 = buildServiceCallsQuery( + EndpointRelationServerSideMetrics.INDEX_NAME, startTB, endTB, EndpointRelationServerSideMetrics.SOURCE_ENDPOINT, @@ -176,19 +169,20 @@ public class TopologyQuery implements ITopologyQueryDAO { ); subQuery2.and(eq(EndpointRelationServerSideMetrics.SOURCE_ENDPOINT, destEndpointId)); - List calls = buildEndpointCalls(buildQuery(subQuery), DetectPoint.SERVER); + final List calls = buildEndpointCalls(buildQuery(subQuery), DetectPoint.SERVER); calls.addAll(buildEndpointCalls(buildQuery(subQuery), DetectPoint.CLIENT)); return calls; } private WhereSubQueryImpl, SelectQueryImpl> buildServiceCallsQuery( - String measurement, - long startTB, - long endTB, - String sourceCName, - String destCName, - List serviceIds) { - WhereSubQueryImpl, SelectQueryImpl> subQuery = select() + final String measurement, + final long startTB, + final long endTB, + final String sourceCName, + final String destCName, + final List serviceIds) { + + final WhereSubQueryImpl, SelectQueryImpl> subQuery = select() .fromSubQuery(client.getDatabase()) .function("distinct", ServiceInstanceRelationServerSideMetrics.COMPONENT_ID) .as(ServiceInstanceRelationClientSideMetrics.COMPONENT_ID) @@ -198,7 +192,8 @@ public class TopologyQuery implements ITopologyQueryDAO { .and(lte(InfluxClient.TIME, InfluxClient.timeIntervalTB(endTB))); if (!serviceIds.isEmpty()) { - WhereNested whereNested = subQuery.andNested(); + WhereNested, SelectQueryImpl>> whereNested = subQuery + .andNested(); for (String id : serviceIds) { whereNested.or(eq(sourceCName, id)) .or(eq(destCName, id)); @@ -209,24 +204,25 @@ public class TopologyQuery implements ITopologyQueryDAO { } private WhereSubQueryImpl, SelectQueryImpl> buildServiceInstanceCallsQuery( - String measurement, - long startTB, - long endTB, - String sourceCName, - String destCName, - String sourceServiceId, - String destServiceId) { + final String measurement, + final long startTB, + final long endTB, + final String sourceCName, + final String destCName, + final String sourceServiceId, + final String destServiceId) { - WhereSubQueryImpl, SelectQueryImpl> subQuery = select() + final WhereSubQueryImpl, SelectQueryImpl> subQuery = select() .fromSubQuery(client.getDatabase()) .function("distinct", ServiceInstanceRelationServerSideMetrics.COMPONENT_ID) .as(ServiceInstanceRelationClientSideMetrics.COMPONENT_ID) .from(measurement) - .where() - .and(gte(InfluxClient.TIME, InfluxClient.timeIntervalTB(startTB))) - .and(lte(InfluxClient.TIME, InfluxClient.timeIntervalTB(endTB))); + .where(); - StringBuilder builder = new StringBuilder("(("); + subQuery.and(gte(InfluxClient.TIME, InfluxClient.timeIntervalTB(startTB))) + .and(lte(InfluxClient.TIME, InfluxClient.timeIntervalTB(endTB))); + + final StringBuilder builder = new StringBuilder("(("); builder.append(sourceCName).append("='").append(sourceServiceId) .append("' and ") .append(destCName).append("='").append(destServiceId) @@ -242,7 +238,7 @@ public class TopologyQuery implements ITopologyQueryDAO { private List buildServiceCalls(Query query, DetectPoint detectPoint) throws IOException { - QueryResult.Series series = client.queryForSingleSeries(query); + final QueryResult.Series series = client.queryForSingleSeries(query); if (log.isDebugEnabled()) { log.debug("SQL: {} result set: {}", query.getCommand(), series); @@ -251,7 +247,7 @@ public class TopologyQuery implements ITopologyQueryDAO { return Collections.emptyList(); } - List calls = new ArrayList<>(); + final List calls = new ArrayList<>(); series.getValues().forEach(values -> { Call.CallDetail call = new Call.CallDetail(); String entityId = String.valueOf(values.get(1)); @@ -270,8 +266,7 @@ public class TopologyQuery implements ITopologyQueryDAO { return query; } - private List buildInstanceCalls(Query query, - DetectPoint detectPoint) throws IOException { + private List buildInstanceCalls(Query query, DetectPoint detectPoint) throws IOException { QueryResult.Series series = client.queryForSingleSeries(query); if (log.isDebugEnabled()) { diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java index 2b969c2c1..efc28dc9d 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java @@ -178,23 +178,24 @@ public class TraceQuery implements ITraceQueryDAO { @Override public List queryByTraceId(String traceId) throws IOException { - WhereQueryImpl query = select().column(SegmentRecord.SEGMENT_ID) - .column(SegmentRecord.TRACE_ID) - .column(SegmentRecord.SERVICE_ID) - .column(SegmentRecord.SERVICE_INSTANCE_ID) - .column(SegmentRecord.ENDPOINT_NAME) - .column(SegmentRecord.START_TIME) - .column(SegmentRecord.END_TIME) - .column(SegmentRecord.LATENCY) - .column(SegmentRecord.IS_ERROR) - .column(SegmentRecord.DATA_BINARY) - .column(SegmentRecord.VERSION) - .from(client.getDatabase(), SegmentRecord.INDEX_NAME) - .where() - .and(eq(SegmentRecord.TRACE_ID, traceId)); - List series = client.queryForSeries(query); + WhereQueryImpl whereQuery = select().column(SegmentRecord.SEGMENT_ID) + .column(SegmentRecord.TRACE_ID) + .column(SegmentRecord.SERVICE_ID) + .column(SegmentRecord.SERVICE_INSTANCE_ID) + .column(SegmentRecord.ENDPOINT_NAME) + .column(SegmentRecord.START_TIME) + .column(SegmentRecord.END_TIME) + .column(SegmentRecord.LATENCY) + .column(SegmentRecord.IS_ERROR) + .column(SegmentRecord.DATA_BINARY) + .column(SegmentRecord.VERSION) + .from(client.getDatabase(), SegmentRecord.INDEX_NAME) + .where(); + + whereQuery.and(eq(SegmentRecord.TRACE_ID, traceId)); + List series = client.queryForSeries(whereQuery); if (log.isDebugEnabled()) { - log.debug("SQL: {} result set: {}", query.getCommand(), series); + log.debug("SQL: {} result set: {}", whereQuery.getCommand(), series); } if (series == null || series.isEmpty()) { return Collections.emptyList(); @@ -226,7 +227,7 @@ public class TraceQuery implements ITraceQueryDAO { } @Override - public List doFlexibleTraceQuery(String traceId) throws IOException { + public List doFlexibleTraceQuery(String traceId) { return Collections.emptyList(); } } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/UITemplateManagementDAOImpl.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/UITemplateManagementDAOImpl.java index ba2e81592..99a0e2366 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/UITemplateManagementDAOImpl.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/UITemplateManagementDAOImpl.java @@ -50,9 +50,9 @@ public class UITemplateManagementDAOImpl implements UITemplateManagementDAO { @Override public List getAllTemplates(final Boolean includingDisabled) throws IOException { - WhereQueryImpl where = select().raw("*::field") - .from(client.getDatabase(), UITemplate.INDEX_NAME) - .where(); + final WhereQueryImpl where = select().raw("*::field") + .from(client.getDatabase(), UITemplate.INDEX_NAME) + .where(); if (!includingDisabled) { where.and(eq(UITemplate.DISABLED, BooleanUtils.FALSE)); } @@ -82,11 +82,11 @@ public class UITemplateManagementDAOImpl implements UITemplateManagementDAO { final UITemplate.Builder builder = new UITemplate.Builder(); final UITemplate uiTemplate = setting.toEntity(); - Point point = Point.measurement(UITemplate.INDEX_NAME) - .tag(InfluxConstants.TagName.ID_COLUMN, uiTemplate.id()) - .fields(builder.data2Map(uiTemplate)) - .time(1L, TimeUnit.NANOSECONDS) - .build(); + final Point point = Point.measurement(UITemplate.INDEX_NAME) + .tag(InfluxConstants.TagName.ID_COLUMN, uiTemplate.id()) + .fields(builder.data2Map(uiTemplate)) + .time(1L, TimeUnit.NANOSECONDS) + .build(); client.write(point); return TemplateChangeStatus.builder().status(true).build(); } @@ -102,11 +102,11 @@ public class UITemplateManagementDAOImpl implements UITemplateManagementDAO { QueryResult.Series series = client.queryForSingleSeries(query); if (Objects.nonNull(series)) { - Point point = Point.measurement(UITemplate.INDEX_NAME) - .fields(builder.data2Map(uiTemplate)) - .tag(InfluxConstants.TagName.ID_COLUMN, uiTemplate.id()) - .time(1L, TimeUnit.NANOSECONDS) - .build(); + final Point point = Point.measurement(UITemplate.INDEX_NAME) + .fields(builder.data2Map(uiTemplate)) + .tag(InfluxConstants.TagName.ID_COLUMN, uiTemplate.id()) + .time(1L, TimeUnit.NANOSECONDS) + .build(); client.write(point); return TemplateChangeStatus.builder().status(true).build(); } else { @@ -121,11 +121,11 @@ public class UITemplateManagementDAOImpl implements UITemplateManagementDAO { .where(eq(InfluxConstants.TagName.ID_COLUMN, name)); QueryResult.Series series = client.queryForSingleSeries(query); if (Objects.nonNull(series)) { - Point point = Point.measurement(UITemplate.INDEX_NAME) - .tag(InfluxConstants.TagName.ID_COLUMN, name) - .addField(UITemplate.DISABLED, BooleanUtils.TRUE) - .time(1L, TimeUnit.NANOSECONDS) - .build(); + final Point point = Point.measurement(UITemplate.INDEX_NAME) + .tag(InfluxConstants.TagName.ID_COLUMN, name) + .addField(UITemplate.DISABLED, BooleanUtils.TRUE) + .time(1L, TimeUnit.NANOSECONDS) + .build(); client.write(point); return TemplateChangeStatus.builder().status(true).build(); } else {