Add instance properties extractor in MAL (#9151)

docs/en/changes/changes.md

@@ -75,6 +75,7 @@
 * Add APIs to query Pod log on demand.
 * Remove OAL for events.
 * Simplify the format index name logical in ES storage.
+* Add instance properties extractor in MAL.
 
 #### UI
 

docs/en/concepts-and-designs/mal.md

 # Meter Analysis Language
 
-The meter system provides a functional analysis language called MAL (Meter Analysis Language) that lets users analyze and 
+The meter system provides a functional analysis language called MAL (Meter Analysis Language) that lets users analyze and
 aggregate meter data in the OAP streaming system. The result of an expression can either be ingested by the agent analyzer,
 or the OC/Prometheus analyzer.
 
@@ -66,7 +66,7 @@ This feature requires authorizing the OAP Server to access K8s's `API Server`.
 
 ##### retagByK8sMeta
 `retagByK8sMeta(newLabelName, K8sRetagType, existingLabelName, namespaceLabelName)`. Add a new tag to the sample family based on the value of an existing label. Provide several internal converting types, including
-- K8sRetagType.Pod2Service  
+- K8sRetagType.Pod2Service
 
 Add a tag to the sample using `service` as the key, `$serviceName.$namespace` as the value, and according to the given value of the tag key, which represents the name of a pod.
 
@@ -104,13 +104,13 @@ Between a sample family and a scalar, the operator is applied to the value of ev
 
 ```
 instance_trace_count + 2
-``` 
+```
 
-or 
+or
 
 ```
 2 + instance_trace_count
-``` 
+```
 
 results in
 
@@ -120,15 +120,15 @@ instance_trace_count{region="us-east",az="az-3"} 22 // 20 + 2
 instance_trace_count{region="asia-north",az="az-1"} 35 // 33 + 2
 ```
 
-Between two sample families, a binary operator is applied to each sample in the sample family on the left and 
+Between two sample families, a binary operator is applied to each sample in the sample family on the left and
 its matching sample in the sample family on the right. A new sample family with empty name will be generated.
 Only the matched tags will be reserved. Samples with no matching samples in the sample family on the right will not be found in the result.
 
-Another sample family `instance_trace_analysis_error_count` is 
+Another sample family `instance_trace_analysis_error_count` is
 
 ```
 instance_trace_analysis_error_count{region="us-west",az="az-1"} 20
-instance_trace_analysis_error_count{region="asia-north",az="az-1"} 11 
+instance_trace_analysis_error_count{region="asia-north",az="az-1"} 11
 ```
 
 Example expression:
@@ -137,7 +137,7 @@ Example expression:
 instance_trace_analysis_error_count / instance_trace_count
 ```
 
-This returns a resulting sample family containing the error rate of trace analysis. Samples with region us-west and az az-3 
+This returns a resulting sample family containing the error rate of trace analysis. Samples with region us-west and az az-3
 have no match and will not show up in the result:
 
 ```
@@ -154,8 +154,8 @@ resulting in a new sample family having fewer samples (sometimes having just a s
  - min (select minimum over dimensions)
  - max (select maximum over dimensions)
  - avg (calculate the average over dimensions)
- 
-These operations can be used to aggregate overall label dimensions or preserve distinct dimensions by inputting `by` parameter. 
+
+These operations can be used to aggregate overall label dimensions or preserve distinct dimensions by inputting `by` parameter.
 
 ```
 <aggr-op>(by: <tag1, tag2, ...>)
@@ -202,19 +202,19 @@ Examples:
 `tag({allTags -> })`: Updates tags of samples. User can add, drop, rename and update tags.
 
 #### histogram
-`histogram(le: '<the tag name of le>')`: Transforms less-based histogram buckets to meter system histogram buckets. 
-`le` parameter represents the tag name of the bucket. 
+`histogram(le: '<the tag name of le>')`: Transforms less-based histogram buckets to meter system histogram buckets.
+`le` parameter represents the tag name of the bucket.
 
 #### histogram_percentile
-`histogram_percentile([<p scalar>])`. Represents the meter-system to calculate the p-percentile (0 ≤ p ≤ 100) from the buckets. 
+`histogram_percentile([<p scalar>])`. Represents the meter-system to calculate the p-percentile (0 ≤ p ≤ 100) from the buckets.
 
 #### time
 `time()`: Returns the number of seconds since January 1, 1970 UTC.
 
 
 ## Down Sampling Operation
-MAL should instruct meter-system on how to downsample for metrics. It doesn't only refer to aggregate raw samples to 
-`minute` level, but also expresses data from `minute` in higher levels, such as `hour` and `day`. 
+MAL should instruct meter-system on how to downsample for metrics. It doesn't only refer to aggregate raw samples to
+`minute` level, but also expresses data from `minute` in higher levels, such as `hour` and `day`.
 
 Down sampling function is called `downsampling` in MAL, and it accepts the following types:
 
@@ -239,13 +239,13 @@ last_server_state_sync_time_in_seconds.tagEqual('production', 'catalog').downsam
 They extract level relevant labels from metric labels, then informs the meter-system the level and [layer](../../../oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Layer.java) to which this metric belongs.
 
  - `service([svc_label1, svc_label2...], Layer)` extracts service level labels from the array argument, extracts layer from `Layer` argument.
- - `instance([svc_label1, svc_label2...], [ins_label1, ins_label2...], Layer)` extracts service level labels from the first array argument, 
-                                                                        extracts instance level labels from the second array argument, extracts layer from `Layer` argument.
- - `endpoint([svc_label1, svc_label2...], [ep_label1, ep_label2...])` extracts service level labels from the first array argument, 
+ - `instance([svc_label1, svc_label2...], [ins_label1, ins_label2...], Layer, Closure<Map<String, String>> propertiesExtractor)` extracts service level labels from the first array argument,
+                                                                        extracts instance level labels from the second array argument, extracts layer from `Layer` argument, `propertiesExtractor` is an optional closure that extracts instance properties from `tags`, e.g. `{ tags -> ['pod': tags.pod, 'namespace': tags.namespace] }`.
+ - `endpoint([svc_label1, svc_label2...], [ep_label1, ep_label2...])` extracts service level labels from the first array argument,
                                                                       extracts endpoint level labels from the second array argument, extracts layer from `Layer` argument.
- - `serviceRelation(DetectPoint, [source_svc_label1...], [dest_svc_label1...], Layer)` DetectPoint including `DetectPoint.CLIENT` and `DetectPoint.SERVER`, 
+ - `serviceRelation(DetectPoint, [source_svc_label1...], [dest_svc_label1...], Layer)` DetectPoint including `DetectPoint.CLIENT` and `DetectPoint.SERVER`,
    extracts `sourceService` labels from the first array argument, extracts `destService` labels from the second array argument, extracts layer from `Layer` argument.
-   
+
 ## More Examples
 
 Please refer to [OAP Self-Observability](../../../oap-server/server-starter/src/main/resources/fetcher-prom-rules/self.yaml)

oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/Analyzer.java

@@ -20,6 +20,7 @@ package org.apache.skywalking.oap.meter.analyzer;
 
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
+import com.google.gson.JsonObject;
 import io.vavr.Tuple;
 import io.vavr.Tuple2;
 import java.util.List;
@@ -87,8 +88,8 @@ public class Analyzer {
             filter = new FilterExpression(filterExpression);
         }
         ExpressionParsingContext ctx = e.parse();
-        Analyzer analyzer = new Analyzer(metricName, filter, e, meterSystem);
-        analyzer.init(ctx);
+        Analyzer analyzer = new Analyzer(metricName, filter, e, meterSystem, ctx);
+        analyzer.init();
         return analyzer;
     }
 
@@ -102,6 +103,8 @@ public class Analyzer {
 
     private final MeterSystem meterSystem;
 
+    private final ExpressionParsingContext ctx;
+
     private MetricType metricType;
 
     private int[] percentiles;
@@ -221,7 +224,7 @@ public class Analyzer {
         private final String literal;
     }
 
-    private void init(final ExpressionParsingContext ctx) {
+    private void init() {
         this.samples = ctx.getSamples();
         if (ctx.isHistogram()) {
             if (ctx.getPercentiles() != null && ctx.getPercentiles().length > 0) {
@@ -282,6 +285,11 @@ public class Analyzer {
             instanceTraffic.setServiceId(entity.serviceId());
             instanceTraffic.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
             instanceTraffic.setLastPingTimestamp(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
+            if (entity.getInstanceProperties() != null && !entity.getInstanceProperties().isEmpty()) {
+                final JsonObject properties = new JsonObject();
+                entity.getInstanceProperties().forEach((k, v) -> properties.addProperty(k, v));
+                instanceTraffic.setProperties(properties);
+            }
             MetricsStreamProcessor.getInstance().in(instanceTraffic);
         }
         if (!com.google.common.base.Strings.isNullOrEmpty(entity.getEndpointName())) {

oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/EntityDescription/InstanceEntityDescription.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.oap.meter.analyzer.dsl.EntityDescription;
 
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import lombok.Getter;
@@ -26,6 +27,7 @@ import lombok.RequiredArgsConstructor;
 import lombok.ToString;
 import org.apache.skywalking.oap.server.core.analysis.Layer;
 import org.apache.skywalking.oap.server.core.analysis.meter.ScopeType;
+import groovy.lang.Closure;
 
 @Getter
 @RequiredArgsConstructor
@@ -35,7 +37,9 @@ public class InstanceEntityDescription implements EntityDescription {
     private final List<String> serviceKeys;
     private final List<String> instanceKeys;
     private final Layer layer;
-    private final String delimiter;
+    private final String serviceDelimiter;
+    private final String instanceDelimiter;
+    private final Closure<Map<String, String>> propertiesExtractor;
 
     @Override
     public List<String> getLabelKeys() {

oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java

@@ -448,7 +448,9 @@ public class SampleFamily {
         return createMeterSamples(new ServiceEntityDescription(labelKeys, layer, delimiter));
     }
 
-    public SampleFamily instance(List<String> serviceKeys, List<String> instanceKeys, Layer layer) {
+    public SampleFamily instance(List<String> serviceKeys, String serviceDelimiter,
+                                 List<String> instanceKeys, String instanceDelimiter,
+                                 Layer layer, Closure<Map<String, String>> propertiesExtractor) {
         Preconditions.checkArgument(serviceKeys.size() > 0);
         Preconditions.checkArgument(instanceKeys.size() > 0);
         ExpressionParsingContext.get().ifPresent(ctx -> {
@@ -459,7 +461,12 @@ public class SampleFamily {
         if (this == EMPTY) {
             return EMPTY;
         }
-        return createMeterSamples(new InstanceEntityDescription(serviceKeys, instanceKeys, layer, Const.POINT));
+        return createMeterSamples(new InstanceEntityDescription(
+            serviceKeys, instanceKeys, layer, serviceDelimiter, instanceDelimiter, propertiesExtractor));
+    }
+
+    public SampleFamily instance(List<String> serviceKeys, List<String> instanceKeys, Layer layer) {
+        return instance(serviceKeys, Const.POINT, instanceKeys, Const.POINT, layer, null);
     }
 
     public SampleFamily endpoint(List<String> serviceKeys, List<String> endpointKeys, Layer layer) {
@@ -620,10 +627,15 @@ public class SampleFamily {
                     );
                 case SERVICE_INSTANCE:
                     InstanceEntityDescription instanceEntityDescription = (InstanceEntityDescription) entityDescription;
+                    Map<String, String> properties = null;
+                    if (instanceEntityDescription.getPropertiesExtractor() != null) {
+                        properties = instanceEntityDescription.getPropertiesExtractor().call(samples.get(0).labels);
+                    }
                     return MeterEntity.newServiceInstance(
-                        InternalOps.dim(samples, instanceEntityDescription.getServiceKeys(), instanceEntityDescription.getDelimiter()),
-                        InternalOps.dim(samples, instanceEntityDescription.getInstanceKeys(), instanceEntityDescription.getDelimiter()),
-                        instanceEntityDescription.getLayer()
+                        InternalOps.dim(samples, instanceEntityDescription.getServiceKeys(), instanceEntityDescription.getServiceDelimiter()),
+                        InternalOps.dim(samples, instanceEntityDescription.getInstanceKeys(), instanceEntityDescription.getInstanceDelimiter()),
+                        instanceEntityDescription.getLayer(),
+                        properties
                     );
                 case ENDPOINT:
                     EndpointEntityDescription endpointEntityDescription = (EndpointEntityDescription) entityDescription;

oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/ScopeTest.java

@@ -413,15 +413,15 @@ public class ScopeTest {
                 new HashMap<MeterEntity, Sample[]>() {
                     {
                         put(
-                            MeterEntity.newServiceInstance("t1", "us", Layer.GENERAL),
+                            MeterEntity.newServiceInstance("t1", "us", Layer.GENERAL, null),
                             new Sample[] {Sample.builder().labels(of()).value(150).name("http_success_request").build()}
                         );
                         put(
-                            MeterEntity.newServiceInstance("t3", "cn", Layer.GENERAL),
+                            MeterEntity.newServiceInstance("t3", "cn", Layer.GENERAL, null),
                             new Sample[] {Sample.builder().labels(of()).value(54).name("http_success_request").build()}
                         );
                         put(
-                            MeterEntity.newServiceInstance("t1", "", Layer.GENERAL),
+                            MeterEntity.newServiceInstance("t1", "", Layer.GENERAL, null),
                             new Sample[] {Sample.builder().labels(of()).value(50).name("http_success_request").build()}
                         );
                     }
@@ -457,7 +457,7 @@ public class ScopeTest {
                 new HashMap<MeterEntity, Sample[]>() {
                     {
                         put(
-                            MeterEntity.newServiceInstance("t1", "us", Layer.GENERAL),
+                            MeterEntity.newServiceInstance("t1", "us", Layer.GENERAL, null),
                             new Sample[] {
                                 Sample.builder()
                                       .labels(of("instance", ""))
@@ -470,7 +470,7 @@ public class ScopeTest {
                             }
                         );
                         put(
-                            MeterEntity.newServiceInstance("t3", "cn", Layer.GENERAL),
+                            MeterEntity.newServiceInstance("t3", "cn", Layer.GENERAL, null),
                             new Sample[] {
                                 Sample.builder()
                                       .labels(of("instance", ""))
@@ -483,7 +483,7 @@ public class ScopeTest {
                             }
                         );
                         put(
-                            MeterEntity.newServiceInstance("t1", "", Layer.GENERAL),
+                            MeterEntity.newServiceInstance("t1", "", Layer.GENERAL, null),
                             new Sample[] {
                                 Sample.builder()
                                       .labels(of("instance", ""))

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/InstanceTraffic.java

@@ -79,7 +79,10 @@ public class InstanceTraffic extends Metrics {
         final InstanceTraffic instanceTraffic = (InstanceTraffic) metrics;
         this.lastPingTimestamp = instanceTraffic.getLastPingTimestamp();
         if (instanceTraffic.getProperties() != null && instanceTraffic.getProperties().size() > 0) {
-            this.properties = instanceTraffic.getProperties();
+            if (this.properties == null) {
+                this.properties = new JsonObject();
+            }
+            instanceTraffic.getProperties().entrySet().forEach(it -> this.properties.add(it.getKey(), it.getValue()));
         }
         /**
          * Keep the time bucket as the same time inserted.

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/MeterEntity.java

@@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.core.analysis.meter;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.ToString;
+import java.util.Map;
 import org.apache.skywalking.oap.server.core.UnexpectedException;
 import org.apache.skywalking.oap.server.core.analysis.IDManager;
 import org.apache.skywalking.oap.server.core.analysis.Layer;
@@ -39,6 +40,7 @@ public class MeterEntity {
     private ScopeType scopeType;
     private String serviceName;
     private String instanceName;
+    private Map<String, String> instanceProperties;
     private String endpointName;
     private String sourceServiceName;
     private String destServiceName;
@@ -103,11 +105,12 @@ public class MeterEntity {
     /**
      * Create a service instance level meter entity.
      */
-    public static MeterEntity newServiceInstance(String serviceName, String serviceInstance, Layer layer) {
+    public static MeterEntity newServiceInstance(String serviceName, String serviceInstance, Layer layer, Map<String, String> properties) {
         final MeterEntity meterEntity = new MeterEntity();
         meterEntity.scopeType = ScopeType.SERVICE_INSTANCE;
         meterEntity.serviceName = NAMING_CONTROL.formatServiceName(serviceName);
         meterEntity.instanceName = NAMING_CONTROL.formatInstanceName(serviceInstance);
+        meterEntity.instanceProperties = properties;
         meterEntity.layer = layer;
         return meterEntity;
     }