From 53b9e7f828adab4eb773bfc81cd2d024feabf7b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=AD=E5=8B=87=E5=8D=87=20pengys?= <8082209@qq.com> Date: Tue, 25 Sep 2018 19:14:37 +0800 Subject: [PATCH] Implementation of aggregation query. (#1702) --- .../oap/server/core/CoreModule.java | 1 + .../oap/server/core/CoreModuleProvider.java | 1 + .../core/query/AggregationQueryService.java | 110 ++++++++++++++ .../oap/server/core/query/entity}/Order.java | 2 +- .../server/core/query/entity}/TopNEntity.java | 6 +- .../server/core/storage/StorageModule.java | 2 +- .../storage/TimePyramidTableNameBuilder.java | 3 - .../storage/query/IAggregationQueryDAO.java | 45 ++++++ .../query/graphql/GraphQLQueryProvider.java | 2 +- .../graphql/resolver/AggregationQuery.java | 64 +++++++- .../query/graphql/resolver/AlarmQuery.java | 1 + .../oap/query/graphql/type/AlarmMessage.java | 2 + .../oap/query/graphql/type/Scope.java | 28 ---- .../oap/query/graphql/type/TopNCondition.java | 30 ---- .../src/main/resources/query-protocol | 2 +- .../endpoint/MultiScopesSpanListener.java | 3 - .../StorageModuleElasticsearchProvider.java | 1 + .../query/AggregationQueryEsDAO.java | 137 ++++++++++++++++++ 18 files changed, 367 insertions(+), 73 deletions(-) create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/AggregationQueryService.java rename oap-server/{server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type => server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity}/Order.java (93%) rename oap-server/{server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type => server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity}/TopNEntity.java (90%) create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IAggregationQueryDAO.java delete mode 100644 oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/Scope.java delete mode 100644 oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/TopNCondition.java create mode 100644 oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java index bf1f666c1..eab389bd9 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java @@ -61,6 +61,7 @@ public class CoreModule extends ModuleDefine { classes.add(MetricQueryService.class); classes.add(TraceQueryService.class); classes.add(MetadataQueryService.class); + classes.add(AggregationQueryService.class); } private void addServerInterface(List classes) { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java index 4dc457321..600577ca0 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java @@ -114,6 +114,7 @@ public class CoreModuleProvider extends ModuleProvider { this.registerServiceImplementation(MetricQueryService.class, new MetricQueryService(getManager())); this.registerServiceImplementation(TraceQueryService.class, new TraceQueryService(getManager())); this.registerServiceImplementation(MetadataQueryService.class, new MetadataQueryService(getManager())); + this.registerServiceImplementation(AggregationQueryService.class, new AggregationQueryService(getManager())); annotationScan.registerListener(storageAnnotationListener); annotationScan.registerListener(streamAnnotationListener); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/AggregationQueryService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/AggregationQueryService.java new file mode 100644 index 000000000..f4873681a --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/AggregationQueryService.java @@ -0,0 +1,110 @@ +/* + * 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 java.io.IOException; +import java.util.List; +import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.cache.*; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.core.register.*; +import org.apache.skywalking.oap.server.core.storage.StorageModule; +import org.apache.skywalking.oap.server.core.storage.query.IAggregationQueryDAO; +import org.apache.skywalking.oap.server.library.module.*; +import org.apache.skywalking.oap.server.library.module.Service; + +/** + * @author peng-yongsheng + */ +public class AggregationQueryService implements Service { + + private final ModuleManager moduleManager; + private IAggregationQueryDAO aggregationQueryDAO; + + public AggregationQueryService(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + private IAggregationQueryDAO getAggregationQueryDAO() { + if (aggregationQueryDAO == null) { + aggregationQueryDAO = moduleManager.find(StorageModule.NAME).getService(IAggregationQueryDAO.class); + } + return aggregationQueryDAO; + } + + public List getServiceTopN(final String name, final int topN, final Step step, final long startTB, + final long endTB, final Order order) throws IOException { + List topNEntities = getAggregationQueryDAO().getServiceTopN(name, topN, step, startTB, endTB, order); + for (TopNEntity entity : topNEntities) { + ServiceInventory inventory = moduleManager.find(CoreModule.NAME).getService(ServiceInventoryCache.class).get(Integer.valueOf(entity.getId())); + if (inventory != null) { + entity.setName(inventory.getName()); + } + } + return topNEntities; + } + + public List getAllServiceInstanceTopN(final String name, final int topN, final Step step, + final long startTB, final long endTB, final Order order) throws IOException { + List topNEntities = getAggregationQueryDAO().getAllServiceInstanceTopN(name, topN, step, startTB, endTB, order); + for (TopNEntity entity : topNEntities) { + ServiceInstanceInventory inventory = moduleManager.find(CoreModule.NAME).getService(ServiceInstanceInventoryCache.class).get(Integer.valueOf(entity.getId())); + if (inventory != null) { + entity.setName(inventory.getName()); + } + } + return topNEntities; + } + + public List getServiceInstanceTopN(final int serviceId, final String name, final int topN, + final Step step, final long startTB, final long endTB, final Order order) throws IOException { + List topNEntities = getAggregationQueryDAO().getServiceInstanceTopN(serviceId, name, topN, step, startTB, endTB, order); + for (TopNEntity entity : topNEntities) { + ServiceInstanceInventory inventory = moduleManager.find(CoreModule.NAME).getService(ServiceInstanceInventoryCache.class).get(Integer.valueOf(entity.getId())); + if (inventory != null) { + entity.setName(inventory.getName()); + } + } + return topNEntities; + } + + public List getAllEndpointTopN(final String name, final int topN, final Step step, + final long startTB, final long endTB, final Order order) throws IOException { + List topNEntities = getAggregationQueryDAO().getAllEndpointTopN(name, topN, step, startTB, endTB, order); + for (TopNEntity entity : topNEntities) { + EndpointInventory inventory = moduleManager.find(CoreModule.NAME).getService(EndpointInventoryCache.class).get(Integer.valueOf(entity.getId())); + if (inventory != null) { + entity.setName(inventory.getName()); + } + } + return topNEntities; + } + + public List getEndpointTopN(final int serviceId, final String name, final int topN, + final Step step, final long startTB, final long endTB, final Order order) throws IOException { + List topNEntities = getAggregationQueryDAO().getEndpointTopN(serviceId, name, topN, step, startTB, endTB, order); + for (TopNEntity entity : topNEntities) { + EndpointInventory inventory = moduleManager.find(CoreModule.NAME).getService(EndpointInventoryCache.class).get(Integer.valueOf(entity.getId())); + if (inventory != null) { + entity.setName(inventory.getName()); + } + } + return topNEntities; + } +} diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/Order.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Order.java similarity index 93% rename from oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/Order.java rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Order.java index 3f7cebe25..fe562c1af 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/Order.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Order.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.oap.query.graphql.type; +package org.apache.skywalking.oap.server.core.query.entity; public enum Order { ASC, diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/TopNEntity.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/TopNEntity.java similarity index 90% rename from oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/TopNEntity.java rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/TopNEntity.java index f3f76583a..411b9e0d3 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/TopNEntity.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/TopNEntity.java @@ -16,8 +16,12 @@ * */ -package org.apache.skywalking.oap.query.graphql.type; +package org.apache.skywalking.oap.server.core.query.entity; +import lombok.*; + +@Getter +@Setter public class TopNEntity { private String name; private String id; diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java index ccaa198ee..424b23564 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java @@ -38,6 +38,6 @@ public class StorageModule extends ModuleDefine { IBatchDAO.class, StorageDAO.class, IRegisterLockDAO.class, IServiceInventoryCacheDAO.class, IServiceInstanceInventoryCacheDAO.class, IEndpointInventoryCacheDAO.class, INetworkAddressInventoryCacheDAO.class, - ITopologyQueryDAO.class, IMetricQueryDAO.class, ITraceQueryDAO.class, IMetadataQueryDAO.class}; + ITopologyQueryDAO.class, IMetricQueryDAO.class, ITraceQueryDAO.class, IMetadataQueryDAO.class, IAggregationQueryDAO.class}; } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/TimePyramidTableNameBuilder.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/TimePyramidTableNameBuilder.java index b413e9ad5..f7f0708a1 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/TimePyramidTableNameBuilder.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/TimePyramidTableNameBuilder.java @@ -40,9 +40,6 @@ public class TimePyramidTableNameBuilder { case HOUR: tableName = tableName + Const.ID_SPLIT + TimePyramid.Hour.getName(); break; - case MINUTE: - tableName = tableName + Const.ID_SPLIT + TimePyramid.Minute.getName(); - break; } return tableName; } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IAggregationQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IAggregationQueryDAO.java new file mode 100644 index 000000000..1310c4e3f --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IAggregationQueryDAO.java @@ -0,0 +1,45 @@ +/* + * 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.storage.query; + +import java.io.IOException; +import java.util.List; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.core.storage.DAO; + +/** + * @author peng-yongsheng + */ +public interface IAggregationQueryDAO extends DAO { + + List getServiceTopN(final String name, final int topN, final Step step, final long startTB, + final long endTB, final Order order) throws IOException; + + List getAllServiceInstanceTopN(final String name, final int topN, final Step step, + final long startTB, final long endTB, final Order order) throws IOException; + + List getServiceInstanceTopN(final int serviceId, final String name, final int topN, + final Step step, final long startTB, final long endTB, final Order order) throws IOException; + + List getAllEndpointTopN(final String name, final int topN, final Step step, + final long startTB, final long endTB, final Order order) throws IOException; + + List getEndpointTopN(final int serviceId, final String name, final int topN, + final Step step, final long startTB, final long endTB, final Order order) throws IOException; +} diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/GraphQLQueryProvider.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/GraphQLQueryProvider.java index 9722cae31..840ef4295 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/GraphQLQueryProvider.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/GraphQLQueryProvider.java @@ -63,7 +63,7 @@ public class GraphQLQueryProvider extends ModuleProvider { .file("query-protocol/trace.graphqls") .resolvers(new TraceQuery(getManager())) .file("query-protocol/aggregation.graphqls") - .resolvers(new AggregationQuery()) + .resolvers(new AggregationQuery(getManager())) .file("query-protocol/alarm.graphqls") .resolvers(new AlarmQuery()) .build() diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AggregationQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AggregationQuery.java index 90fab3e8b..47812c3a2 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AggregationQuery.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AggregationQuery.java @@ -19,11 +19,67 @@ package org.apache.skywalking.oap.query.graphql.resolver; import com.coxautodev.graphql.tools.GraphQLQueryResolver; -import java.util.*; -import org.apache.skywalking.oap.query.graphql.type.*; +import java.io.IOException; +import java.util.List; +import org.apache.skywalking.oap.query.graphql.type.Duration; +import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.query.*; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.library.module.ModuleManager; public class AggregationQuery implements GraphQLQueryResolver { - public List getTopN(final TopNCondition condition, final Duration duration) { - return Collections.emptyList(); + + private final ModuleManager moduleManager; + private AggregationQueryService queryService; + + public AggregationQuery(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + private AggregationQueryService getQueryService() { + if (queryService == null) { + this.queryService = moduleManager.find(CoreModule.NAME).getService(AggregationQueryService.class); + } + return queryService; + } + + public List getServiceTopN(final String name, final int topN, final Duration duration, + final Order order) throws IOException { + long startTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getStart()); + long endTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getEnd()); + + return getQueryService().getServiceTopN(name, topN, duration.getStep(), startTimeBucket, endTimeBucket, order); + } + + public List getAllServiceInstanceTopN(final String name, final int topN, final Duration duration, + final Order order) throws IOException { + long startTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getStart()); + long endTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getEnd()); + + return getQueryService().getAllServiceInstanceTopN(name, topN, duration.getStep(), startTimeBucket, endTimeBucket, order); + } + + public List getServiceInstanceTopN(final int serviceId, final String name, final int topN, + final Duration duration, final Order order) throws IOException { + long startTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getStart()); + long endTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getEnd()); + + return getQueryService().getServiceInstanceTopN(serviceId, name, topN, duration.getStep(), startTimeBucket, endTimeBucket, order); + } + + public List getAllEndpointTopN(final String name, final int topN, + final Duration duration, final Order order) throws IOException { + long startTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getStart()); + long endTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getEnd()); + + return getQueryService().getAllEndpointTopN(name, topN, duration.getStep(), startTimeBucket, endTimeBucket, order); + } + + public List getEndpointTopN(final int serviceId, final String name, final int topN, + final Duration duration, final Order order) throws IOException { + long startTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getStart()); + long endTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getEnd()); + + return getQueryService().getEndpointTopN(serviceId, name, topN, duration.getStep(), startTimeBucket, endTimeBucket, order); } } diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java index 06e094ff2..20b570346 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java @@ -21,6 +21,7 @@ package org.apache.skywalking.oap.query.graphql.resolver; import com.coxautodev.graphql.tools.GraphQLQueryResolver; import org.apache.skywalking.oap.query.graphql.type.*; import org.apache.skywalking.oap.server.core.query.entity.Pagination; +import org.apache.skywalking.oap.server.core.source.Scope; public class AlarmQuery implements GraphQLQueryResolver { public AlarmTrend getAlarmTrend(final Duration duration) { diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/AlarmMessage.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/AlarmMessage.java index 900856296..e79e93831 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/AlarmMessage.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/AlarmMessage.java @@ -18,6 +18,8 @@ package org.apache.skywalking.oap.query.graphql.type; +import org.apache.skywalking.oap.server.core.source.Scope; + public class AlarmMessage { private Scope scope; private String id; diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/Scope.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/Scope.java deleted file mode 100644 index 45d5a427b..000000000 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/Scope.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * 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.query.graphql.type; - -public enum Scope { - SERVICE, - SERVICE_INSTANCE, - ENDPOINT, - SERVICE_RELATION, - SERVICE_INSTANCE_RELATION, - ENDPOINT_RELATION -} diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/TopNCondition.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/TopNCondition.java deleted file mode 100644 index a3dd5c0e2..000000000 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/TopNCondition.java +++ /dev/null @@ -1,30 +0,0 @@ -/* - * 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.query.graphql.type; - -import lombok.Getter; - -@Getter -public class TopNCondition { - private String name; - private int topN; - private Order order; - private Scope filterScope; - private int filterId; -} diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol index f2e54c2cd..d71d3f183 160000 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol @@ -1 +1 @@ -Subproject commit f2e54c2cd3f7fdb2cdc975cf791e1bb1d9aab96e +Subproject commit d71d3f183a1e498aafad8c59c8d3373408140ac9 diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java index ff083b41d..1854533a2 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java @@ -180,9 +180,6 @@ public class MultiScopesSpanListener implements EntrySpanListener, ExitSpanListe exitSourceBuilder.setSourceEndpointName(endpointInventoryCache.get(exitSourceBuilder.getSourceEndpointId()).getName()); exitSourceBuilder.setTimeBucket(minuteTimeBucket); - sourceReceiver.receive(exitSourceBuilder.toService()); - sourceReceiver.receive(exitSourceBuilder.toServiceInstance()); - sourceReceiver.receive(exitSourceBuilder.toEndpoint()); sourceReceiver.receive(exitSourceBuilder.toServiceRelation()); sourceReceiver.receive(exitSourceBuilder.toServiceInstanceRelation()); sourceReceiver.receive(exitSourceBuilder.toEndpointRelation()); diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java index 39128a198..83e85c47f 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java @@ -79,6 +79,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider { this.registerServiceImplementation(IMetricQueryDAO.class, new MetricQueryEsDAO(elasticSearchClient)); this.registerServiceImplementation(ITraceQueryDAO.class, new TraceQueryEsDAO(elasticSearchClient)); this.registerServiceImplementation(IMetadataQueryDAO.class, new MetadataQueryEsDAO(elasticSearchClient)); + this.registerServiceImplementation(IAggregationQueryDAO.class, new AggregationQueryEsDAO(elasticSearchClient)); } @Override diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java new file mode 100644 index 000000000..2868da61b --- /dev/null +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java @@ -0,0 +1,137 @@ +/* + * 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.storage.plugin.elasticsearch.query; + +import java.io.IOException; +import java.util.*; +import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.core.register.*; +import org.apache.skywalking.oap.server.core.storage.TimePyramidTableNameBuilder; +import org.apache.skywalking.oap.server.core.storage.query.IAggregationQueryDAO; +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.search.SearchResponse; +import org.elasticsearch.index.query.*; +import org.elasticsearch.search.aggregations.*; +import org.elasticsearch.search.aggregations.bucket.terms.*; +import org.elasticsearch.search.aggregations.metrics.avg.Avg; +import org.elasticsearch.search.builder.SearchSourceBuilder; + +/** + * @author peng-yongsheng + */ +public class AggregationQueryEsDAO extends EsDAO implements IAggregationQueryDAO { + + public AggregationQueryEsDAO(ElasticSearchClient client) { + super(client); + } + + @Override + public List getServiceTopN(String name, int topN, Step step, long startTB, + long endTB, Order order) throws IOException { + String indexName = TimePyramidTableNameBuilder.build(step, name); + + SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); + sourceBuilder.query(QueryBuilders.rangeQuery(Indicator.TIME_BUCKET).lte(endTB).gte(startTB)); + return aggregation(indexName, sourceBuilder, topN, order); + } + + @Override public List getAllServiceInstanceTopN(String name, int topN, Step step, + long startTB, long endTB, Order order) throws IOException { + String indexName = TimePyramidTableNameBuilder.build(step, name); + + SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); + sourceBuilder.query(QueryBuilders.rangeQuery(Indicator.TIME_BUCKET).lte(endTB).gte(startTB)); + return aggregation(indexName, sourceBuilder, topN, order); + } + + @Override public List getServiceInstanceTopN(int serviceId, String name, int topN, + Step step, long startTB, long endTB, Order order) throws IOException { + String indexName = TimePyramidTableNameBuilder.build(step, name); + + SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); + + BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); + sourceBuilder.query(boolQueryBuilder); + + boolQueryBuilder.must().add(QueryBuilders.rangeQuery(Indicator.TIME_BUCKET).lte(endTB).gte(startTB)); + boolQueryBuilder.must().add(QueryBuilders.termQuery(ServiceInstanceInventory.SERVICE_ID, serviceId)); + + return aggregation(indexName, sourceBuilder, topN, order); + } + + @Override + public List getAllEndpointTopN(String name, int topN, Step step, long startTB, + long endTB, Order order) throws IOException { + String indexName = TimePyramidTableNameBuilder.build(step, name); + + SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); + sourceBuilder.query(QueryBuilders.rangeQuery(Indicator.TIME_BUCKET).lte(endTB).gte(startTB)); + return aggregation(indexName, sourceBuilder, topN, order); + } + + @Override + public List getEndpointTopN(int serviceId, String name, int topN, Step step, + long startTB, long endTB, Order order) throws IOException { + String indexName = TimePyramidTableNameBuilder.build(step, name); + + SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); + + BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); + sourceBuilder.query(boolQueryBuilder); + + boolQueryBuilder.must().add(QueryBuilders.rangeQuery(Indicator.TIME_BUCKET).lte(endTB).gte(startTB)); + boolQueryBuilder.must().add(QueryBuilders.termQuery(EndpointInventory.SERVICE_ID, serviceId)); + + return aggregation(indexName, sourceBuilder, topN, order); + } + + private List aggregation(String indexName, SearchSourceBuilder sourceBuilder, int topN, + Order order) throws IOException { + boolean asc = false; + if (order.equals(Order.ASC)) { + asc = true; + } + + TermsAggregationBuilder aggregationBuilder = AggregationBuilders + .terms(Indicator.ENTITY_ID) + .field(Indicator.ENTITY_ID) + .order(BucketOrder.aggregation("value", asc)) + .size(topN) + .subAggregation( + AggregationBuilders.avg("value").field("value") + ); + sourceBuilder.aggregation(aggregationBuilder); + + SearchResponse response = getClient().search(indexName, sourceBuilder); + + List topNEntities = new ArrayList<>(); + Terms idTerms = response.getAggregations().get(Indicator.ENTITY_ID); + for (Terms.Bucket termsBucket : idTerms.getBuckets()) { + TopNEntity topNEntity = new TopNEntity(); + topNEntity.setId(termsBucket.getKeyAsString()); + Avg value = termsBucket.getAggregations().get("value"); + topNEntity.setValue((int)value.getValue()); + topNEntities.add(topNEntity); + } + + return topNEntities; + } +}