Repository: apache/skywalking (language: python)

Issue:
[Feature] add sumLabeled @MeterFunction for mal sum downsampling
### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/skywalking/issues?q=is%3Aissue) and found no similar feature requirement.


### Description

This is my scenario：
1. otel-collector scrape gateway metric endpoint in every 15S
2. skywalking mal engine calculate every http code count increasement in this 15s and downsampling by sum method
3. skywalking will persist this metric in meter_sum_xxxxxxxx table in value meter mode
```
metricsRules:
  - name: http_200_code_count
    exp: http_server_requests_seconds_count.tagEqual('status', '200').sum(['application']).increase('PT15S').service(['application'], Layer.GENERAL).downsampling(SUM)
  - name: http_404_code_count
    exp: http_server_requests_seconds_count.tagEqual('status', '404').sum(['application']).increase('PT15S').service(['application'], Layer.GENERAL).downsampling(SUM)
  ...
  ...
  ...
```

### Use case

I want to change my mal like this, the sum aggregate method use 'application' and 'status' as aggregate-by params. If skywalking meter system support sumlabeled function, this metric will be persisted in meter_sumlabeled_xxxxxxxx table in datatable meter mode. 

The  sumlabeled function is easy to extend under skywalking meter system excellent architecture design, and it can bring us benefits in query efficiency and storage economy.

```
metricsRules:
  - name: http_code_count
    exp: http_server_requests_seconds_count.sum(['application','status']).increase('PT15S').service(['application'], Layer.GENERAL).downsampling(SUM)
```

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)



Discussion / hints:
I notice that someone already submit this feature #10632 , sorry about this.
Am I using mal in wrong way and are there any ready-made mal function can support my scenario？
@BFergerson was adding that, but we don't know what is the use case. Then, this should be OK to add for now.
@wu-sheng, you were actually right. It turned out I didn't need it. Glad someone found a use though.
@weixiang1862 Are you going to open the pull request soon? Or should I move this to discussion only
@wu-sheng yes, I will submit the pull request today. This is my first time participating in a open source project, thank you for the opportunity.

Fix patch:
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 164a03fae5dd..9a205470b34c 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -57,6 +57,7 @@
 * Support HTTP URI formatting through `ai-pipeline` to do pattern recognition.
 * Add new HTTP URI grouping engine with benchmark. 
 * [Breaking Change] Use the new HTTP URI grouping engine to replace the old regex based mechanism.
+* support `sumLabeled` in `MAL`.
 
 #### UI
 * Revert: cpm5d function. This feature is cancelled from backend.
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/sum/SumLabeledFunction.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/sum/SumLabeledFunction.java
new file mode 100644
index 000000000000..cf9b8e61e75b
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/sum/SumLabeledFunction.java
@@ -0,0 +1,191 @@
+/*
+ * 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.meter.function.sum;
+
+import java.util.Objects;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+import org.apache.skywalking.oap.server.core.UnexpectedException;
+import org.apache.skywalking.oap.server.core.analysis.manual.instance.InstanceTraffic;
+import org.apache.skywalking.oap.server.core.analysis.meter.Meter;
+import org.apache.skywalking.oap.server.core.analysis.meter.MeterEntity;
+import org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableValue;
+import org.apache.skywalking.oap.server.core.analysis.meter.function.MeterFunction;
+import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
+import org.apache.skywalking.oap.server.core.analysis.metrics.LabeledValueHolder;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
+import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.Entrance;
+import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.SourceFrom;
+import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
+import org.apache.skywalking.oap.server.core.storage.StorageID;
+import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.type.Convert2Entity;
+import org.apache.skywalking.oap.server.core.storage.type.Convert2Storage;
+import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
+
+@MeterFunction(functionName = "sumLabeled")
+@ToString
+public abstract class SumLabeledFunction extends Meter implements AcceptableValue<DataTable>, LabeledValueHolder {
+
+    protected static final String VALUE = "datatable_value";
+
+    @Setter
+    @Getter
+    @Column(name = ENTITY_ID, length = 512)
+    @BanyanDB.SeriesID(index = 0)
+    private String entityId;
+
+    @Setter
+    @Getter
+    @Column(name = InstanceTraffic.SERVICE_ID)
+    private String serviceId;
+
+    @Getter
+    @Setter
+    @Column(name = VALUE, dataType = Column.ValueDataType.LABELED_VALUE, storageOnly = true)
+    @BanyanDB.MeasureField
+    private DataTable value = new DataTable(30);
+
+    @Entrance
+    public final void combine(@SourceFrom DataTable value) {
+        this.value.append(value);
+    }
+
+    @Override
+    public final boolean combine(Metrics metrics) {
+        SumLabeledFunction sumLabeledFunction = (SumLabeledFunction) metrics;
+        value.append(sumLabeledFunction.value);
+        return true;
+    }
+
+    @Override
+    public final void calculate() {
+
+    }
+
+    @Override
+    public Metrics toHour() {
+        SumLabeledFunction metrics = (SumLabeledFunction) createNew();
+        metrics.setEntityId(getEntityId());
+        metrics.setTimeBucket(toTimeBucketInHour());
+        metrics.setServiceId(getServiceId());
+        metrics.getValue().copyFrom(getValue());
+        return metrics;
+    }
+
+    @Override
+    public Metrics toDay() {
+        SumLabeledFunction metrics = (SumLabeledFunction) createNew();
+        metrics.setEntityId(getEntityId());
+        metrics.setTimeBucket(toTimeBucketInDay());
+        metrics.setServiceId(getServiceId());
+        metrics.getValue().copyFrom(getValue());
+        return metrics;
+    }
+
+    @Override
+    public int remoteHashCode() {
+        return entityId.hashCode();
+    }
+
+    @Override
+    public void deserialize(final RemoteData remoteData) {
+        setValue(new DataTable(remoteData.getDataObjectStrings(0)));
+        setTimeBucket(remoteData.getDataLongs(0));
+
+        this.entityId = remoteData.getDataStrings(0);
+        this.serviceId = remoteData.getDataStrings(1);
+    }
+
+    @Override
+    public RemoteData.Builder serialize() {
+        RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
+        remoteBuilder.addDataObjectStrings(value.toStorageData());
+        remoteBuilder.addDataLongs(getTimeBucket());
+
+        remoteBuilder.addDataStrings(entityId);
+        remoteBuilder.addDataStrings(serviceId);
+
+        return remoteBuilder;
+    }
+
+    @Override
+    protected StorageID id0() {
+        return new StorageID()
+            .append(TIME_BUCKET, getTimeBucket())
+            .append(ENTITY_ID, getEntityId());
+    }
+
+    @Override
+    public void accept(final MeterEntity entity, final DataTable value) {
+        this.entityId = entity.id();
+        this.serviceId = entity.serviceId();
+        this.value.append(value);
+    }
+
+    @Override
+    public Class<? extends SumLabeledStorageBuilder> builder() {
+        return SumLabeledStorageBuilder.class;
+    }
+
+    public static class SumLabeledStorageBuilder implements StorageBuilder<SumLabeledFunction> {
+        @Override
+        public SumLabeledFunction storage2Entity(final Convert2Entity converter) {
+            SumLabeledFunction metrics = new SumLabeledFunction() {
+                @Override
+                public AcceptableValue<DataTable> createNew() {
+                    throw new UnexpectedException("createNew should not be called");
+                }
+            };
+            metrics.setValue(new DataTable((String) converter.get(VALUE)));
+            metrics.setTimeBucket(((Number) converter.get(TIME_BUCKET)).longValue());
+            metrics.setServiceId((String) converter.get(InstanceTraffic.SERVICE_ID));
+            metrics.setEntityId((String) converter.get(ENTITY_ID));
+            return metrics;
+        }
+
+        @Override
+        public void entity2Storage(final SumLabeledFunction storageData, final Convert2Storage converter) {
+            converter.accept(VALUE, storageData.getValue());
+            converter.accept(TIME_BUCKET, storageData.getTimeBucket());
+            converter.accept(InstanceTraffic.SERVICE_ID, storageData.getServiceId());
+            converter.accept(ENTITY_ID, storageData.getEntityId());
+        }
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof SumLabeledFunction)) {
+            return false;
+        }
+        SumLabeledFunction function = (SumLabeledFunction) o;
+        return Objects.equals(entityId, function.entityId) &&
+            getTimeBucket() == function.getTimeBucket();
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(entityId, getTimeBucket());
+    }
+}


Test patch:
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/sum/SumLabeledFunctionTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/sum/SumLabeledFunctionTest.java
new file mode 100644
index 000000000000..ad83b0dcd2b2
--- /dev/null
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/sum/SumLabeledFunctionTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.meter.function.sum;
+
+import java.util.Map;
+import org.apache.skywalking.oap.server.core.analysis.Layer;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.analysis.meter.MeterEntity;
+import org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableValue;
+import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
+import org.apache.skywalking.oap.server.core.config.NamingControl;
+import org.apache.skywalking.oap.server.core.config.group.EndpointNameGrouping;
+import org.apache.skywalking.oap.server.core.storage.type.HashMapConverter;
+import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.skywalking.oap.server.core.analysis.meter.function.sum.SumLabeledFunction.VALUE;
+
+@ExtendWith(MockitoExtension.class)
+public class SumLabeledFunctionTest {
+
+    private SumLabeledFunction function;
+
+    private static final DataTable HTTP_CODE_COUNT_1 = new DataTable("200,1|301,2|404,3|502,4");
+    private static final DataTable HTTP_CODE_COUNT_2 = new DataTable("200,2|404,4|502,5|505,1");
+
+    @BeforeAll
+    public static void setup() {
+        MeterEntity.setNamingControl(
+            new NamingControl(512, 512, 512, new EndpointNameGrouping()));
+    }
+
+    @BeforeEach
+    public void before() {
+        function = new SumLabeledFunctionInst();
+        function.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
+    }
+
+    @AfterAll
+    public static void tearDown() {
+        MeterEntity.setNamingControl(null);
+    }
+
+    @Test
+    public void testAccept() {
+        function.accept(
+            MeterEntity.newService("service-test", Layer.GENERAL),
+            HTTP_CODE_COUNT_1
+        );
+        function.accept(
+            MeterEntity.newService("service-test", Layer.GENERAL),
+            HTTP_CODE_COUNT_2
+        );
+
+        Assertions.assertEquals(function.getValue(), new DataTable("200,3|301,2|404,7|502,9|505,1"));
+    }
+
+    @Test
+    public void testToHour() {
+        function.accept(
+            MeterEntity.newService("service-test", Layer.GENERAL),
+            HTTP_CODE_COUNT_1
+        );
+        function.accept(
+            MeterEntity.newService("service-test", Layer.GENERAL),
+            HTTP_CODE_COUNT_2
+        );
+        function.calculate();
+
+        final SumLabeledFunction hourFunction = (SumLabeledFunction) function.toHour();
+        hourFunction.calculate();
+
+        Assertions.assertEquals(hourFunction.getValue(), new DataTable("200,3|301,2|404,7|502,9|505,1"));
+    }
+
+    @Test
+    public void testToDay() {
+        function.accept(
+            MeterEntity.newService("service-test", Layer.GENERAL),
+            HTTP_CODE_COUNT_1
+        );
+        function.accept(
+            MeterEntity.newService("service-test", Layer.GENERAL),
+            HTTP_CODE_COUNT_2
+        );
+        function.calculate();
+
+        final SumLabeledFunction dayFunction = (SumLabeledFunction) function.toDay();
+        dayFunction.calculate();
+
+        Assertions.assertEquals(dayFunction.getValue(), new DataTable("200,3|301,2|404,7|502,9|505,1"));
+    }
+
+    @Test
+    public void testSerialization() {
+        function.accept(
+            MeterEntity.newService("service-test", Layer.GENERAL),
+            HTTP_CODE_COUNT_1
+        );
+
+        SumLabeledFunction function2 = new SumLabeledFunctionInst();
+        function2.deserialize(function.serialize().build());
+
+        Assertions.assertEquals(function, function2);
+        Assertions.assertEquals(function.getValue(), function2.getValue());
+    }
+
+    @Test
+    public void testBuilder() throws IllegalAccessException, InstantiationException {
+        function.accept(
+            MeterEntity.newService("service-test", Layer.GENERAL),
+            HTTP_CODE_COUNT_1
+        );
+        function.calculate();
+
+        StorageBuilder<SumLabeledFunction> storageBuilder = function.builder().newInstance();
+
+        final HashMapConverter.ToStorage toStorage = new HashMapConverter.ToStorage();
+        storageBuilder.entity2Storage(function, toStorage);
+        final Map<String, Object> map = toStorage.obtain();
+        map.put(VALUE, ((DataTable) map.get(VALUE)).toStorageData());
+
+        SumLabeledFunction function2 = storageBuilder.storage2Entity(new HashMapConverter.ToEntity(map));
+
+        Assertions.assertEquals(function, function2);
+        Assertions.assertEquals(function2.getValue(), function2.getValue());
+    }
+
+    private static class SumLabeledFunctionInst extends SumLabeledFunction {
+        @Override
+        public AcceptableValue<DataTable> createNew() {
+            return new SumLabeledFunctionInst();
+        }
+    }
+}
\ No newline at end of file