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

chore: code polish (#6025)

This commit is contained in:
Daming 2020-12-17 22:54:36 +08:00 committed by GitHub
parent 1297d2e1d2
commit 8d33f0275e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 261 additions and 291 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java
View File

@ -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);
}
}

11
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageProvider.java
View File

@ -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 {
}

5
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxTableInstaller.java
View File

@ -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
}
}

8
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/TableMetaInfo.java
View File

@ -41,9 +41,9 @@ import org.apache.skywalking.oap.server.core.storage.model.ModelColumn;
public class TableMetaInfo {
private static final Map<String, TableMetaInfo> TABLES = new HashMap<>();
private Map<String, String> storageAndColumnMap;
private Map<String, String> storageAndTagMap;
private Model model;
private final Map<String, String> storageAndColumnMap;
private final Map<String, String> storageAndTagMap;
private final Model model;
public static void addModel(Model model) {
final List<ModelColumn> 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)

10
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxInsertRequest.java
View File

@ -36,17 +36,17 @@ import org.influxdb.dto.Point;
* InfluxDB Point wrapper.
*/
public class InfluxInsertRequest implements InsertRequest, UpdateRequest {
private Point.Builder builder;
private Map<String, Object> fields = Maps.newHashMap();
private final Point.Builder builder;
private final Map<String, Object> fields = Maps.newHashMap();
public InfluxInsertRequest(Model model, StorageData storageData, StorageBuilder storageBuilder) {
Map<String, Object> objectMap = storageBuilder.data2Map(storageData);
public <T extends StorageData> InfluxInsertRequest(Model model, T storageData, StorageBuilder<T> storageBuilder) {
final Map<String, Object> 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(

2
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/InfluxStorageDAO.java
View File

@ -44,7 +44,7 @@ public class InfluxStorageDAO implements StorageDAO {
@Override
public IRecordDAO newRecordDao(StorageBuilder<Record> storageBuilder) {
return new RecordDAO(influxClient, storageBuilder);
return new RecordDAO(storageBuilder);
}
@Override

4
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/ManagementDAO.java
View File

@ -40,8 +40,8 @@ import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.select;
@Slf4j
public class ManagementDAO implements IManagementDAO {
private InfluxClient client;
private StorageBuilder<ManagementData> storageBuilder;
private final InfluxClient client;
private final StorageBuilder<ManagementData> storageBuilder;
public ManagementDAO(InfluxClient client, StorageBuilder<ManagementData> storageBuilder) {
this.client = client;

18
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/MetricsDAO.java
View File

@ -58,7 +58,7 @@ public class MetricsDAO implements IMetricsDAO {
@Override
public List<Metrics> multiGet(Model model, List<String> ids) throws IOException {
WhereQueryImpl<SelectQueryImpl> query = select()
final WhereQueryImpl<SelectQueryImpl> 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> metrics = Lists.newArrayList();
List<String> columns = series.getColumns();
final List<String> columns = series.getColumns();
TableMetaInfo metaInfo = TableMetaInfo.get(model.getName());
Map<String, String> storageAndColumnMap = metaInfo.getStorageAndColumnMap();
final TableMetaInfo metaInfo = TableMetaInfo.get(model.getName());
final Map<String, String> storageAndColumnMap = metaInfo.getStorageAndColumnMap();
series.getValues().forEach(values -> {
Map<String, Object> 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);
}
}

11
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/NoneStreamDAO.java
View File

@ -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<NoneStream> storageBuilder;
private final InfluxClient client;
private final StorageBuilder<NoneStream> storageBuilder;
public NoneStreamDAO(InfluxClient client, StorageBuilder<NoneStream> 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());
}
}

29
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/RecordDAO.java
View File

@ -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<Record> storageBuilder;
private final StorageBuilder<Record> storageBuilder;
public RecordDAO(InfluxClient client, StorageBuilder<Record> storageBuilder) {
this.client = client;
public RecordDAO(StorageBuilder<Record> 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<String, List<SpanTag>> 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;
}

22
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AggregationQuery.java
View File

@ -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<SelectedRecord> ASCENDING =
Comparator.comparingLong(a -> Long.parseLong(a.getValue()));
private static final Comparator<SelectedRecord> 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<SelectSubQueryImpl<SelectQueryImpl>, 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<SelectQueryImpl> 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<SelectedRecord> ASCENDING = (a, b) -> Long.compare(
Long.parseLong(a.getValue()), Long.parseLong(b.getValue()));
private static final Comparator<SelectedRecord> DESCENDING = (a, b) -> Long.compare(
Long.parseLong(b.getValue()), Long.parseLong(a.getValue()));
}

4
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/AlarmQuery.java
View File

@ -73,9 +73,7 @@ public class AlarmQuery implements IAlarmQueryDAO {
WhereQueryImpl<SelectQueryImpl> 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<QueryResult.Result> results = client.query(query);

24
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetadataQuery.java
View File

@ -80,7 +80,7 @@ public class MetadataQuery implements IMetadataQueryDAO {
@Override
public List<Service> getAllBrowserServices() throws IOException {
WhereQueryImpl<SelectQueryImpl> query = select(ID_COLUMN, NAME, ServiceTraffic.GROUP)
final WhereQueryImpl<SelectQueryImpl> 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<Database> getAllDatabases() throws IOException {
WhereQueryImpl<SelectQueryImpl> query = select(ID_COLUMN, NAME, ServiceTraffic.GROUP)
final WhereQueryImpl<SelectQueryImpl> 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<Database> databases = Lists.newArrayList();
final List<Database> databases = Lists.newArrayList();
if (Objects.nonNull(series)) {
for (List<Object> values : series.getValues()) {
Database database = new Database();
@ -111,8 +111,7 @@ public class MetadataQuery implements IMetadataQueryDAO {
@Override
public List<Service> searchServices(String keyword) throws IOException {
final WhereQueryImpl<SelectQueryImpl> where = select(
ID_COLUMN, NAME, ServiceTraffic.GROUP)
final WhereQueryImpl<SelectQueryImpl> 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<SelectQueryImpl> where = select(
ID_COLUMN, NAME, ServiceTraffic.GROUP)
final WhereQueryImpl<SelectQueryImpl> 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<Endpoint> 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<List<Object>> result = series.getValues();

54
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/MetricsQuery.java
View File

@ -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<SelectQueryImpl> queryWhereQuery = query.from(client.getDatabase(), measurement).where();
final WhereQueryImpl<SelectQueryImpl> 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<QueryResult.Series> seriesList = client.queryForSeries(queryWhereQuery);
final List<QueryResult.Series> 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<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
List<String> ids = new ArrayList<>(pointOfTimes.size());
pointOfTimes.forEach(pointOfTime -> {
ids.add(pointOfTime.id(condition.getEntity().buildId()));
});
final List<String> ids = new ArrayList<>(pointOfTimes.size());
pointOfTimes.forEach(pointOfTime -> ids.add(pointOfTime.id(condition.getEntity().buildId())));
WhereQueryImpl<SelectQueryImpl> query = select()
final WhereQueryImpl<SelectQueryImpl> query = select()
.column(ID_COLUMN)
.column(valueColumnName)
.from(client.getDatabase(), condition.getName())
@ -156,12 +152,10 @@ public class MetricsQuery implements IMetricsQueryDAO {
final List<String> labels,
final Duration duration) throws IOException {
final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
List<String> ids = new ArrayList<>(pointOfTimes.size());
pointOfTimes.forEach(pointOfTime -> {
ids.add(pointOfTime.id(condition.getEntity().buildId()));
});
final List<String> ids = new ArrayList<>(pointOfTimes.size());
pointOfTimes.forEach(pointOfTime -> ids.add(pointOfTime.id(condition.getEntity().buildId())));
WhereQueryImpl<SelectQueryImpl> query = select()
final WhereQueryImpl<SelectQueryImpl> 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<QueryResult.Series> series = client.queryForSeries(query);
final List<QueryResult.Series> series = client.queryForSeries(query);
if (log.isDebugEnabled()) {
log.debug("SQL: {} result set: {}", query.getCommand(), series);
}
Map<String, DataTable> idMap = new HashMap<>();
final Map<String, DataTable> 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<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
List<String> ids = new ArrayList<>(pointOfTimes.size());
pointOfTimes.forEach(pointOfTime -> {
ids.add(pointOfTime.id(condition.getEntity().buildId()));
});
final List<String> ids = new ArrayList<>(pointOfTimes.size());
pointOfTimes.forEach(pointOfTime -> ids.add(pointOfTime.id(condition.getEntity().buildId())));
WhereQueryImpl<SelectQueryImpl> query = select()
final WhereQueryImpl<SelectQueryImpl> query = select()
.column(ID_COLUMN)
.column(valueColumnName)
.from(client.getDatabase(), condition.getName())
.where(contains(ID_COLUMN, Joiner.on("|").join(ids)));
Map<String, List<Long>> 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<Object> 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;
}
}
}

2
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/NetworkAddressAliasDAO.java
View File

@ -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;

22
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskLogQuery.java
View File

@ -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;
}
}

14
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileTaskQuery.java
View File

@ -52,7 +52,7 @@ public class ProfileTaskQuery implements IProfileTaskQueryDAO {
final Long startTimeBucket,
final Long endTimeBucket,
final Integer limit) throws IOException {
WhereQueryImpl<SelectQueryImpl> query =
final WhereQueryImpl<SelectQueryImpl> query =
select(
InfluxConstants.ID_COLUMN,
ProfileTaskRecord.SERVICE_ID,
@ -83,15 +83,13 @@ public class ProfileTaskQuery implements IProfileTaskQueryDAO {
query.limit(limit);
}
List<ProfileTask> tasks = Lists.newArrayList();
final List<ProfileTask> 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<Object> values) {
private static ProfileTask profileTaskBuilder(List<Object> values) {
return ProfileTask.builder()
.id((String) values.get(1))
.serviceId((String) values.get(2))

92
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java
View File

@ -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<BasicTrace> queryProfiledSegments(String taskId) throws IOException {
WhereQueryImpl query = select(ProfileThreadSnapshotRecord.SEGMENT_ID)
final WhereQueryImpl<SelectQueryImpl> 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<String> 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<SelectQueryImpl> 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<BasicTrace> 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<ProfileThreadSnapshotRecord> queryRecords(String segmentId, int minSequence,
int maxSequence) throws IOException {
WhereQueryImpl query = select(
WhereQueryImpl<SelectQueryImpl> 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<ProfileThreadSnapshotRecord> result = new ArrayList<>(maxSequence - minSequence);
final ArrayList<ProfileThreadSnapshotRecord> 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<QueryResult.Series> series = client.queryForSeries(query);
WhereQueryImpl<SelectQueryImpl> 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<QueryResult.Series> 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<Object> values = series.get(0).getValues().get(0);
SegmentRecord segmentRecord = new SegmentRecord();
final List<Object> 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<SelectQueryImpl> 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);
}
}

14
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopNRecordsQuery.java
View File

@ -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<SelectQueryImpl> 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<SelectedRecord> ASCENDING = (a, b) -> Long.compare(
Long.parseLong(a.getValue()), Long.parseLong(b.getValue()));
private static final Comparator<SelectedRecord> ASCENDING = Comparator.comparingLong(
a -> Long.parseLong(a.getValue()));
private static final Comparator<SelectedRecord> DESCENDING = (a, b) -> Long.compare(
Long.parseLong(b.getValue()), Long.parseLong(a.getValue()));

125
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TopologyQuery.java
View File

@ -54,11 +54,11 @@ public class TopologyQuery implements ITopologyQueryDAO {
}
@Override
public List<Call.CallDetail> loadServiceRelationsDetectedAtServerSide(long startTB,
long endTB,
List<String> serviceIds) throws IOException {
String measurement = ServiceRelationServerSideMetrics.INDEX_NAME;
WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = buildServiceCallsQuery(
public List<Call.CallDetail> loadServiceRelationsDetectedAtServerSide(final long startTB,
final long endTB,
final List<String> serviceIds) throws IOException {
final String measurement = ServiceRelationServerSideMetrics.INDEX_NAME;
final WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = buildServiceCallsQuery(
measurement,
startTB,
endTB,
@ -71,12 +71,11 @@ public class TopologyQuery implements ITopologyQueryDAO {
}
@Override
public List<Call.CallDetail> loadServiceRelationDetectedAtClientSide(long startTB,
long endTB,
public List<Call.CallDetail> loadServiceRelationDetectedAtClientSide(final long startTB,
final long endTB,
List<String> serviceIds) throws IOException {
String measurement = ServiceRelationClientSideMetrics.INDEX_NAME;
WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = buildServiceCallsQuery(
measurement,
final WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = buildServiceCallsQuery(
ServiceRelationClientSideMetrics.INDEX_NAME,
startTB,
endTB,
ServiceRelationServerSideMetrics.SOURCE_SERVICE_ID,
@ -87,11 +86,10 @@ public class TopologyQuery implements ITopologyQueryDAO {
}
@Override
public List<Call.CallDetail> loadServiceRelationsDetectedAtServerSide(long startTB,
long endTB) throws IOException {
String measurement = ServiceRelationServerSideMetrics.INDEX_NAME;
WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = buildServiceCallsQuery(
measurement,
public List<Call.CallDetail> loadServiceRelationsDetectedAtServerSide(final long startTB,
final long endTB) throws IOException {
final WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = buildServiceCallsQuery(
ServiceRelationServerSideMetrics.INDEX_NAME,
startTB,
endTB,
ServiceRelationServerSideMetrics.SOURCE_SERVICE_ID,
@ -102,11 +100,10 @@ public class TopologyQuery implements ITopologyQueryDAO {
}
@Override
public List<Call.CallDetail> loadServiceRelationDetectedAtClientSide(long startTB,
long endTB) throws IOException {
String tableName = ServiceRelationClientSideMetrics.INDEX_NAME;
public List<Call.CallDetail> loadServiceRelationDetectedAtClientSide(final long startTB,
final long endTB) throws IOException {
WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, 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<Call.CallDetail> loadInstanceRelationDetectedAtServerSide(String clientServiceId,
String serverServiceId,
long startTB,
long endTB) throws IOException {
String measurement = ServiceInstanceRelationServerSideMetrics.INDEX_NAME;
public List<Call.CallDetail> loadInstanceRelationDetectedAtServerSide(final String clientServiceId,
final String serverServiceId,
final long startTB,
final long endTB) throws IOException {
WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, 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<Call.CallDetail> loadInstanceRelationDetectedAtClientSide(String clientServiceId,
String serverServiceId,
long startTB,
long endTB) throws IOException {
String measurement = ServiceInstanceRelationClientSideMetrics.INDEX_NAME;
public List<Call.CallDetail> loadInstanceRelationDetectedAtClientSide(final String clientServiceId,
final String serverServiceId,
final long startTB,
final long endTB) throws IOException {
WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, 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<Call.CallDetail> loadEndpointRelation(long startTB,
long endTB,
String destEndpointId) throws IOException {
String measurement = EndpointRelationServerSideMetrics.INDEX_NAME;
WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = buildServiceCallsQuery(
measurement,
public List<Call.CallDetail> loadEndpointRelation(final long startTB,
final long endTB,
final String destEndpointId) throws IOException {
final WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, 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<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery2 = buildServiceCallsQuery(
measurement,
final WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, 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<Call.CallDetail> calls = buildEndpointCalls(buildQuery(subQuery), DetectPoint.SERVER);
final List<Call.CallDetail> calls = buildEndpointCalls(buildQuery(subQuery), DetectPoint.SERVER);
calls.addAll(buildEndpointCalls(buildQuery(subQuery), DetectPoint.CLIENT));
return calls;
}
private WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> buildServiceCallsQuery(
String measurement,
long startTB,
long endTB,
String sourceCName,
String destCName,
List<String> serviceIds) {
WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = select()
final String measurement,
final long startTB,
final long endTB,
final String sourceCName,
final String destCName,
final List<String> serviceIds) {
final WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, 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<WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, 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<SelectSubQueryImpl<SelectQueryImpl>, 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<SelectSubQueryImpl<SelectQueryImpl>, SelectQueryImpl> subQuery = select()
final WhereSubQueryImpl<SelectSubQueryImpl<SelectQueryImpl>, 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<Call.CallDetail> 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<Call.CallDetail> calls = new ArrayList<>();
final List<Call.CallDetail> 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<Call.CallDetail> buildInstanceCalls(Query query,
DetectPoint detectPoint) throws IOException {
private List<Call.CallDetail> buildInstanceCalls(Query query, DetectPoint detectPoint) throws IOException {
QueryResult.Series series = client.queryForSingleSeries(query);
if (log.isDebugEnabled()) {

35
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java
View File

@ -178,23 +178,24 @@ public class TraceQuery implements ITraceQueryDAO {
@Override
public List<SegmentRecord> 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<QueryResult.Series> series = client.queryForSeries(query);
WhereQueryImpl<SelectQueryImpl> 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<QueryResult.Series> 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<Span> doFlexibleTraceQuery(String traceId) throws IOException {
public List<Span> doFlexibleTraceQuery(String traceId) {
return Collections.emptyList();
}
}

36
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/UITemplateManagementDAOImpl.java
View File

@ -50,9 +50,9 @@ public class UITemplateManagementDAOImpl implements UITemplateManagementDAO {
@Override
public List<DashboardConfiguration> getAllTemplates(final Boolean includingDisabled) throws IOException {
WhereQueryImpl<SelectQueryImpl> where = select().raw("*::field")
.from(client.getDatabase(), UITemplate.INDEX_NAME)
.where();
final WhereQueryImpl<SelectQueryImpl> 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 {