Support ThermodynamicIndicator (#1659)

* Add 3 cases for existed indicators.

* Add Pxx indicator.

* Support All in trace analysis listener.

* Fix CI and add P99/95/90/75/50 to source ALL.

* Add ThermodynamicIndicator and its test. Also ALL source Thermodynamic metric.

* Try to fix CI about javadoc.

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/generated/all/AllDispatcher.java

@@ -35,6 +35,7 @@ public class AllDispatcher implements SourceDispatcher<All> {
         doAllP90(source);
         doAllP75(source);
         doAllP50(source);
+        doAllHeatmap(source);
     }
 
     private void doAllP99(All source) {
@@ -77,4 +78,13 @@ public class AllDispatcher implements SourceDispatcher<All> {
         indicator.combine(source.getLatency(), 10);
         IndicatorProcess.INSTANCE.in(indicator);
     }
+
+    private void doAllHeatmap(All source) {
+        AllHeatmapIndicator indicator = new AllHeatmapIndicator();
+
+
+        indicator.setTimeBucket(source.getTimeBucket());
+        indicator.combine(source.getLatency(), 100, 20);
+        IndicatorProcess.INSTANCE.in(indicator);
+    }
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/generated/all/AllHeatmapIndicator.java

+/*
+ * 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.generated.all;
+
+import java.util.*;
+import org.apache.skywalking.oap.server.core.alarm.AlarmMeta;
+import org.apache.skywalking.oap.server.core.alarm.AlarmSupported;
+import org.apache.skywalking.oap.server.core.analysis.indicator.*;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.IndicatorType;
+import org.apache.skywalking.oap.server.core.remote.annotation.StreamData;
+import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
+import org.apache.skywalking.oap.server.core.storage.annotation.*;
+import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
+import org.apache.skywalking.oap.server.core.source.Scope;
+
+/**
+ * This class is auto generated. Please don't change this class manually.
+ *
+ * @author Observability Analysis Language code generator
+ */
+@IndicatorType
+@StreamData
+@StorageEntity(name = "all_heatmap", builder = AllHeatmapIndicator.Builder.class)
+public class AllHeatmapIndicator extends ThermodynamicIndicator implements AlarmSupported {
+
+
+    @Override public String id() {
+        String splitJointId = String.valueOf(getTimeBucket());
+        return splitJointId;
+    }
+
+    @Override public int hashCode() {
+        int result = 17;
+        result = 31 * result + (int)getTimeBucket();
+        return result;
+    }
+
+    @Override public boolean equals(Object obj) {
+        if (this == obj)
+            return true;
+        if (obj == null)
+            return false;
+        if (getClass() != obj.getClass())
+            return false;
+
+        AllHeatmapIndicator indicator = (AllHeatmapIndicator)obj;
+
+        if (getTimeBucket() != indicator.getTimeBucket())
+            return false;
+
+        return true;
+    }
+
+    @Override public RemoteData.Builder serialize() {
+        RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
+
+        remoteBuilder.setDataLongs(0, getTimeBucket());
+
+
+        remoteBuilder.setDataIntegers(0, getStep());
+        remoteBuilder.setDataIntegers(1, getNumOfSteps());
+        getDetailGroup().forEach(element -> remoteBuilder.addDataIntLongPairList(element.serialize()));
+
+        return remoteBuilder;
+    }
+
+    @Override public void deserialize(RemoteData remoteData) {
+
+        setTimeBucket(remoteData.getDataLongs(0));
+
+
+        setStep(remoteData.getDataIntegers(0));
+        setNumOfSteps(remoteData.getDataIntegers(1));
+
+        setDetailGroup(new ArrayList<>(30));
+        remoteData.getDataIntLongPairListList().forEach(element -> {
+            getDetailGroup().add(new IntKeyLongValue(element.getKey(), element.getValue()));
+        });
+
+    }
+
+    @Override public AlarmMeta getAlarmMeta() {
+        return new AlarmMeta("All_heatmap", Scope.All);
+    }
+
+    public static class Builder implements StorageBuilder<AllHeatmapIndicator> {
+
+        @Override public Map<String, Object> data2Map(AllHeatmapIndicator storageData) {
+            Map<String, Object> map = new HashMap<>();
+            map.put("step", storageData.getStep());
+            map.put("num_of_steps", storageData.getNumOfSteps());
+            map.put("detail_group", storageData.getDetailGroup());
+            map.put("time_bucket", storageData.getTimeBucket());
+            return map;
+        }
+
+        @Override public AllHeatmapIndicator map2Data(Map<String, Object> dbMap) {
+            AllHeatmapIndicator indicator = new AllHeatmapIndicator();
+            indicator.setStep(((Number)dbMap.get("step")).intValue());
+            indicator.setNumOfSteps(((Number)dbMap.get("num_of_steps")).intValue());
+            indicator.setDetailGroup((List)dbMap.get("detail_group"));
+            indicator.setTimeBucket(((Number)dbMap.get("time_bucket")).longValue());
+            return indicator;
+        }
+    }
+}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/ThermodynamicIndicator.java

+/*
+ * 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.indicator;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.Arg;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.Entrance;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.IndicatorOperator;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.SourceFrom;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+
+/**
+ * Thermodynamic indicator represents the calculator for heat map.
+ *
+ * It groups the given collection of values by the given step and number of steps.
+ *
+ * A heat map (or heatmap) is a graphical representation of data where the individual values contained in a matrix are
+ * represented as colors.
+ *
+ * @author wusheng
+ */
+@IndicatorOperator
+public abstract class ThermodynamicIndicator extends Indicator {
+    protected static final String DETAIL_GROUP = "detail_group";
+    protected static final String STEP = "step";
+    protected static final String NUM_OF_STEPS = "num_of_steps";
+
+    @Getter @Setter @Column(columnName = STEP) private int step = 0;
+    @Getter @Setter @Column(columnName = NUM_OF_STEPS) private int numOfSteps = 0;
+    @Getter @Setter @Column(columnName = DETAIL_GROUP) private List<IntKeyLongValue> detailGroup = new ArrayList<>(30);
+
+    private Map<Integer, IntKeyLongValue> detailIndex;
+
+    /**
+     * Data will be grouped in
+     *
+     * [0, step), [step, step * 2), ..., [step * (maxNumOfSteps - 1), step * maxNumOfSteps), [step * maxNumOfSteps,
+     * MAX)
+     *
+     * @param value
+     * @param step the size of each step. A positive integer.
+     * @param maxNumOfSteps Steps are used to group incoming value.
+     */
+    @Entrance
+    public final void combine(@SourceFrom int value, @Arg int step, @Arg int maxNumOfSteps) {
+        if (this.step == 0) {
+            this.step = step;
+        }
+        if (this.numOfSteps == 0) {
+            this.numOfSteps = maxNumOfSteps + 1;
+        }
+
+        indexCheckAndInit();
+
+        int index = value / step;
+        if (index > maxNumOfSteps) {
+            index = numOfSteps;
+        }
+        IntKeyLongValue element = detailIndex.get(index);
+        if (element == null) {
+            element = new IntKeyLongValue();
+            element.setKey(index);
+            element.setValue(1);
+            addElement(element);
+        } else {
+            element.addValue(1);
+        }
+    }
+
+    @Override
+    public void combine(Indicator indicator) {
+        ThermodynamicIndicator thermodynamicIndicator = (ThermodynamicIndicator)indicator;
+        this.indexCheckAndInit();
+        thermodynamicIndicator.indexCheckAndInit();
+
+        thermodynamicIndicator.detailIndex.forEach((key, element) -> {
+            IntKeyLongValue existingElement = this.detailIndex.get(key);
+            if (existingElement == null) {
+                existingElement = new IntKeyLongValue();
+                existingElement.setKey(key);
+                existingElement.setValue(element.getValue());
+                addElement(element);
+            } else {
+                existingElement.addValue(element.getValue());
+            }
+        });
+    }
+
+    /**
+     * For Thermodynamic indicator, no single value field. Need to do nothing here.
+     */
+    @Override
+    public final void calculate() {
+
+    }
+
+    private void addElement(IntKeyLongValue element) {
+        detailGroup.add(element);
+        detailIndex.put(element.getKey(), element);
+    }
+
+    private void indexCheckAndInit() {
+        if (detailIndex == null) {
+            detailIndex = new HashMap<>();
+            detailGroup.forEach(element -> detailIndex.put(element.getKey(), element));
+        }
+    }
+}

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/indicator/ThermodynamicIndicatorTest.java

+/*
+ * 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.indicator;
+
+import java.util.Map;
+import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
+import org.junit.Assert;
+import org.junit.Test;
+import org.powermock.reflect.Whitebox;
+
+/**
+ * @author wusheng
+ */
+public class ThermodynamicIndicatorTest {
+    private int step = 10;//ms
+    private int maxNumOfSteps = 10;//count
+
+    @Test
+    public void testEntrance() {
+        ThermodynamicIndicatorTest.ThermodynamicIndicatorMocker indicatorMocker = new ThermodynamicIndicatorTest.ThermodynamicIndicatorMocker();
+
+        indicatorMocker.combine(2000, step, maxNumOfSteps);
+        indicatorMocker.combine(110, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+        indicatorMocker.combine(50, step, maxNumOfSteps);
+        indicatorMocker.combine(50, step, maxNumOfSteps);
+        indicatorMocker.combine(28, step, maxNumOfSteps);
+        indicatorMocker.combine(50, step, maxNumOfSteps);
+        indicatorMocker.combine(61, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+
+        Map<Integer, IntKeyLongValue> index = Whitebox.getInternalState(indicatorMocker, "detailIndex");
+        Assert.assertEquals(5, index.size());
+
+        Assert.assertEquals(1, index.get(2).getValue());
+        Assert.assertEquals(3, index.get(5).getValue());
+        Assert.assertEquals(1, index.get(6).getValue());
+        Assert.assertEquals(6, index.get(10).getValue());
+        Assert.assertEquals(2, index.get(11).getValue());
+    }
+
+    @Test
+    public void testMerge() {
+        ThermodynamicIndicatorTest.ThermodynamicIndicatorMocker indicatorMocker = new ThermodynamicIndicatorTest.ThermodynamicIndicatorMocker();
+
+        indicatorMocker.combine(2000, step, maxNumOfSteps);
+        indicatorMocker.combine(110, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+        indicatorMocker.combine(100, step, maxNumOfSteps);
+        indicatorMocker.combine(50, step, maxNumOfSteps);
+        indicatorMocker.combine(50, step, maxNumOfSteps);
+
+        ThermodynamicIndicatorTest.ThermodynamicIndicatorMocker indicatorMocker2 = new ThermodynamicIndicatorTest.ThermodynamicIndicatorMocker();
+
+        indicatorMocker2.combine(28, step, maxNumOfSteps);
+        indicatorMocker2.combine(50, step, maxNumOfSteps);
+        indicatorMocker2.combine(61, step, maxNumOfSteps);
+        indicatorMocker2.combine(100, step, maxNumOfSteps);
+        indicatorMocker2.combine(100, step, maxNumOfSteps);
+        indicatorMocker2.combine(100, step, maxNumOfSteps);
+
+        indicatorMocker.combine(indicatorMocker2);
+
+        Map<Integer, IntKeyLongValue> index = Whitebox.getInternalState(indicatorMocker, "detailIndex");
+        Assert.assertEquals(5, index.size());
+
+        Assert.assertEquals(1, index.get(2).getValue());
+        Assert.assertEquals(3, index.get(5).getValue());
+        Assert.assertEquals(1, index.get(6).getValue());
+        Assert.assertEquals(6, index.get(10).getValue());
+        Assert.assertEquals(2, index.get(11).getValue());
+    }
+
+    public class ThermodynamicIndicatorMocker extends ThermodynamicIndicator {
+
+        @Override public String id() {
+            return null;
+        }
+
+        @Override public void deserialize(RemoteData remoteData) {
+
+        }
+
+        @Override public RemoteData.Builder serialize() {
+            return null;
+        }
+    }
+}