Add pagination to event query (#7274)

CHANGES.md

@@ -78,6 +78,7 @@ Release Notes.
 * Configuration: Allow to configure server maximum request header size.
 * Add thread state metric and class loaded info metric to JVMMetric.
 * Performance: compile LAL DSL statically and run with type checked.
+* Add pagination to event query protocol.
 
 #### UI
 

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/PaginationUtils.java

@@ -18,6 +18,8 @@
 
 package org.apache.skywalking.oap.server.core.query;
 
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
 import org.apache.skywalking.oap.server.core.query.type.Pagination;
 
 public enum PaginationUtils {
@@ -30,21 +32,10 @@ public enum PaginationUtils {
         return new Page(from, limit);
     }
 
-    public class Page {
-        private int from;
-        private int limit;
-
-        Page(int from, int limit) {
-            this.from = from;
-            this.limit = limit;
-        }
-
-        public int getFrom() {
-            return from;
-        }
-
-        public int getLimit() {
-            return limit;
-        }
+    @Data
+    @RequiredArgsConstructor
+    public static class Page {
+        private final int from;
+        private final int limit;
     }
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/Pagination.java

@@ -18,11 +18,15 @@
 
 package org.apache.skywalking.oap.server.core.query.type;
 
+import lombok.AllArgsConstructor;
 import lombok.Getter;
+import lombok.NoArgsConstructor;
 import lombok.Setter;
 
 @Getter
 @Setter
+@NoArgsConstructor
+@AllArgsConstructor
 public class Pagination {
     private int pageNum;
     private int pageSize;

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/event/EventQueryCondition.java

@@ -24,9 +24,7 @@ import lombok.Data;
 import lombok.NoArgsConstructor;
 import org.apache.skywalking.oap.server.core.query.enumeration.Order;
 import org.apache.skywalking.oap.server.core.query.input.Duration;
-
-import static org.apache.skywalking.oap.server.core.storage.query.IEventQueryDAO.DEFAULT_SIZE;
-import static org.apache.skywalking.oap.server.core.storage.query.IEventQueryDAO.MAX_SIZE;
+import org.apache.skywalking.oap.server.core.query.type.Pagination;
 
 @Data
 @AllArgsConstructor
@@ -45,9 +43,5 @@ public class EventQueryCondition {
 
     private Order order;
 
-    private int size;
-
-    public int getSize() {
-        return size > 0 ? Math.min(size, MAX_SIZE) : DEFAULT_SIZE;
-    }
+    private Pagination paging;
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IEventQueryDAO.java

@@ -25,7 +25,6 @@ import org.apache.skywalking.oap.server.core.storage.DAO;
 import java.util.List;
 
 public interface IEventQueryDAO extends DAO {
-    int DEFAULT_SIZE = 20;
     int MAX_SIZE = 100;
 
     Events queryEvents(final EventQueryCondition condition) throws Exception;

oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java

@@ -88,7 +88,9 @@ public class AlarmQuery implements GraphQLQueryResolver {
         }
         long startSecondTB = 0;
         long endSecondTB = 0;
-        final EventQueryCondition.EventQueryConditionBuilder conditionPrototype = EventQueryCondition.builder().size(IEventQueryDAO.MAX_SIZE);
+        final EventQueryCondition.EventQueryConditionBuilder conditionPrototype =
+            EventQueryCondition.builder()
+                               .paging(new Pagination(1, IEventQueryDAO.MAX_SIZE, false));
         if (nonNull(duration)) {
             startSecondTB = duration.getStartTimeBucketInSec();
             endSecondTB = duration.getEndTimeBucketInSec();

query-protocol @ b827d0d7

-Subproject commit 9257b0287282cd7a1096167710d3fdb9e2e6fc61
+Subproject commit b827d0d7d95396b9ae87a10eef80a67af843cf3f

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/ESEventQueryDAO.java

@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.apache.skywalking.oap.server.core.query.PaginationUtils;
 import org.apache.skywalking.oap.server.core.source.Event;
 import org.apache.skywalking.oap.server.core.query.enumeration.Order;
 import org.apache.skywalking.oap.server.core.query.input.Duration;
@@ -129,7 +130,10 @@ public class ESEventQueryDAO extends EsDAO implements IEventQueryDAO {
         EventQueryCondition condition = conditionList.get(0);
         final Order queryOrder = isNull(condition.getOrder()) ? Order.DES : condition.getOrder();
         sourceBuilder.sort(Event.START_TIME, Order.DES.equals(queryOrder) ? SortOrder.DESC : SortOrder.ASC);
-        sourceBuilder.size(condition.getSize());
+
+        final PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging());
+        sourceBuilder.from(page.getFrom());
+        sourceBuilder.size(page.getLimit());
         return sourceBuilder;
     }
 
@@ -144,7 +148,10 @@ public class ESEventQueryDAO extends EsDAO implements IEventQueryDAO {
 
         final Order queryOrder = isNull(condition.getOrder()) ? Order.DES : condition.getOrder();
         sourceBuilder.sort(Event.START_TIME, Order.DES.equals(queryOrder) ? SortOrder.DESC : SortOrder.ASC);
-        sourceBuilder.size(condition.getSize());
+
+        final PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging());
+        sourceBuilder.from(page.getFrom());
+        sourceBuilder.size(page.getLimit());
 
         return sourceBuilder;
     }

oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/EventQueryDAO.java

@@ -26,6 +26,7 @@ import java.util.Map;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.query.PaginationUtils;
 import org.apache.skywalking.oap.server.core.source.Event;
 import org.apache.skywalking.oap.server.core.query.enumeration.Order;
 import org.apache.skywalking.oap.server.core.query.input.Duration;
@@ -110,10 +111,12 @@ public class EventQueryDAO implements IEventQueryDAO {
 
     protected List<WhereQueryImpl<SelectQueryImpl>> buildWhereQueries(final EventQueryCondition condition) {
         List<WhereQueryImpl<SelectQueryImpl>> queries = new ArrayList<>(2);
+
+        final PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging());
         final String topFunc = Order.DES.equals(condition.getOrder()) ? InfluxConstants.SORT_DES : InfluxConstants.SORT_ASC;
         final WhereQueryImpl<SelectQueryImpl> recallWhereQuery =
                 select().raw(ALL_FIELDS)
-                        .function(topFunc, Event.START_TIME, condition.getSize())
+                        .function(topFunc, Event.START_TIME, page.getLimit() + page.getFrom())
                         .from(client.getDatabase(), Event.INDEX_NAME)
                         .where();
         final SelectQueryImpl countQuery = select().count(Event.UUID).from(client.getDatabase(), Event.INDEX_NAME);

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2EventQueryDAO.java

@@ -28,6 +28,8 @@ import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.query.PaginationUtils;
+import org.apache.skywalking.oap.server.core.query.type.Pagination;
 import org.apache.skywalking.oap.server.core.source.Event;
 import org.apache.skywalking.oap.server.core.query.input.Duration;
 import org.apache.skywalking.oap.server.core.query.type.event.EventQueryCondition;
@@ -65,7 +67,10 @@ public class H2EventQueryDAO implements IEventQueryDAO {
                 result.setTotal(resultSet.getInt("total"));
             }
 
-            sql = "select * from " + Event.INDEX_NAME + whereClause + " limit " + condition.getSize();
+            PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging());
+
+            sql = "select * from " + Event.INDEX_NAME + whereClause
+                + " limit " + page.getLimit() + " offset " + page.getFrom();
             if (log.isDebugEnabled()) {
                 log.debug("Query SQL: {}, parameters: {}", sql, parameters);
             }
@@ -92,7 +97,8 @@ public class H2EventQueryDAO implements IEventQueryDAO {
                                                        .map(Tuple2::_1)
                                                        .map(it -> it.collect(Collectors.joining(" and ")))
                                                        .collect(Collectors.joining(" or ", " where ", ""));
-        final int size = conditions.stream().mapToInt(EventQueryCondition::getSize).sum();
+        final int size = conditions.stream().map(EventQueryCondition::getPaging)
+                                   .mapToInt(Pagination::getPageSize).sum();
 
         final Events result = new Events();
         try (final Connection connection = client.getConnection()) {

test/e2e/e2e-data/src/main/java/org/apache/skywalking/e2e/SimpleQueryClient.java

@@ -456,7 +456,10 @@ public class SimpleQueryClient {
         final String queryString = Resources.readLines(queryFileUrl, StandardCharsets.UTF_8)
                                             .stream().filter(it -> !it.startsWith("#"))
                                             .collect(Collectors.joining())
-                                            .replace("{uuid}", query.uuid());
+                                            .replace("{uuid}", query.uuid())
+                                            .replace("{pageNum}", query.pageNum())
+                                            .replace("{pageSize}", query.pageSize())
+                                            .replace("{needTotal}", query.needTotal());
         LOGGER.info("Query: {}", queryString);
         final ResponseEntity<GQLResponse<EventData>> responseEntity = restTemplate.exchange(
             new RequestEntity<>(queryString, HttpMethod.POST, URI.create(endpointUrl)),

test/e2e/e2e-data/src/main/java/org/apache/skywalking/e2e/event/EventsQuery.java

@@ -27,4 +27,7 @@ import org.apache.skywalking.e2e.AbstractQuery;
 @Accessors(fluent = true)
 public class EventsQuery extends AbstractQuery<EventsQuery> {
     private String uuid;
+    private String pageNum = "1";
+    private String pageSize = "20";
+    private String needTotal = "true";
 }

test/e2e/e2e-data/src/main/resources/events.gql

@@ -33,7 +33,12 @@
     }",
     "variables": {
         "condition": {
-            "uuid": "{uuid}"
+            "uuid": "{uuid}",
+            "paging": {
+                "pageNum": {pageNum},
+                "pageSize": {pageSize},
+                "needTotal": {needTotal}
+            }
         }
     }
 }