diff --git a/CHANGES.md b/CHANGES.md index 598f0d465..dae2bc0fa 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -42,6 +42,7 @@ Release Notes. * Upgrade commons-lang3 to avoid potential NPE in some JDK versions. * OAL supports generating metrics from events. * Support endpoint name grouping by OpenAPI definitions. +* Fix CounterWindow increase computing issue. #### UI * Fix the date component for log conditions. diff --git a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/Sample.java b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/Sample.java index 3e716a16f..ded0dfdc1 100644 --- a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/Sample.java +++ b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/Sample.java @@ -51,4 +51,10 @@ public class Sample { double nv = transform.apply(i._2, i._1); return newValue(ignored -> nv); } + + Sample increase(Function2 transform) { + Tuple2 i = CounterWindow.INSTANCE.pop(name, labels, value, timestamp); + double nv = transform.apply(i._2, i._1); + return newValue(ignored -> nv); + } } diff --git a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java index b1f887010..f38887bf6 100644 --- a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java +++ b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java @@ -288,7 +288,20 @@ public class SampleFamily { } public SampleFamily irate() { - return rate("PT1S"); + if (this == EMPTY) { + return EMPTY; + } + return SampleFamily.build( + this.context, + Arrays.stream(samples) + .map(sample -> sample.increase( + (lowerBoundValue, lowerBoundTime) -> { + final long timeDiff = (sample.timestamp - lowerBoundTime) / 1000; + return timeDiff < 1L ? 0.0 : (sample.value - lowerBoundValue) / timeDiff; + } + )) + .toArray(Sample[]::new) + ); } @SuppressWarnings(value = "unchecked") diff --git a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/counter/CounterWindow.java b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/counter/CounterWindow.java index 46ac2a56e..ecdc32941 100644 --- a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/counter/CounterWindow.java +++ b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/counter/CounterWindow.java @@ -22,8 +22,8 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; import io.vavr.Tuple; import io.vavr.Tuple2; -import java.util.LinkedList; import java.util.Map; +import java.util.PriorityQueue; import java.util.Queue; import lombok.AccessLevel; import lombok.EqualsAndHashCode; @@ -42,20 +42,47 @@ public class CounterWindow { public static final CounterWindow INSTANCE = new CounterWindow(); + private final Map> lastElementMap = Maps.newHashMap(); private final Map>> windows = Maps.newHashMap(); public Tuple2 increase(String name, ImmutableMap labels, Double value, long windowSize, long now) { ID id = new ID(name, labels); if (!windows.containsKey(id)) { - windows.put(id, new LinkedList<>()); + windows.put(id, new PriorityQueue<>()); } + Queue> window = windows.get(id); window.offer(Tuple.of(now, value)); - Tuple2 ps = window.element(); - if ((now - ps._1) >= windowSize) { - window.remove(); + long waterLevel = now - windowSize; + Tuple2 peek = window.peek(); + if (peek._1 > waterLevel) { + return peek; } - return ps; + + Tuple2 result = peek; + while (peek._1 < waterLevel) { + result = window.poll(); + peek = window.element(); + } + + // Choose the closed slot to the expected timestamp + if (waterLevel - result._1 <= peek._1 - waterLevel) { + return result; + } + + return peek; + } + + public Tuple2 pop(String name, ImmutableMap labels, Double value, long now) { + ID id = new ID(name, labels); + + Tuple2 element = Tuple.of(now, value); + Tuple2 result = lastElementMap.get(id); + lastElementMap.put(id, element); + if (result == null) { + return element; + } + return result; } public void reset() { diff --git a/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/counter/CounterWindowTest.java b/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/counter/CounterWindowTest.java new file mode 100644 index 000000000..36e7749b1 --- /dev/null +++ b/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/counter/CounterWindowTest.java @@ -0,0 +1,96 @@ +/* + * 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.meter.analyzer.dsl.counter; + +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Lists; +import io.vavr.Tuple2; +import java.time.Duration; +import java.util.List; +import org.junit.Assert; +import org.junit.Test; + +import static java.time.Instant.parse; + +public class CounterWindowTest { + + public static List> parameters() { + return Lists.newArrayList( + new Tuple2<>(parse("2020-09-11T11:11:01.03Z").toEpochMilli(), 10d), + new Tuple2<>(parse("2020-09-11T11:11:15.99Z").toEpochMilli(), 11d), + new Tuple2<>(parse("2020-09-11T11:11:31.00Z").toEpochMilli(), 12d), + new Tuple2<>(parse("2020-09-11T11:11:46.09Z").toEpochMilli(), 13d), + new Tuple2<>(parse("2020-09-11T11:12:00.97Z").toEpochMilli(), 14d), + new Tuple2<>(parse("2020-09-11T11:11:00.97Z").toEpochMilli(), 15d), + new Tuple2<>(parse("2020-09-11T11:12:16.60Z").toEpochMilli(), 16d), + new Tuple2<>(parse("2020-09-11T11:12:31.66Z").toEpochMilli(), 17d) + ); + } + + @Test + public void testPT15S() { + double[] actuals = parameters().stream().mapToDouble(e -> { + Tuple2 increase = CounterWindow.INSTANCE.increase( + "test", ImmutableMap.builder().build(), e._2, + Duration.parse("PT15S").getSeconds() * 1000, e._1 + ); + return e._2 - increase._2; + }).toArray(); + + Assert.assertArrayEquals(new double[] {0, 1d, 1d, 1d, 1d, 0d, 2d, 1d}, actuals, 0.d); + } + + @Test + public void testPT35S() { + double[] actuals = parameters().stream().mapToDouble(e -> { + Tuple2 increase = CounterWindow.INSTANCE.increase( + "test", ImmutableMap.builder().build(), e._2, + Duration.parse("PT35S").getSeconds() * 1000, e._1 + ); + return e._2 - increase._2; + }).toArray(); + + Assert.assertArrayEquals(new double[] {0, 1d, 2d, 2d, 2d, 0d, 3d, 3d}, actuals, 0.d); + } + + @Test + public void testPT1M() { + double[] actuals = parameters().stream().mapToDouble(e -> { + Tuple2 increase = CounterWindow.INSTANCE.increase( + "test", ImmutableMap.builder().build(), e._2, + Duration.parse("PT1M").getSeconds() * 1000, e._1 + ); + return e._2 - increase._2; + }).toArray(); + + Assert.assertArrayEquals(new double[] {0, 1d, 2d, 3d, 4d, 0d, 5d, 5d}, actuals, 0.d); + } + + @Test + public void testPT2M() { + double[] actuals = parameters().stream().mapToDouble(e -> { + Tuple2 increase = CounterWindow.INSTANCE.increase( + "test", ImmutableMap.builder().build(), e._2, + Duration.parse("PT2M").getSeconds() * 1000, e._1 + ); + return e._2 - increase._2; + }).toArray(); + + Assert.assertArrayEquals(new double[] {0, 1d, 2d, 3d, 4d, 0d, 1d, 2d}, actuals, 0.d); + } +}