From 4e63331d357f2e27f91d2174527b58b738ae4778 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=AD=E5=8B=87=E5=8D=87=20pengys?= Date: Wed, 12 Jun 2019 18:56:11 +0800 Subject: [PATCH] #2860 (#2861) Get the document by index name. --- .../skywalking/oap/server/core/query/ID.java | 49 +++++++++++++++++++ .../server/core/query/MetricQueryService.java | 13 +++-- .../core/storage/query/IMetricsQueryDAO.java | 5 +- .../elasticsearch/ElasticSearchClient.java | 7 --- .../elasticsearch/ITElasticSearchClient.java | 8 +-- .../receiver/trace/mock/AgentDataMock.java | 4 +- .../elasticsearch/base/MetricsEsDAO.java | 3 +- .../elasticsearch/base/TimeSeriesUtils.java | 32 ++++++------ .../query/MetricsQueryEsDAO.java | 31 ++++++------ .../plugin/jdbc/h2/dao/H2MetricsQueryDAO.java | 17 ++++--- 10 files changed, 110 insertions(+), 59 deletions(-) create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/ID.java diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/ID.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/ID.java new file mode 100644 index 000000000..fd8160a86 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/ID.java @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.skywalking.oap.server.core.query; + +import lombok.Getter; +import org.apache.skywalking.oap.server.core.Const; + +/** + * @author peng-yongsheng + */ +@Getter +public class ID { + + private final long timeBucket; + private final String entityId; + + public ID(long timeBucket, String entityId) { + this.timeBucket = timeBucket; + this.entityId = entityId; + } + + public ID(long timeBucket) { + this.timeBucket = timeBucket; + this.entityId = Const.EMPTY_STRING; + } + + @Override public String toString() { + if (entityId.length() > 0) { + return timeBucket + Const.ID_SPLIT + entityId; + } else { + return String.valueOf(timeBucket); + } + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java index 40c6fc7ba..016bcd601 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java @@ -22,7 +22,6 @@ import java.io.IOException; import java.text.ParseException; import java.util.*; import org.apache.skywalking.apm.util.StringUtil; -import org.apache.skywalking.oap.server.core.Const; import org.apache.skywalking.oap.server.core.analysis.Downsampling; import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics; import org.apache.skywalking.oap.server.core.query.entity.*; @@ -71,11 +70,11 @@ public class MetricQueryService implements Service { public IntValues getLinearIntValues(final String indName, final String id, final Downsampling downsampling, final long startTB, final long endTB) throws IOException, ParseException { List durationPoints = DurationUtils.INSTANCE.getDurationPoints(downsampling, startTB, endTB); - List ids = new ArrayList<>(); + List ids = new ArrayList<>(); if (StringUtil.isEmpty(id)) { - durationPoints.forEach(durationPoint -> ids.add(String.valueOf(durationPoint.getPoint()))); + durationPoints.forEach(durationPoint -> ids.add(new ID(durationPoint.getPoint()))); } else { - durationPoints.forEach(durationPoint -> ids.add(durationPoint.getPoint() + Const.ID_SPLIT + id)); + durationPoints.forEach(durationPoint -> ids.add(new ID(durationPoint.getPoint(), id))); } return getMetricQueryDAO().getLinearIntValues(indName, downsampling, ids, ValueColumnIds.INSTANCE.getValueCName(indName)); @@ -84,12 +83,12 @@ public class MetricQueryService implements Service { public Thermodynamic getThermodynamic(final String indName, final String id, final Downsampling downsampling, final long startTB, final long endTB) throws IOException, ParseException { List durationPoints = DurationUtils.INSTANCE.getDurationPoints(downsampling, startTB, endTB); - List ids = new ArrayList<>(); + List ids = new ArrayList<>(); durationPoints.forEach(durationPoint -> { if (id == null) { - ids.add(durationPoint.getPoint() + ""); + ids.add(new ID(durationPoint.getPoint())); } else { - ids.add(durationPoint.getPoint() + Const.ID_SPLIT + id); + ids.add(new ID(durationPoint.getPoint(), id)); } }); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IMetricsQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IMetricsQueryDAO.java index ff92d41bd..9da99a110 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IMetricsQueryDAO.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IMetricsQueryDAO.java @@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.core.storage.query; import java.io.IOException; import java.util.List; import org.apache.skywalking.oap.server.core.analysis.Downsampling; +import org.apache.skywalking.oap.server.core.query.ID; import org.apache.skywalking.oap.server.core.query.entity.*; import org.apache.skywalking.oap.server.core.query.sql.*; import org.apache.skywalking.oap.server.core.storage.DAO; @@ -32,7 +33,7 @@ public interface IMetricsQueryDAO extends DAO { IntValues getValues(String indName, Downsampling downsampling, long startTB, long endTB, Where where, String valueCName, Function function) throws IOException; - IntValues getLinearIntValues(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException; + IntValues getLinearIntValues(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException; - Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException; + Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException; } diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java index f37d0f65e..1e9a5794f 100644 --- a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java +++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java @@ -216,13 +216,6 @@ public class ElasticSearchClient implements Client { return client.get(request); } - public MultiGetResponse multiGet(String indexName, List ids) throws IOException { - final String newIndexName = formatIndexName(indexName); - MultiGetRequest request = new MultiGetRequest(); - ids.forEach(id -> request.add(newIndexName, TYPE, id)); - return client.multiGet(request); - } - public void forceInsert(String indexName, String id, XContentBuilder source) throws IOException { IndexRequest request = prepareInsert(indexName, id, source); request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); diff --git a/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ITElasticSearchClient.java b/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ITElasticSearchClient.java index a89913377..3ba3b03f2 100644 --- a/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ITElasticSearchClient.java +++ b/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ITElasticSearchClient.java @@ -23,7 +23,7 @@ import java.io.IOException; import java.util.*; import java.util.concurrent.TimeUnit; import org.elasticsearch.action.bulk.BulkProcessor; -import org.elasticsearch.action.get.*; +import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.common.xcontent.*; @@ -113,12 +113,6 @@ public class ITElasticSearchClient { Assert.assertEquals("pengys", response.getSource().get("user")); Assert.assertEquals("trying out Elasticsearch", response.getSource().get("message")); - List ids = new ArrayList<>(); - ids.add(id); - MultiGetResponse responses = client.multiGet(indexName, ids); - Assert.assertEquals(1, responses.getResponses().length); - Assert.assertEquals("pengys", responses.getResponses()[0].getResponse().getSource().get("user")); - SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.termQuery("user", "pengys")); SearchResponse searchResponse = client.search(indexName, sourceBuilder); diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java index 0883f4705..ebf0bb780 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java @@ -22,6 +22,7 @@ import io.grpc.*; import io.grpc.stub.StreamObserver; import java.util.concurrent.TimeUnit; import org.apache.skywalking.apm.network.language.agent.*; +import org.joda.time.DateTime; /** * @author peng-yongsheng @@ -38,7 +39,8 @@ public class AgentDataMock { StreamObserver streamObserver = createStreamObserver(); UniqueId.Builder globalTraceId = UniqueIdBuilder.INSTANCE.create(); - long startTimestamp = System.currentTimeMillis(); +// long startTimestamp = System.currentTimeMillis(); + long startTimestamp = new DateTime().minusDays(2).getMillis(); // ServiceAMock ServiceAMock serviceAMock = new ServiceAMock(registerMock); diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java index 58a42c76d..82bed926d 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java @@ -41,7 +41,8 @@ public class MetricsEsDAO extends EsDAO implements IMetricsDAO ids, String valueCName) throws IOException { + @Override public IntValues getLinearIntValues(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException { String indexName = ModelName.build(downsampling, indName); - MultiGetResponse response = getClient().multiGet(indexName, ids); - IntValues intValues = new IntValues(); - for (MultiGetItemResponse itemResponse : response.getResponses()) { + for (ID id : ids) { + String modelName = TimeSeriesUtils.timeSeries(indexName, id.getTimeBucket(), downsampling); + GetResponse response = getClient().get(modelName, id.toString()); KVInt kvInt = new KVInt(); - kvInt.setId(itemResponse.getId()); + kvInt.setId(response.getId()); kvInt.setValue(0); - Map source = itemResponse.getResponse().getSource(); + Map source = response.getSource(); if (source != null) { kvInt.setValue(((Number)source.getOrDefault(valueCName, 0)).longValue()); } intValues.getValues().add(kvInt); } + return intValues; } - @Override public Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException { + @Override public Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException { String indexName = ModelName.build(downsampling, indName); - MultiGetResponse response = getClient().multiGet(indexName, ids); - Thermodynamic thermodynamic = new Thermodynamic(); List> thermodynamicValueMatrix = new ArrayList<>(); int numOfSteps = 0; - for (MultiGetItemResponse itemResponse : response.getResponses()) { - Map source = itemResponse.getResponse().getSource(); + for (ID id : ids) { + String modelName = TimeSeriesUtils.timeSeries(indexName, id.getTimeBucket(), downsampling); + GetResponse response = getClient().get(modelName, id.toString()); + + Map source = response.getSource(); if (source == null) { // add empty list to represent no data exist for this time bucket thermodynamicValueMatrix.add(new ArrayList<>()); diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricsQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricsQueryDAO.java index 7abb9d110..1c6c2792b 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricsQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricsQueryDAO.java @@ -23,6 +23,7 @@ import java.sql.*; import java.util.*; import org.apache.skywalking.oap.server.core.analysis.Downsampling; import org.apache.skywalking.oap.server.core.analysis.metrics.*; +import org.apache.skywalking.oap.server.core.query.ID; import org.apache.skywalking.oap.server.core.query.entity.*; import org.apache.skywalking.oap.server.core.query.sql.*; import org.apache.skywalking.oap.server.core.storage.model.ModelName; @@ -33,6 +34,7 @@ import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariC * @author wusheng */ public class H2MetricsQueryDAO extends H2SQLExecutor implements IMetricsQueryDAO { + private JDBCHikariCPClient h2Client; public H2MetricsQueryDAO(JDBCHikariCPClient h2Client) { @@ -73,7 +75,7 @@ public class H2MetricsQueryDAO extends H2SQLExecutor implements IMetricsQueryDAO ids.add(id); valueCollection.append("'").append(id).append("'"); } - whereSql.append(keyValues.getKey()).append(" in (" + valueCollection + ")"); + whereSql.append(keyValues.getKey()).append(" in (").append(valueCollection).append(")"); } whereSql.append(") and "); } @@ -99,8 +101,7 @@ public class H2MetricsQueryDAO extends H2SQLExecutor implements IMetricsQueryDAO return orderWithDefault0(intValues, ids); } - @Override public IntValues getLinearIntValues(String indName, Downsampling downsampling, List ids, - String valueCName) throws IOException { + @Override public IntValues getLinearIntValues(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException { String tableName = ModelName.build(downsampling, indName); StringBuilder idValues = new StringBuilder(); @@ -125,7 +126,12 @@ public class H2MetricsQueryDAO extends H2SQLExecutor implements IMetricsQueryDAO } catch (SQLException e) { throw new IOException(e); } - return orderWithDefault0(intValues, ids); + + List idList = new ArrayList<>(); + for (ID id : ids) { + idList.add(id.toString()); + } + return orderWithDefault0(intValues, idList); } /** @@ -148,8 +154,7 @@ public class H2MetricsQueryDAO extends H2SQLExecutor implements IMetricsQueryDAO return intValues; } - @Override public Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List ids, - String valueCName) throws IOException { + @Override public Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List ids, String valueCName) throws IOException { String tableName = ModelName.build(downsampling, indName); StringBuilder idValues = new StringBuilder();