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