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

#2860 (#2861) 

Get the document by index name.
This commit is contained in:
彭勇升 pengys 2019-06-12 18:56:11 +08:00 committed by 吴晟 Wu Sheng
parent dbb8bd8a7f
commit 4e63331d35
10 changed files with 110 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/ID.java Normal file
View File

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

13
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java
View File

@ -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<DurationPoint> durationPoints = DurationUtils.INSTANCE.getDurationPoints(downsampling, startTB, endTB);
List<String> ids = new ArrayList<>();
List<ID> 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<DurationPoint> durationPoints = DurationUtils.INSTANCE.getDurationPoints(downsampling, startTB, endTB);
List<String> ids = new ArrayList<>();
List<ID> 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));
}
});

5
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IMetricsQueryDAO.java
View File

@ -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<String> ids, String valueCName) throws IOException;
IntValues getLinearIntValues(String indName, Downsampling downsampling, List<ID> ids, String valueCName) throws IOException;
Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List<String> ids, String valueCName) throws IOException;
Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List<ID> ids, String valueCName) throws IOException;
}

7
oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java
View File

@ -216,13 +216,6 @@ public class ElasticSearchClient implements Client {
return client.get(request);
}
public MultiGetResponse multiGet(String indexName, List<String> 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);

8
oap-server/server-library/library-client/src/test/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ITElasticSearchClient.java
View File

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

4
oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java
View File

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

3
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java
View File

@ -41,7 +41,8 @@ public class MetricsEsDAO extends EsDAO implements IMetricsDAO<IndexRequest, Upd
}
@Override public Metrics get(Model model, Metrics metrics) throws IOException {
GetResponse response = getClient().get(model.getName(), metrics.id());
String modelName = TimeSeriesUtils.timeSeries(model, metrics.getTimeBucket());
GetResponse response = getClient().get(modelName, metrics.id());
if (response.isExists()) {
return storageBuilder.map2Data(response.getSource());
} else {

32
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtils.java
View File

@ -18,36 +18,40 @@
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
import org.apache.skywalking.oap.server.core.analysis.*;
import org.apache.skywalking.oap.server.core.storage.model.Model;
/**
* @author peng-yongsheng
*/
class TimeSeriesUtils {
public class TimeSeriesUtils {
static String timeSeries(Model model) {
long timeBucket = TimeBucket.getTimeBucket(System.currentTimeMillis(), model.getDownsampling());
return timeSeries(model, timeBucket);
}
public static String timeSeries(String modelName, long timeBucket, Downsampling downsampling) {
switch (downsampling) {
case None:
return modelName;
case Hour:
return modelName + Const.LINE + timeBucket / 100;
case Minute:
return modelName + Const.LINE + timeBucket / 10000;
case Second:
return modelName + Const.LINE + timeBucket / 1000000;
default:
return modelName + Const.LINE + timeBucket;
}
}
static String timeSeries(Model model, long timeBucket) {
if (!model.isCapableOfTimeSeries()) {
return model.getName();
}
switch (model.getDownsampling()) {
case None:
return model.getName();
case Hour:
return model.getName() + Const.LINE + timeBucket / 100;
case Minute:
return model.getName() + Const.LINE + timeBucket / 10000;
case Second:
return model.getName() + Const.LINE + timeBucket / 1000000;
default:
return model.getName() + Const.LINE + timeBucket;
}
return timeSeries(model.getName(), timeBucket, model.getDownsampling());
}
static long indexTimeSeries(String indexName) {

31
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetricsQueryEsDAO.java
View File

@ -22,13 +22,14 @@ import java.io.IOException;
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;
import org.apache.skywalking.oap.server.core.storage.query.IMetricsQueryDAO;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
import org.elasticsearch.action.get.*;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.*;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.*;
@ -62,7 +63,7 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
IntValues intValues = new IntValues();
Terms idTerms = response.getAggregations().get(Metrics.ENTITY_ID);
for (Terms.Bucket idBucket : idTerms.getBuckets()) {
long value = 0;
long value;
switch (function) {
case Sum:
Sum sum = idBucket.getAggregations().get(valueCName);
@ -100,37 +101,39 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
}
}
@Override public IntValues getLinearIntValues(String indName, Downsampling downsampling, List<String> ids, String valueCName) throws IOException {
@Override public IntValues getLinearIntValues(String indName, Downsampling downsampling, List<ID> 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<String, Object> source = itemResponse.getResponse().getSource();
Map<String, Object> 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<String> ids, String valueCName) throws IOException {
@Override public Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List<ID> ids, String valueCName) throws IOException {
String indexName = ModelName.build(downsampling, indName);
MultiGetResponse response = getClient().multiGet(indexName, ids);
Thermodynamic thermodynamic = new Thermodynamic();
List<List<Long>> thermodynamicValueMatrix = new ArrayList<>();
int numOfSteps = 0;
for (MultiGetItemResponse itemResponse : response.getResponses()) {
Map<String, Object> source = itemResponse.getResponse().getSource();
for (ID id : ids) {
String modelName = TimeSeriesUtils.timeSeries(indexName, id.getTimeBucket(), downsampling);
GetResponse response = getClient().get(modelName, id.toString());
Map<String, Object> source = response.getSource();
if (source == null) {
// add empty list to represent no data exist for this time bucket
thermodynamicValueMatrix.add(new ArrayList<>());

17
oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricsQueryDAO.java
View File

@ -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<String> ids,
String valueCName) throws IOException {
@Override public IntValues getLinearIntValues(String indName, Downsampling downsampling, List<ID> 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<String> 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<String> ids,
String valueCName) throws IOException {
@Override public Thermodynamic getThermodynamic(String indName, Downsampling downsampling, List<ID> ids, String valueCName) throws IOException {
String tableName = ModelName.build(downsampling, indName);
StringBuilder idValues = new StringBuilder();