BanyanDB: add topN cold stage data query.

.github/workflows/skywalking.yaml

@@ -1021,7 +1021,7 @@ jobs:
             # check if segment files exist
             if docker exec $CONTAINER_ID sh -c '[ -n "$(ls /tmp/measure-data/measure/data/day/seg* 2>/dev/null)" ]'; then
               echo "✅ found segment files"
-              sleep 30
+              sleep 180
               # create and copy files
               docker cp $CONTAINER_ID:/tmp ${BANYANDB_DATA_GENERATE_ROOT}
               docker cp $CONTAINER_ID:/tmp/measure-data/measure/data/index ${BANYANDB_DATA_GENERATE_ROOT}

docs/en/changes/changes.md

@@ -26,6 +26,7 @@
 * Enhance the trace `List/Tree/Table` graph to support displaying multiple refs of spans and distinguishing different parents.
 * Fix: correct the same labels for metrics.
 * Refactor: use the Fetch API to instead of Axios.
+* Support cold stage data for metrics, trace and log.
 
 #### Documentation
 

oap-server-bom/pom.xml

@@ -72,7 +72,7 @@
         <httpcore.version>4.4.16</httpcore.version>
         <httpasyncclient.version>4.1.5</httpasyncclient.version>
         <commons-compress.version>1.21</commons-compress.version>
-        <banyandb-java-client.version>0.9.0-rc0</banyandb-java-client.version>
+        <banyandb-java-client.version>0.9.0-rc2</banyandb-java-client.version>
         <kafka-clients.version>3.4.0</kafka-clients.version>
         <spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
         <consul.client.version>1.5.3</consul.client.version>

oap-server/server-query-plugin/promql-plugin/src/main/java/org/apache/skywalking/oap/query/promql/handler/PromQLApiHandler.java

@@ -511,7 +511,7 @@ public HttpResponse query_range(
     @Get
     @Post
     @Path("/api/v1/format_query")
-    public HttpResponse query_range(@Param("query") String query) throws IOException {
+    public HttpResponse format_query(@Param("query") String query) throws IOException {
         QueryFormatRsp rsp = new QueryFormatRsp();
         rsp.setData(query.replaceAll("\\s", ""));
         return jsonResponse(rsp);

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java

@@ -78,7 +78,6 @@ public List<SelectedRecord> sortMetrics(TopNCondition condition, String valueCol
         return directMetricsTopN(isColdStage, condition, schema, valueColumnName, spec, getTimestampRange(duration), additionalConditions);
     }
 
-    //todo: query cold stage
     List<SelectedRecord> serverSideTopN(boolean isColdStage, TopNCondition condition, MetadataRegistry.Schema schema, MetadataRegistry.ColumnSpec valueColumnSpec,
                                         TimestampRange timestampRange, List<KeyValue> additionalConditions) throws IOException {
         TopNQueryResponse resp = null;

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java

@@ -139,22 +139,6 @@ protected StreamQueryResponse queryDebuggable(boolean isColdStage,
         }
     }
 
-    protected TopNQueryResponse topN(MetadataRegistry.Schema schema,
-                                     TimestampRange timestampRange,
-                                     int number,
-                                     List<KeyValue> additionalConditions,
-                                     List<AttrCondition> attributes) throws IOException {
-        return topNQuery(schema, timestampRange, number, AbstractQuery.Sort.DESC, additionalConditions, attributes);
-    }
-
-    protected TopNQueryResponse bottomN(MetadataRegistry.Schema schema,
-                                        TimestampRange timestampRange,
-                                        int number,
-                                        List<KeyValue> additionalConditions,
-                                        List<AttrCondition> attributes) throws IOException {
-        return topNQuery(schema, timestampRange, number, AbstractQuery.Sort.ASC, additionalConditions, attributes);
-    }
-
     protected TopNQueryResponse topNQueryDebuggable(boolean isColdStage,
                                                     MetadataRegistry.Schema schema,
                                                     TimestampRange timestampRange,
@@ -185,7 +169,7 @@ protected TopNQueryResponse topNQueryDebuggable(boolean isColdStage,
                        .append(isColdStage);
                 span.setMsg(builder.toString());
             }
-            TopNQueryResponse response = topNQuery(schema, timestampRange, number, sort, additionalConditions, attributes);
+            TopNQueryResponse response = topNQuery(isColdStage, schema, timestampRange, number, sort, additionalConditions, attributes);
             if (traceContext != null && traceContext.isDumpStorageRsp()) {
                 builder.append("\n").append(" Response: ").append(new Gson().toJson(response.getTopNLists()));
                 span.setMsg(builder.toString());
@@ -198,7 +182,8 @@ protected TopNQueryResponse topNQueryDebuggable(boolean isColdStage,
         }
     }
 
-    private TopNQueryResponse topNQuery(MetadataRegistry.Schema schema,
+    private TopNQueryResponse topNQuery(boolean isColdStage,
+                                        MetadataRegistry.Schema schema,
                                         TimestampRange timestampRange,
                                         int number,
                                         AbstractQuery.Sort sort,
@@ -225,6 +210,9 @@ private TopNQueryResponse topNQuery(MetadataRegistry.Schema schema,
             });
         }
         q.setConditions(conditions);
+        if (isColdStage) {
+            q.setStages(List.of(BanyanDBStorageConfig.StageName.cold.name()));
+        }
 
         return getClient().query(q);
     }

test/e2e-v2/cases/storage/banyandb/stages/e2e.yaml

@@ -69,3 +69,8 @@ verify:
             | yq e '.traces | select(.[].endpointnames[0]=="/dubbox-case/case/dubbox-rest/404-test") | .[2].traceids[0]' -
         ) --start="-96h" --end="-48h" --cold=true
       expected: ../../expected/cold/trace-detail.yml
+    ## topN
+    - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics exec --expression="top_n(service_resp_time,3,des,attr0='GENERAL')" --start="-96h" --end="-48h" --cold=true --step=DAY
+      expected: ../../expected/cold/topN-OP-service.yml
+    - query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics exec --expression="top_n(endpoint_resp_time,3,des)" --start="-96h" --end="-48h" --cold=true --step=DAY
+      expected: ../../expected/cold/topN-OP-endpoint.yml

test/e2e-v2/cases/storage/expected/cold/dependency-instance.yml

@@ -37,5 +37,6 @@ calls:
   targetcomponents: []
   id: {{ b64enc "mock_a_service" }}.1_{{ b64enc "mock_a_service_instance" }}-{{ b64enc "mock_b_service" }}.1_{{ b64enc "mock_b_service_instance" }}
   detectpoints:
+    - CLIENT
     - SERVER
 {{- end }}

test/e2e-v2/cases/storage/expected/cold/topN-OP-endpoint.yml

@@ -0,0 +1,58 @@
+# 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.
+
+debuggingtrace: null
+type: SORTED_LIST
+results:
+  - metric:
+      labels: []
+    values:
+      - id: "mock_a_service - /dubbox-case/case/dubbox-rest/404-test"
+        value: "{{ (index (index .results 0).values 0).value }}"
+        traceid: null
+        owner:
+          scope: Endpoint
+          serviceid: {{ b64enc "mock_a_service"}}.1
+          servicename: mock_a_service
+          normal: true
+          serviceinstanceid: null
+          serviceinstancename: null
+          endpointid: {{ b64enc "mock_a_service" }}.1_{{ b64enc "/dubbox-case/case/dubbox-rest/404-test" }}
+          endpointname: /dubbox-case/case/dubbox-rest/404-test
+      - id: "mock_b_service - org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()"
+        value: "{{ (index (index .results 0).values 1).value }}"
+        traceid: null
+        owner:
+          scope: Endpoint
+          serviceid: {{ b64enc "mock_b_service" }}.1
+          servicename: mock_b_service
+          normal: true
+          serviceinstanceid: null
+          serviceinstancename: null
+          endpointid: {{ b64enc "mock_b_service" }}.1_{{ b64enc "org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()" }}
+          endpointname: org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()
+      - id: "mock_c_service - org.apache.skywalking.RocketMQ"
+        value: "{{ (index (index .results 0).values 2).value }}"
+        traceid: null
+        owner:
+          scope: Endpoint
+          serviceid: {{ b64enc "mock_c_service" }}.1
+          servicename: mock_c_service
+          normal: true
+          serviceinstanceid: null
+          serviceinstancename: null
+          endpointid: {{ b64enc "mock_c_service" }}.1_{{ b64enc "org.apache.skywalking.RocketMQ" }}
+          endpointname: org.apache.skywalking.RocketMQ
+error: null

test/e2e-v2/cases/storage/expected/cold/topN-OP-service.yml

@@ -0,0 +1,58 @@
+# 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.
+
+debuggingtrace: null
+type: SORTED_LIST
+results:
+  - metric:
+      labels: []
+    values:
+      - id: mock_a_service
+        value: "{{ (index (index .results 0).values 0).value }}"
+        traceid: null
+        owner:
+          scope: Service
+          serviceid: {{ b64enc "mock_a_service"}}.1
+          servicename: mock_a_service
+          normal: true
+          serviceinstanceid: null
+          serviceinstancename: null
+          endpointid: null
+          endpointname: null
+      - id: mock_b_service
+        value: "{{ (index (index .results 0).values 1).value }}"
+        traceid: null
+        owner:
+          scope: Service
+          serviceid: {{ b64enc "mock_b_service"}}.1
+          servicename: mock_b_service
+          normal: true
+          serviceinstanceid: null
+          serviceinstancename: null
+          endpointid: null
+          endpointname: null
+      - id: mock_c_service
+        value: "{{ (index (index .results 0).values 2).value }}"
+        traceid: null
+        owner:
+          scope: Service
+          serviceid: {{ b64enc "mock_c_service" }}.1
+          servicename: mock_c_service
+          normal: true
+          serviceinstanceid: null
+          serviceinstancename: null
+          endpointid: null
+          endpointname: null
+error: null

test/e2e-v2/java-test-service/trace-mocker/src/main/java/org/apache/skywalking/mock/AgentDataMock.java

@@ -73,6 +73,7 @@ public static void main(String[] args) throws InterruptedException {
                 streamObserver, traceId, serviceBSegmentId, serviceASegmentId, startTimestamp);
             serviceCMock.mock(
                 streamObserver, traceId, serviceCSegmentId, serviceBSegmentId, startTimestamp);
+            TimeUnit.SECONDS.sleep(30);
         }
 
         streamObserver.onCompleted();