From 12b0978daa665957be9ec5386b5e13f4b904c8f0 Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Mon, 19 Oct 2026 22:15:51 +0800 Subject: [PATCH] Fix expired records not being dropped on receipt RecordStreamProcessor decoded a record's second-precision time bucket (yyyyMMddHHmmss) as a minute bucket, which put the record's time far in the future, and compared the record's age in milliseconds against recordDataTTL converted to microseconds, so no record was ever treated as expired. Decode the bucket at second precision and convert the TTL to milliseconds, as the metrics check in IMetricsDAO.isExpiredCache does. The BanyanDB stages data generator seeds the cold node with records stamped 3 days in the past, so its OAP raises recordDataTTL to 5 days, matching the BanyanDB TTLs it sets; the stages case keeps the default. Co-authored-by: Wu Sheng --- docs/en/changes/changes.md | 1 + .../worker/RecordStreamProcessor.java | 4 +- .../worker/RecordStreamProcessorTest.java | 77 +++++++++++++++++++ .../banyandb/data-generate/docker-compose.yml | 4 + 4 files changed, 84 insertions(+), 2 deletions(-) create mode 100644 oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessorTest.java diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index 1a3aeb6efa45..9a6391a2681b 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -58,6 +58,7 @@ * Fix OAL `>`, `>=`, `<` and `<=` filters on a `long` or `double` source field, and `==` on a `long` one, failing to compile when they compare it with a number, such as `filter(sideCar.internalRequestLatencyNanos >= 0)`. The generated code passed the number as an `int`, and the comparison matchers have no `match(long, int)` or `match(double, int)` overload, as Javassist does not widen an argument. The number is now written in the field's type, `0L` or `0.0`; filters on `int` fields, which already worked, generate the same code. * Add `ZIPKIN` to the GenAI evaluation record's `traceRef.type`, for spans received by `receiver-zipkin` and OTLP spans converted with `otlpTraceStorage: zipkin`, so the type names the store holding the judged span: `SKYWALKING_NATIVE`, `OTLP` for OTLP spans stored natively, or `ZIPKIN`. Zipkin and OTLP spans are both addressed by span ID, and one OAP can hold both, so an `OTLP` reference alone could not tell a client which store to read the trace from. The type is decided when the span is received, so records written under different `otlpTraceStorage` settings each name their own store. * Fix the Zabbix receiver dropping the sub-second part of agent sample times. Each `agent data` item carries `clock` (Unix seconds) and `ns` (nanoseconds within that second), but the sample time used only `clock * 1000`; it is now `clock * 1000 + ns / 1_000_000`. The sub-second precision can affect values produced by custom Zabbix rules using MAL `rate`, `irate` or `increase`; results from the bundled Zabbix rules and the assignment of samples to minute buckets are unchanged. +* Fix expired records never being dropped on receipt. `RecordStreamProcessor` decoded a record's second-precision time bucket as a minute bucket, which put the record's time far in the future, so no record was ever treated as expired. It also compared the record's age against `recordDataTTL` converted to microseconds instead of milliseconds. #### UI * Add a Virtual GenAI evaluation-record page and evaluation-score chart in Horizon UI, so operators can inspect evaluation result, level, reason, judge model, timestamp, trace linkage, and the `gen_ai_model_evaluation_score_ppm` trend for evaluated records. diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java index 0665f327cd95..464141f049d4 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java @@ -65,8 +65,8 @@ public static RecordStreamProcessor getInstance() { @Override public void in(Record record) { final var now = System.currentTimeMillis(); - final var recordTimestamp = TimeBucket.getTimestamp(record.getTimeBucket(), DownSampling.Minute); - final var isExpired = now - recordTimestamp > TimeUnit.DAYS.toMicros(recordDataTTL); + final var recordTimestamp = TimeBucket.getTimestamp(record.getTimeBucket(), DownSampling.Second); + final var isExpired = now - recordTimestamp > TimeUnit.DAYS.toMillis(recordDataTTL); if (isExpired && !isTestingTTL) { log.debug("Receiving expired record: {}, time: {}, ignored", record.id(), record.getTimeBucket()); return; diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessorTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessorTest.java new file mode 100644 index 000000000000..c4c5ec4ac12c --- /dev/null +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessorTest.java @@ -0,0 +1,77 @@ +/* + * 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.analysis.worker; + +import java.lang.reflect.Field; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.skywalking.oap.server.core.analysis.TimeBucket; +import org.apache.skywalking.oap.server.core.analysis.record.Record; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class RecordStreamProcessorTest { + private static final int TTL_DAYS = 3; + + private RecordStreamProcessor processor; + private RecordPersistentWorker worker; + private Record record; + + @BeforeEach + @SuppressWarnings("unchecked") + void setUp() throws Exception { + processor = new RecordStreamProcessor(); + processor.setRecordDataTTL(TTL_DAYS); + worker = mock(RecordPersistentWorker.class); + record = mock(Record.class); + + final Field workersField = RecordStreamProcessor.class.getDeclaredField("workers"); + workersField.setAccessible(true); + ((Map, RecordPersistentWorker>) workersField.get(processor)) + .put(record.getClass(), worker); + } + + @Test + void acceptsRecordWithinTTL() { + givenRecordAged(TimeUnit.DAYS.toMillis(TTL_DAYS) - TimeUnit.MINUTES.toMillis(2)); + + processor.in(record); + + verify(worker).in(record); + } + + @Test + void dropsRecordBeyondTTL() { + givenRecordAged(TimeUnit.DAYS.toMillis(TTL_DAYS) + TimeUnit.MINUTES.toMillis(2)); + + processor.in(record); + + verifyNoInteractions(worker); + } + + private void givenRecordAged(final long ageMillis) { + when(record.getTimeBucket()).thenReturn( + TimeBucket.getRecordTimeBucket(System.currentTimeMillis() - ageMillis)); + } +} diff --git a/test/e2e-v2/cases/storage/banyandb/data-generate/docker-compose.yml b/test/e2e-v2/cases/storage/banyandb/data-generate/docker-compose.yml index 29a4ed98d574..fa0e8b63cbbb 100644 --- a/test/e2e-v2/cases/storage/banyandb/data-generate/docker-compose.yml +++ b/test/e2e-v2/cases/storage/banyandb/data-generate/docker-compose.yml @@ -34,6 +34,10 @@ services: service: oap environment: SW_STORAGE: banyandb + # The mockers send records MINUS_DAYS (3) in the past, past the default 3-day recordDataTTL that + # RecordStreamProcessor drops on receipt. Only this generator OAP accepts them; the stages case + # runs with the default TTL and reads them from the files mounted on its cold node. + SW_CORE_RECORD_DATA_TTL: 5 SW_STORAGE_BANYANDB_RECORDS_TTL_DAYS: 5 SW_STORAGE_BANYANDB_TRACE_TTL_DAYS: 5 SW_STORAGE_BANYANDB_ZIPKIN_TRACE_TTL_DAYS: 5