BanyanDB: Support cold stage data query for metrics/traces/logs

docs/en/api/query-protocol.md

@@ -48,6 +48,10 @@ extend type Query {
     estimateProcessScale(serviceId: ID!, labels: [String!]!): Long!
 
     getTimeInfo: TimeInfo
+    # Get the TTL info of records
+    getRecordsTTL: RecordsTTL
+    # Get the TTL info of metrics
+    getMetricsTTL: MetricsTTL
 }
 ```
 
@@ -156,6 +160,8 @@ extend type Query {
     queryBasicTraces(condition: TraceQueryCondition, debug: Boolean): TraceBrief
     # Read the specific trace ID with given trace ID
     queryTrace(traceId: ID!, debug: Boolean): Trace
+    # Only for BanyanDB, can be used to query the trace in the cold stage.
+    queryTraceFromColdStage(traceId: ID!, duration: Duration!, debug: Boolean): Trace
     # Read the list of searchable keys
     queryTraceTagAutocompleteKeys(duration: Duration!):[String!]
     # Search the available value options of the given key.
@@ -305,6 +311,8 @@ input Duration {
     start: String!
     end: String!
     step: Step!
+    # Only for BanyanDB, the flag to query from cold stage, default is false.
+    coldStage: Boolean
 }
 
 enum Step {

docs/en/changes/changes.md

@@ -8,11 +8,12 @@
 
 * BanyanDB: Support `hot/warm/cold` stages configuration.
 * Fix query continues profiling policies error when the policy is already in the cache.
-* Support `hot/warm/cold` stages TTL query in the status API.
+* Support `hot/warm/cold` stages TTL query in the status API and graphQL API.
 * PromQL Service: traffic query support `limit` and regex match.
 * Fix an edge case of HashCodeSelector(Integer#MIN_VALUE causes ArrayIndexOutOfBoundsException).
 * Support Flink monitoring.
 * BanyanDB: Support `@ShardingKey` for Measure tags and set to TopNAggregation group tag by default.
+* BanyanDB: Support cold stage data query for metrics/traces/logs.
 
 #### UI
 

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/trace/ProfileTaskQueryService.java

@@ -188,13 +188,13 @@ public ProfileAnalyzation getProfileAnalyze(final List<SegmentProfileAnalyzeQuer
 
     public List<SegmentRecord> getTaskSegments(String taskId) throws IOException {
         final List<String> profiledSegmentIdList = getProfileThreadSnapshotQueryDAO().queryProfiledSegmentIdList(taskId);
-        return getTraceQueryDAO().queryBySegmentIdList(profiledSegmentIdList);
+        return getTraceQueryDAO().queryBySegmentIdList(profiledSegmentIdList, null);
     }
 
     public List<ProfiledTraceSegments> getProfileTaskSegments(String taskId) throws IOException {
         // query all profiled segments
         final List<String> profiledSegmentIdList = getProfileThreadSnapshotQueryDAO().queryProfiledSegmentIdList(taskId);
-        final List<SegmentRecord> segmentRecords = getTraceQueryDAO().queryBySegmentIdList(profiledSegmentIdList);
+        final List<SegmentRecord> segmentRecords = getTraceQueryDAO().queryBySegmentIdList(profiledSegmentIdList, null);
         if (CollectionUtils.isEmpty(segmentRecords)) {
             return Collections.emptyList();
         }
@@ -215,7 +215,7 @@ public List<ProfiledTraceSegments> getProfileTaskSegments(String taskId) throws
         }
         final List<SegmentRecord> traceRelatedSegments = getTraceQueryDAO().queryByTraceIdWithInstanceId(
             new ArrayList<>(traceIdList),
-            new ArrayList<>(instanceIdList));
+            new ArrayList<>(instanceIdList), null);
 
         // group by the traceId + service instanceId
         final Map<String, List<SegmentRecord>> instanceTraceWithSegments = traceRelatedSegments.stream().filter(s -> {

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

@@ -28,6 +28,7 @@
 import java.util.Objects;
 
 import com.google.protobuf.InvalidProtocolBufferException;
+import javax.annotation.Nullable;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.skywalking.apm.network.common.v3.KeyIntValuePair;
 import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
@@ -142,7 +143,10 @@ public TraceBrief queryBasicTraces(final String serviceId,
         }
     }
 
-    public Trace queryTrace(final String traceId) throws IOException {
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    public Trace queryTrace(final String traceId, @Nullable final Duration duration) throws IOException {
         DebuggingTraceContext traceContext = TRACE_CONTEXT.get();
         DebuggingSpan span = null;
         try {
@@ -152,18 +156,18 @@ public Trace queryTrace(final String traceId) throws IOException {
                 msg.append("Condition: TraceId: ").append(traceId);
                 span.setMsg(msg.toString());
             }
-            return invokeQueryTrace(traceId);
+            return invokeQueryTrace(traceId, duration);
         } finally {
             if (traceContext != null && span != null) {
                 traceContext.stopSpan(span);
             }
         }
     }
 
-    private Trace invokeQueryTrace(final String traceId) throws IOException {
+    private Trace invokeQueryTrace(final String traceId, @Nullable final Duration duration) throws IOException {
         Trace trace = new Trace();
 
-        List<SegmentRecord> segmentRecords = getTraceQueryDAO().queryByTraceIdDebuggable(traceId);
+        List<SegmentRecord> segmentRecords = getTraceQueryDAO().queryByTraceIdDebuggable(traceId, duration);
         if (segmentRecords.isEmpty()) {
             trace.getSpans().addAll(getTraceQueryDAO().doFlexibleTraceQuery(traceId));
         } else {
@@ -192,7 +196,7 @@ private Trace invokeQueryTrace(final String traceId) throws IOException {
 
         if (CollectionUtils.isNotEmpty(sortedSpans)) {
             final List<SpanAttachedEventRecord> spanAttachedEvents = getSpanAttachedEventQueryDAO().
-                querySpanAttachedEventsDebuggable(SpanAttachedEventTraceType.SKYWALKING, Arrays.asList(traceId));
+                querySpanAttachedEventsDebuggable(SpanAttachedEventTraceType.SKYWALKING, Arrays.asList(traceId), duration);
             appendAttachedEventsToSpanDebuggable(sortedSpans, spanAttachedEvents);
         }
 

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

@@ -34,6 +34,7 @@ public class Duration {
     private String start;
     private String end;
     private Step step;
+    private boolean coldStage = false;
 
     /**
      * See {@link DurationUtils#convertToTimeBucket(Step, String)}

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

@@ -18,8 +18,10 @@
 
 package org.apache.skywalking.oap.server.core.storage.query;
 
+import javax.annotation.Nullable;
 import org.apache.skywalking.oap.server.core.analysis.manual.spanattach.SpanAttachedEventRecord;
 import org.apache.skywalking.oap.server.core.analysis.manual.spanattach.SpanAttachedEventTraceType;
+import org.apache.skywalking.oap.server.core.query.input.Duration;
 import org.apache.skywalking.oap.server.core.query.type.debugging.DebuggingSpan;
 import org.apache.skywalking.oap.server.core.query.type.debugging.DebuggingTraceContext;
 import org.apache.skywalking.oap.server.library.module.Service;
@@ -28,7 +30,10 @@
 import java.util.List;
 
 public interface ISpanAttachedEventQueryDAO extends Service {
-    default List<SpanAttachedEventRecord> querySpanAttachedEventsDebuggable(SpanAttachedEventTraceType type, List<String> traceIds) throws IOException {
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    default List<SpanAttachedEventRecord> querySpanAttachedEventsDebuggable(SpanAttachedEventTraceType type, List<String> traceIds, @Nullable Duration duration) throws IOException {
         DebuggingTraceContext traceContext = DebuggingTraceContext.TRACE_CONTEXT.get();
         DebuggingSpan span = null;
         try {
@@ -41,13 +46,16 @@ default List<SpanAttachedEventRecord> querySpanAttachedEventsDebuggable(SpanAtta
                        .append(traceIds);
                 span.setMsg(builder.toString());
             }
-            return querySpanAttachedEvents(type, traceIds);
+            return querySpanAttachedEvents(type, traceIds, duration);
         } finally {
             if (traceContext != null && span != null) {
                 traceContext.stopSpan(span);
             }
         }
     }
 
-    List<SpanAttachedEventRecord> querySpanAttachedEvents(SpanAttachedEventTraceType type, List<String> traceIds) throws IOException;
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    List<SpanAttachedEventRecord> querySpanAttachedEvents(SpanAttachedEventTraceType type, List<String> traceIds, @Nullable Duration duration) throws IOException;
 }

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

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import javax.annotation.Nullable;
 import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.Tag;
 import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord;
 import org.apache.skywalking.oap.server.core.query.input.Duration;
@@ -88,7 +89,10 @@ default TraceBrief queryBasicTracesDebuggable(Duration duration,
         }
     }
 
-    default List<SegmentRecord> queryByTraceIdDebuggable(String traceId) throws IOException {
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    default List<SegmentRecord> queryByTraceIdDebuggable(String traceId, @Nullable Duration duration) throws IOException {
         DebuggingTraceContext traceContext = DebuggingTraceContext.TRACE_CONTEXT.get();
         DebuggingSpan span = null;
         try {
@@ -99,7 +103,7 @@ default List<SegmentRecord> queryByTraceIdDebuggable(String traceId) throws IOEx
                        .append(traceId);
                 span.setMsg(builder.toString());
             }
-            return queryByTraceId(traceId);
+            return queryByTraceId(traceId, duration);
         } finally {
             if (traceContext != null && span != null) {
                 traceContext.stopSpan(span);
@@ -120,14 +124,23 @@ TraceBrief queryBasicTraces(Duration duration,
                                 QueryOrder queryOrder,
                                 final List<Tag> tags) throws IOException;
 
-    List<SegmentRecord> queryByTraceId(String traceId) throws IOException;
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    List<SegmentRecord> queryByTraceId(String traceId, @Nullable Duration duration) throws IOException;
 
-    List<SegmentRecord> queryBySegmentIdList(List<String> segmentIdList) throws IOException;
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    List<SegmentRecord> queryBySegmentIdList(List<String> segmentIdList, @Nullable Duration duration) throws IOException;
 
-    List<SegmentRecord> queryByTraceIdWithInstanceId(List<String> traceIdList, List<String> instanceIdList) throws IOException;
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    List<SegmentRecord> queryByTraceIdWithInstanceId(List<String> traceIdList, List<String> instanceIdList, @Nullable Duration duration) throws IOException;
 
     /**
-     * This method gives more flexible for 3rd trace without segment concept, which can't search data through {@link #queryByTraceId(String)}
+     * This method gives more flexible for 3rd trace without segment concept, which can't search data through {@link #queryByTraceId(String, Duration)}
      */
     List<Span> doFlexibleTraceQuery(String traceId) throws IOException;
 }

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

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.core.storage.query;
 
+import javax.annotation.Nullable;
 import org.apache.skywalking.oap.server.core.query.input.Duration;
 import org.apache.skywalking.oap.server.core.query.type.debugging.DebuggingSpan;
 import org.apache.skywalking.oap.server.core.query.type.debugging.DebuggingTraceContext;
@@ -52,15 +53,18 @@ default List<List<Span>> getTracesDebuggable(final QueryRequest request,
         }
     }
 
-    default List<Span> getTraceDebuggable(final String traceId) throws IOException {
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    default List<Span> getTraceDebuggable(final String traceId, @Nullable final Duration duration) throws IOException {
         DebuggingTraceContext traceContext = DebuggingTraceContext.TRACE_CONTEXT.get();
         DebuggingSpan span = null;
         try {
             if (traceContext != null) {
                 span = traceContext.createSpan("Query Dao: getTrace");
                 span.setMsg("Condition: TraceId: " + traceId);
             }
-            return getTrace(traceId);
+            return getTrace(traceId, duration);
         } finally {
             if (traceContext != null && span != null) {
                 traceContext.stopSpan(span);
@@ -74,9 +78,18 @@ default List<Span> getTraceDebuggable(final String traceId) throws IOException {
 
     List<String> getSpanNames(final String serviceName) throws IOException;
 
-    List<Span> getTrace(final String traceId) throws IOException;
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    List<Span> getTrace(final String traceId, @Nullable final Duration duration) throws IOException;
 
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
     List<List<Span>> getTraces(final QueryRequest request, final Duration duration) throws IOException;
 
-    List<List<Span>> getTraces(final Set<String> traceIds) throws IOException;
+    /**
+     * @param duration nullable unless for BanyanDB query from cold stage
+     */
+    List<List<Span>> getTraces(final Set<String> traceIds, @Nullable final Duration duration) throws IOException;
 }

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

@@ -28,12 +28,15 @@
 import org.apache.skywalking.oap.server.core.CoreModule;
 import org.apache.skywalking.oap.server.core.analysis.IDManager;
 import org.apache.skywalking.oap.server.core.query.MetadataQueryService;
+import org.apache.skywalking.oap.server.core.query.TTLStatusQuery;
 import org.apache.skywalking.oap.server.core.query.input.Duration;
 import org.apache.skywalking.oap.server.core.query.type.Endpoint;
 import org.apache.skywalking.oap.server.core.query.type.EndpointInfo;
 import org.apache.skywalking.oap.server.core.query.type.Process;
 import org.apache.skywalking.oap.server.core.query.type.Service;
 import org.apache.skywalking.oap.server.core.query.type.ServiceInstance;
+import org.apache.skywalking.oap.server.core.storage.ttl.MetricsTTL;
+import org.apache.skywalking.oap.server.core.storage.ttl.RecordsTTL;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
 
 import static org.apache.skywalking.oap.query.graphql.AsyncQueryUtils.queryAsync;
@@ -47,6 +50,7 @@ public class MetadataQueryV2 implements GraphQLQueryResolver {
 
     private final ModuleManager moduleManager;
     private MetadataQueryService metadataQueryService;
+    private TTLStatusQuery ttlStatusQuery;
 
     public MetadataQueryV2(ModuleManager moduleManager) {
         this.moduleManager = moduleManager;
@@ -61,6 +65,15 @@ private MetadataQueryService getMetadataQueryService() {
         return metadataQueryService;
     }
 
+    private TTLStatusQuery getTTLStatusQuery() {
+        if (ttlStatusQuery == null) {
+            ttlStatusQuery = moduleManager.find(CoreModule.NAME)
+                                          .provider()
+                                          .getService(TTLStatusQuery.class);
+        }
+        return ttlStatusQuery;
+    }
+
     public CompletableFuture<Set<String>> listLayers() {
         return queryAsync(() -> getMetadataQueryService().listLayers());
     }
@@ -115,4 +128,12 @@ public TimeInfo getTimeInfo() {
         timeInfo.setTimezone(timezoneFormat.format(date));
         return timeInfo;
     }
+
+    public RecordsTTL getRecordsTTL() {
+        return getTTLStatusQuery().getTTL().getRecords();
+    }
+
+    public MetricsTTL getMetricsTTL() {
+        return getTTLStatusQuery().getTTL().getMetrics();
+    }
 }

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

@@ -117,7 +117,28 @@ public CompletableFuture<Trace> queryTrace(final String traceId, boolean debug)
             DebuggingTraceContext.TRACE_CONTEXT.set(traceContext);
             DebuggingSpan span = traceContext.createSpan("Query trace");
             try {
-                Trace trace = getQueryService().queryTrace(traceId);
+                Trace trace = getQueryService().queryTrace(traceId, null);
+                if (debug) {
+                    trace.setDebuggingTrace(traceContext.getExecTrace());
+                }
+                return trace;
+            } finally {
+                traceContext.stopSpan(span);
+                traceContext.stopTrace();
+                TRACE_CONTEXT.remove();
+            }
+        });
+    }
+
+    public CompletableFuture<Trace> queryTraceFromColdStage(final String traceId, Duration duration, boolean debug) {
+        duration.setColdStage(true);
+        return queryAsync(() -> {
+            DebuggingTraceContext traceContext = new DebuggingTraceContext(
+                "TraceId: " + traceId, debug, false);
+            DebuggingTraceContext.TRACE_CONTEXT.set(traceContext);
+            DebuggingSpan span = traceContext.createSpan("Query trace from cold stage");
+            try {
+                Trace trace = getQueryService().queryTrace(traceId, duration);
                 if (debug) {
                     trace.setDebuggingTrace(traceContext.getExecTrace());
                 }