Fix: TTL query add metadata TTL and PersistentWorker used wrong TTL for metrics cache if the storage is BanyanDB. (#13827)

Author: wankai123

.github/workflows/skywalking.yaml

@@ -390,8 +390,9 @@ jobs:
           - name: Storage OpenSearch 3.0.0
             config: test/e2e-v2/cases/storage/opensearch/e2e.yaml
             env: OPENSEARCH_VERSION=3.0.0
-          - name: Storage ES Sharding
+          - name: Storage ES Sharding 8.18.8
             config: test/e2e-v2/cases/storage/es/es-sharding/e2e.yaml
+            env: ES_VERSION=8.18.8
 
           - name: Alarm ES
             config: test/e2e-v2/cases/alarm/es/e2e.yaml

docs/en/changes/changes.md

@@ -29,6 +29,8 @@
 * LAL: add `layer: auto` mode for dynamic layer assignment when `service.layer` is absent.
 * Add two-phase `SpanListener` SPI mechanism for extensible trace span processing. Refactor GenAI from hardcoded `SpanForward.processGenAILogic()` to `GenAISpanListener`.
 * Add OTLP/HTTP receiver support for traces, logs, and metrics (`/v1/traces`, `/v1/logs`, `/v1/metrics`). Supports both `application/x-protobuf` and `application/json` content types.
+* Fix: TTL query add metadata TTL.
+* Fix: PersistentWorker used wrong TTL for metrics cache if the storage is BanyanDB.
 
 #### UI
 

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

@@ -394,7 +394,6 @@ TTLStatusQuery.class, new TTLStatusQuery(
         final MetricsStreamProcessor metricsStreamProcessor = MetricsStreamProcessor.getInstance();
         metricsStreamProcessor.setL1FlushPeriod(moduleConfig.getL1FlushPeriod());
         metricsStreamProcessor.setStorageSessionTimeout(moduleConfig.getStorageSessionTimeout());
-        metricsStreamProcessor.setMetricsDataTTL(moduleConfig.getMetricsDataTTL());
         RecordStreamProcessor.getInstance().setRecordDataTTL(moduleConfig.getRecordDataTTL());
         TopNStreamProcessor.getInstance().setTopNWorkerReportCycle(moduleConfig.getTopNReportPeriod());
         apdexThresholdConfig = new ApdexThresholdConfig(this);

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java

@@ -29,6 +29,7 @@
 import org.apache.skywalking.oap.server.core.analysis.StreamProcessor;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.config.DownSamplingConfigService;
+import org.apache.skywalking.oap.server.core.query.TTLStatusQuery;
 import org.apache.skywalking.oap.server.core.storage.IMetricsDAO;
 import org.apache.skywalking.oap.server.core.storage.StorageBuilderFactory;
 import org.apache.skywalking.oap.server.core.storage.StorageDAO;
@@ -82,11 +83,6 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
      */
     @Setter
     private long storageSessionTimeout = 70_000;
-    /**
-     * @since 8.7.0 TTL settings from {@link org.apache.skywalking.oap.server.core.CoreModuleConfig#getMetricsDataTTL()}
-     */
-    @Setter
-    private int metricsDataTTL = 3;
 
     public static MetricsStreamProcessor getInstance() {
         return PROCESSOR;
@@ -145,6 +141,9 @@ private void create(ModuleDefineHolder moduleDefineHolder,
         DownSamplingConfigService configService = moduleDefineHolder.find(CoreModule.NAME)
                                                                     .provider()
                                                                     .getService(DownSamplingConfigService.class);
+        TTLStatusQuery ttlStatusQuery = moduleDefineHolder.find(CoreModule.NAME)
+                                                          .provider()
+                                                          .getService(TTLStatusQuery.class);
 
         MetricsPersistentWorker hourPersistentWorker = null;
         MetricsPersistentWorker dayPersistentWorker = null;
@@ -168,13 +167,15 @@ private void create(ModuleDefineHolder moduleDefineHolder,
                 Model model = modelSetter.add(
                     metricsClass, stream.getScopeId(), new Storage(stream.getName(), timeRelativeID, DownSampling.Hour)
                 );
-                hourPersistentWorker = downSamplingWorker(moduleDefineHolder, metricsDAO, model, supportUpdate, kind);
+                int hourTTL = ttlStatusQuery.getMetricsTTL(model);
+                hourPersistentWorker = downSamplingWorker(moduleDefineHolder, metricsDAO, model, supportUpdate, kind, hourTTL);
             }
             if (configService.shouldToDay()) {
                 Model model = modelSetter.add(
                     metricsClass, stream.getScopeId(), new Storage(stream.getName(), timeRelativeID, DownSampling.Day)
                 );
-                dayPersistentWorker = downSamplingWorker(moduleDefineHolder, metricsDAO, model, supportUpdate, kind);
+                int dayTTL = ttlStatusQuery.getMetricsTTL(model);
+                dayPersistentWorker = downSamplingWorker(moduleDefineHolder, metricsDAO, model, supportUpdate, kind, dayTTL);
             }
 
             transWorker = new MetricsTransWorker(
@@ -184,8 +185,9 @@ private void create(ModuleDefineHolder moduleDefineHolder,
         Model model = modelSetter.add(
             metricsClass, stream.getScopeId(), new Storage(stream.getName(), timeRelativeID, DownSampling.Minute)
         );
+        int minuteTTL = ttlStatusQuery.getMetricsTTL(model);
         MetricsPersistentWorker minutePersistentWorker = minutePersistentWorker(
-            moduleDefineHolder, metricsDAO, model, transWorker, supportUpdate, kind, metricsClass);
+            moduleDefineHolder, metricsDAO, model, transWorker, supportUpdate, kind, metricsClass, minuteTTL);
 
         String remoteReceiverWorkerName = stream.getName() + "_rec";
         IWorkerInstanceSetter workerInstanceSetter = moduleDefineHolder.find(CoreModule.NAME)
@@ -205,7 +207,8 @@ private MetricsPersistentWorker minutePersistentWorker(ModuleDefineHolder module
                                                            MetricsTransWorker transWorker,
                                                            boolean supportUpdate,
                                                            MetricStreamKind kind,
-                                                           Class<? extends Metrics> metricsClass) {
+                                                           Class<? extends Metrics> metricsClass,
+                                                           int metricsDataTTL) {
         AlarmNotifyWorker alarmNotifyWorker = new AlarmNotifyWorker(moduleDefineHolder);
         ExportMetricsWorker exportWorker = new ExportMetricsWorker(moduleDefineHolder);
 
@@ -222,7 +225,8 @@ private MetricsPersistentWorker downSamplingWorker(ModuleDefineHolder moduleDefi
                                                        IMetricsDAO metricsDAO,
                                                        Model model,
                                                        boolean supportUpdate,
-                                                       MetricStreamKind kind) {
+                                                       MetricStreamKind kind,
+                                                       int metricsDataTTL) {
         MetricsPersistentWorker persistentWorker = new MetricsPersistentWorker(
             moduleDefineHolder, model, metricsDAO,
             supportUpdate, storageSessionTimeout, metricsDataTTL, kind

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

@@ -20,6 +20,7 @@
 
 import lombok.RequiredArgsConstructor;
 import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
 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.core.storage.ttl.StorageTTLStatusQuery;
@@ -51,10 +52,27 @@ public TTLDefinition getTTL() {
         TTLDefinition ttlDefinition = getStorageTTLStatusQuery().getTTL();
         if (ttlDefinition == null) {
             ttlDefinition = new TTLDefinition(
-                new MetricsTTL(coreMetricsDataTTL, coreMetricsDataTTL, coreMetricsDataTTL),
+                new MetricsTTL(coreMetricsDataTTL, coreMetricsDataTTL, coreMetricsDataTTL, coreMetricsDataTTL),
                 new RecordsTTL(coreRecordDataTTL, coreRecordDataTTL, coreRecordDataTTL, coreRecordDataTTL, coreRecordDataTTL)
             );
         }
         return ttlDefinition;
     }
+
+    /**
+     * Get the effective TTL (in days) for a specific metrics model.
+     * The returned value should include both hot and warm stage TTL,
+     * representing the total period during which data is accessible.
+     *
+     * @param model the metrics model
+     * @return TTL in days, or -1 if the storage does not customize per-model TTL
+     *         (consumer falls back to core metricsDataTTL)
+     */
+    public int getMetricsTTL(Model model) {
+        int ttl = getStorageTTLStatusQuery().getMetricsTTL(model);
+        if (ttl < 0) {
+            ttl = coreMetricsDataTTL;
+        }
+        return ttl;
+    }
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/MetricsTTL.java

@@ -30,6 +30,7 @@
  */
 @Data
 public class MetricsTTL {
+    private final int metadata;
     private final int minute;
     private final int hour;
     private final int day;

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/StorageTTLStatusQuery.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.core.storage.ttl;
 
+import org.apache.skywalking.oap.server.core.storage.model.Model;
 import org.apache.skywalking.oap.server.library.module.Service;
 
 public interface StorageTTLStatusQuery extends Service {
@@ -30,4 +31,17 @@ public interface StorageTTLStatusQuery extends Service {
     default TTLDefinition getTTL() {
         return null;
     }
+
+    /**
+     * Get the effective TTL (in days) for a specific metrics model.
+     * The returned value should include both hot and warm stage TTL,
+     * representing the total period during which data is accessible.
+     *
+     * @param model the metrics model
+     * @return TTL in days, or -1 if the storage does not customize per-model TTL
+     *         (consumer falls back to core metricsDataTTL)
+     */
+    default int getMetricsTTL(Model model) {
+        return -1;
+    }
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/TTLDefinition.java

@@ -39,6 +39,7 @@ public String toString() {
         ttlDefinition.append("#\n");
         ttlDefinition.append("# TTLs for each granularity metrics are listed separately.\n");
         ttlDefinition.append("#\n");
+        ttlDefinition.append("metrics.metadata=").append(metrics.getMetadata()).append("\n");
         ttlDefinition.append("# Cover hot and warm data for BanyanDB.\n");
         ttlDefinition.append("metrics.minute=").append(metrics.getMinute()).append("\n");
         ttlDefinition.append("metrics.hour=").append(metrics.getHour()).append("\n");

oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol

@@ -18,6 +18,8 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.banyandb;
 
+import org.apache.skywalking.oap.server.core.UnexpectedException;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
 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.core.storage.ttl.StorageTTLStatusQuery;
@@ -42,6 +44,7 @@ public class BanyanDBTTLStatusQuery implements StorageTTLStatusQuery {
     private int gmColdMinuteTTLDays = -1;
     private int gmColdHourTTLDays = -1;
     private int gmColdDayTTLDays = -1;
+    private int gmMetadataTTLDays = -1;
 
     public BanyanDBTTLStatusQuery(BanyanDBStorageConfig config) {
         grNormalTTLDays = config.getRecordsNormal().getTtl();
@@ -52,6 +55,8 @@ public BanyanDBTTLStatusQuery(BanyanDBStorageConfig config) {
         gmMinuteTTLDays = config.getMetricsMin().getTtl();
         gmHourTTLDays = config.getMetricsHour().getTtl();
         gmDayTTLDays = config.getMetricsDay().getTtl();
+        gmMetadataTTLDays = config.getMetadata().getTtl();
+
         config.getRecordsNormal().getAdditionalLifecycleStages().forEach(stage -> {
             if (stage.getName().equals(BanyanDBStorageConfig.StageName.warm)) {
                 grNormalTTLDays = grNormalTTLDays + stage.getTtl();
@@ -113,7 +118,7 @@ public BanyanDBTTLStatusQuery(BanyanDBStorageConfig config) {
     @Override
     public TTLDefinition getTTL() {
         TTLDefinition definition = new TTLDefinition(
-            new MetricsTTL(gmMinuteTTLDays, gmHourTTLDays, gmDayTTLDays),
+            new MetricsTTL(gmMetadataTTLDays, gmMinuteTTLDays, gmHourTTLDays, gmDayTTLDays),
             new RecordsTTL(grNormalTTLDays, grTraceTTLDays, grZipkinTraceTTLDays, grLogTTLDays, grBrowserErrorLogTTLDays)
         );
         definition.getRecords().setColdNormal(grColdNormalTTLDays);
@@ -126,4 +131,21 @@ public TTLDefinition getTTL() {
         definition.getMetrics().setColdDay(gmColdDayTTLDays);
         return definition;
     }
+
+    @Override
+    public int getMetricsTTL(Model model) {
+        if (model.getBanyanDBModelExtension().isIndexMode()) {
+            return gmMetadataTTLDays;
+        }
+        switch (model.getDownsampling()) {
+            case Hour:
+                return gmHourTTLDays;
+            case Day:
+                return gmDayTTLDays;
+            case Minute:
+                return gmMinuteTTLDays;
+            default:
+                throw new UnexpectedException("Unexpected downsampling " + model.getDownsampling());
+        }
+    }
 }