Add filter mechanism in MAL core and fix some bugs (#8157)

CHANGES.md

@@ -51,6 +51,9 @@ Release Notes.
 * Upgrade Kubernetes Java client to 14.0.0, supports GCP token refreshing and fixes some bugs.
 * Change `SO11Y` metric `envoy_als_in_count` to calculate the ALS message count.
 * Support Istio `1.10.3`, `1.11.4`, `1.12.0` release.(Tested through e2e)
+* Add filter mechanism in MAL core to filter metrics.
+* Fix concurrency bug in MAL `increase`-related calculation.
+* Fix a null pointer bug when building `SampleFamily`.
 
 #### UI
 

docs/en/setup/backend/backend-meter.md

@@ -57,6 +57,8 @@ If you're using Spring Sleuth, see [Spring Sleuth Setup](spring-sleuth-setup.md)
 ### Meters configuration
 
 ```yaml
+# filter the metrics, only those metrics that satisfy this condition will be passed into the `metricsRules` below.
+filter: <closure> # example: '{ tags -> tags.job_name == "vm-monitoring" }'
 # expSuffix is appended to all expression in this file.
 expSuffix: <string>
 # insert metricPrefix into metric name:  <metricPrefix>_<raw_metric_name>

docs/en/setup/backend/backend-telemetry.md

@@ -128,7 +128,7 @@ Set this up following these steps:
   ```
 2. Set up OpenTelemetry Collector and config a scrape job:
 ``` yaml
-- job_name: 'skywalking'
+- job_name: 'skywalking-so11y' # make sure to use this in the so11y.yaml to filter only so11y metrics
   metrics_path: '/metrics'
   kubernetes_sd_configs:
   - role: pod

docs/en/setup/backend/otel-collector-oap.yaml

@@ -30,7 +30,7 @@ data:
             scrape_interval: 10s
             evaluation_interval: 30s
           scrape_configs:
-          - job_name: 'skywalking'
+          - job_name: 'skywalking-so11y'
             metrics_path: '/metrics'
             kubernetes_sd_configs:
             - role: pod
@@ -177,4 +177,4 @@ roleRef:
 subjects:
 - kind: ServiceAccount
   name: default
-  namespace: istio-system
\ No newline at end of file
+  namespace: istio-system

docs/en/setup/backend/prometheus-metrics.md

@@ -37,6 +37,8 @@ staticConfig:
   # Labels assigned to all metrics fetched from the targets.
   labels:
     [ <labelname>: <labelvalue> ... ]
+# filter the metrics, only those metrics that satisfy this condition will be passed into the `metricsRules` below.
+filter: <closure> # example: '{ tags -> tags.job_name == "vm-monitoring" }'
 # expSuffix is appended to all expression in this file.
 expSuffix: <string>
 # insert metricPrefix into metric name:  <metricPrefix>_<raw_metric_name>

oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/meter/config/MeterConfig.java

@@ -29,6 +29,7 @@ import java.util.List;
 public class MeterConfig implements MetricRuleConfig {
     private String metricPrefix;
     private String expSuffix;
+    private String filter;
     private List<Rule> metricsRules;
 
     @Data

oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/meter/process/MeterProcessor.java

@@ -133,7 +133,7 @@ public class MeterProcessor {
         }
 
         try {
-            converts.stream().forEach(convert -> convert.toMeter(meters.entrySet().stream().collect(toImmutableMap(
+            converts.forEach(convert -> convert.toMeter(meters.entrySet().stream().collect(toImmutableMap(
                 Map.Entry::getKey,
                 v -> SampleFamilyBuilder.newBuilder(
                     v.getValue().stream().map(s -> s.build(service, serviceInstance, timestamp)).toArray(Sample[]::new)

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

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.meter.analyzer;
 
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
 import io.vavr.Tuple;
 import io.vavr.Tuple2;
@@ -36,6 +37,7 @@ import org.apache.skywalking.oap.meter.analyzer.dsl.DSL;
 import org.apache.skywalking.oap.meter.analyzer.dsl.DownsamplingType;
 import org.apache.skywalking.oap.meter.analyzer.dsl.Expression;
 import org.apache.skywalking.oap.meter.analyzer.dsl.ExpressionParsingContext;
+import org.apache.skywalking.oap.meter.analyzer.dsl.FilterExpression;
 import org.apache.skywalking.oap.meter.analyzer.dsl.Result;
 import org.apache.skywalking.oap.meter.analyzer.dsl.Sample;
 import org.apache.skywalking.oap.meter.analyzer.dsl.SampleFamily;
@@ -56,6 +58,7 @@ import org.apache.skywalking.oap.server.core.analysis.meter.function.PercentileA
 import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
 import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
 
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.mapping;
@@ -74,11 +77,17 @@ public class Analyzer {
 
     public static final Tuple2<String, SampleFamily> NIL = Tuple.of("", null);
 
-    public static Analyzer build(final String metricName, final String expression,
+    public static Analyzer build(final String metricName,
+                                 final String filterExpression,
+                                 final String expression,
                                  final MeterSystem meterSystem) {
         Expression e = DSL.parse(expression);
+        FilterExpression filter = null;
+        if (!Strings.isNullOrEmpty(filterExpression)) {
+            filter = new FilterExpression(filterExpression);
+        }
         ExpressionParsingContext ctx = e.parse();
-        Analyzer analyzer = new Analyzer(metricName, e, meterSystem);
+        Analyzer analyzer = new Analyzer(metricName, filter, e, meterSystem);
         analyzer.init(ctx);
         return analyzer;
     }
@@ -89,6 +98,8 @@ public class Analyzer {
 
     private final String metricName;
 
+    private final FilterExpression filterExpression;
+
     private final Expression expression;
 
     private final MeterSystem meterSystem;
@@ -103,16 +114,19 @@ public class Analyzer {
      * @param sampleFamilies input samples.
      */
     public void analyse(final ImmutableMap<String, SampleFamily> sampleFamilies) {
-        ImmutableMap<String, SampleFamily> input = samples.stream()
-                                                          .map(s -> Tuple.of(s, sampleFamilies.get(s)))
-                                                          .filter(t -> t._2 != null)
-                                                          .collect(ImmutableMap.toImmutableMap(t -> t._1, t -> t._2));
+        Map<String, SampleFamily> input = samples.stream()
+                                                 .map(s -> Tuple.of(s, sampleFamilies.get(s)))
+                                                 .filter(t -> t._2 != null)
+                                                 .collect(toImmutableMap(t -> t._1, t -> t._2));
         if (input.size() < 1) {
             if (log.isDebugEnabled()) {
                 log.debug("{} is ignored due to the lack of {}", expression, samples);
             }
             return;
         }
+        if (filterExpression != null) {
+            input = filterExpression.filter(input);
+        }
         Result r = expression.run(input);
         if (!r.isSuccess()) {
             return;

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

@@ -51,6 +51,7 @@ public class MetricConvert {
         this.analyzers = rule.getMetricsRules().stream().map(
             r -> Analyzer.build(
                 formatMetricName(rule, r.getName()),
+                rule.getFilter(),
                 Strings.isNullOrEmpty(rule.getExpSuffix()) ?
                     r.getExp() : String.format("(%s).%s", r.getExp(), rule.getExpSuffix()),
                 service

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

@@ -40,6 +40,8 @@ public interface MetricRuleConfig {
      */
     List<? extends RuleConfig> getMetricsRules();
 
+    String getFilter();
+
     interface RuleConfig {
         /**
          * Get definition metrics name

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

@@ -23,6 +23,7 @@ import groovy.lang.ExpandoMetaClass;
 import groovy.lang.GroovyObjectSupport;
 import groovy.util.DelegatingScript;
 import java.time.Instant;
+import java.util.Map;
 import lombok.RequiredArgsConstructor;
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
@@ -38,7 +39,7 @@ public class Expression {
 
     private final DelegatingScript expression;
 
-    private final ThreadLocal<ImmutableMap<String, SampleFamily>> propertyRepository = new ThreadLocal<>();
+    private final ThreadLocal<Map<String, SampleFamily>> propertyRepository = new ThreadLocal<>();
 
     public Expression(final String literal, final DelegatingScript expression) {
         this.literal = literal;
@@ -71,7 +72,7 @@ public class Expression {
      * @param sampleFamilies a data map includes all of candidates to be analysis.
      * @return The result of execution.
      */
-    public Result run(final ImmutableMap<String, SampleFamily> sampleFamilies) {
+    public Result run(final Map<String, SampleFamily> sampleFamilies) {
         propertyRepository.set(sampleFamilies);
         try {
             SampleFamily sf = (SampleFamily) expression.run();
@@ -114,7 +115,7 @@ public class Expression {
         public static final DownsamplingType LATEST = DownsamplingType.LATEST;
 
         private final String literal;
-        private final ThreadLocal<ImmutableMap<String, SampleFamily>> propertyRepository;
+        private final ThreadLocal<Map<String, SampleFamily>> propertyRepository;
 
         public SampleFamily propertyMissing(String metricName) {
             ExpressionParsingContext.get().ifPresent(ctx -> {
@@ -122,7 +123,7 @@ public class Expression {
                     ctx.samples.add(metricName);
                 }
             });
-            ImmutableMap<String, SampleFamily> sampleFamilies = propertyRepository.get();
+            Map<String, SampleFamily> sampleFamilies = propertyRepository.get();
             if (sampleFamilies == null) {
                 return SampleFamily.EMPTY;
             }

oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/FilterExpression.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.meter.analyzer.dsl;
+
+import groovy.lang.Closure;
+import groovy.lang.GroovyShell;
+import java.util.Map;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+
+import static java.util.stream.Collectors.toMap;
+
+@Slf4j
+@ToString(of = {"literal"})
+public class FilterExpression {
+    private final String literal;
+    private final Closure<Boolean> filterClosure;
+
+    @SuppressWarnings("unchecked")
+    public FilterExpression(final String literal) {
+        this.literal = literal;
+
+        GroovyShell sh = new GroovyShell();
+        filterClosure = (Closure<Boolean>) sh.evaluate(literal);
+    }
+
+    public Map<String, SampleFamily> filter(final Map<String, SampleFamily> sampleFamilies) {
+        try {
+            return sampleFamilies.entrySet().stream().collect(toMap(
+                Map.Entry::getKey,
+                it -> it.getValue().filter(filterClosure)
+            ));
+        } catch (Throwable t) {
+            log.error("failed to run \"{}\"", literal, t);
+        }
+        return sampleFamilies;
+    }
+}

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

@@ -18,22 +18,13 @@
 
 package org.apache.skywalking.oap.meter.analyzer.dsl;
 
-import com.google.common.base.CharMatcher;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Maps;
-import com.google.common.util.concurrent.AtomicDouble;
-import groovy.lang.Closure;
-import io.vavr.Function2;
-import io.vavr.Function3;
-import lombok.AccessLevel;
-import lombok.Builder;
-import lombok.EqualsAndHashCode;
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-import lombok.Setter;
-import lombok.ToString;
+import static java.util.function.UnaryOperator.identity;
+import static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.mapping;
+import static java.util.stream.Collectors.toList;
+
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
+
 import org.apache.skywalking.oap.meter.analyzer.dsl.EntityDescription.EndpointEntityDescription;
 import org.apache.skywalking.oap.meter.analyzer.dsl.EntityDescription.EntityDescription;
 import org.apache.skywalking.oap.meter.analyzer.dsl.EntityDescription.InstanceEntityDescription;
@@ -43,6 +34,7 @@ import org.apache.skywalking.oap.meter.analyzer.dsl.tagOpt.K8sRetagType;
 import org.apache.skywalking.oap.server.core.UnexpectedException;
 import org.apache.skywalking.oap.server.core.analysis.meter.MeterEntity;
 import org.apache.skywalking.oap.server.core.analysis.meter.ScopeType;
+import org.apache.skywalking.oap.server.core.source.DetectPoint;
 
 import java.util.Arrays;
 import java.util.Comparator;
@@ -57,13 +49,25 @@ import java.util.function.DoubleBinaryOperator;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
-import org.apache.skywalking.oap.server.core.source.DetectPoint;
 
-import static com.google.common.collect.ImmutableMap.toImmutableMap;
-import static java.util.function.UnaryOperator.identity;
-import static java.util.stream.Collectors.groupingBy;
-import static java.util.stream.Collectors.mapping;
-import static java.util.stream.Collectors.toList;
+import com.google.common.base.CharMatcher;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.AtomicDouble;
+
+import groovy.lang.Closure;
+import io.vavr.Function2;
+import io.vavr.Function3;
+import lombok.AccessLevel;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * SampleFamily represents a collection of {@link Sample}.
@@ -71,13 +75,17 @@ import static java.util.stream.Collectors.toList;
 @RequiredArgsConstructor(access = AccessLevel.PRIVATE)
 @EqualsAndHashCode
 @ToString
+@Slf4j
 public class SampleFamily {
     public static final SampleFamily EMPTY = new SampleFamily(new Sample[0], RunningContext.EMPTY);
 
     static SampleFamily build(RunningContext ctx, Sample... samples) {
         Preconditions.checkNotNull(samples);
-        samples = Arrays.stream(samples).filter(sample -> !Double.isNaN(sample.getValue())).toArray(Sample[]::new);
         Preconditions.checkArgument(samples.length > 0);
+        samples = Arrays.stream(samples).filter(sample -> !Double.isNaN(sample.getValue())).toArray(Sample[]::new);
+        if (samples.length == 0) {
+            return EMPTY;
+        }
         return new SampleFamily(samples, Optional.ofNullable(ctx).orElseGet(RunningContext::instance));
     }
 
@@ -332,6 +340,19 @@ public class SampleFamily {
         );
     }
 
+    public SampleFamily filter(Closure<Boolean> filter) {
+        if (this == EMPTY) {
+            return EMPTY;
+        }
+        final Sample[] filtered = Arrays.stream(samples)
+                                        .filter(it -> filter.call(it.labels))
+                                        .toArray(Sample[]::new);
+        if (filtered.length == 0) {
+            return EMPTY;
+        }
+        return SampleFamily.build(context, filtered);
+    }
+
     /* k8s retags*/
     public SampleFamily retagByK8sMeta(String newLabelName,
                                        K8sRetagType type,
@@ -468,7 +489,8 @@ public class SampleFamily {
               ))
               .forEach((labels, samples) -> {
                   MeterEntity meterEntity = InternalOps.buildMeterEntity(samples, entityDescription);
-                  meterSamples.put(meterEntity, InternalOps.left(samples, entityDescription.getLabelKeys()));
+                  meterSamples.put(
+                      meterEntity, InternalOps.left(samples, entityDescription.getLabelKeys()));
               });
 
         this.context.setMeterSamples(meterSamples);
@@ -604,7 +626,7 @@ public class SampleFamily {
                     );
                 default:
                     throw new UnexpectedException(
-                        "Unexpected scope type of entityDescription " + entityDescription.toString());
+                        "Unexpected scope type of entityDescription " + entityDescription);
             }
         }
 

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

@@ -19,12 +19,12 @@
 package org.apache.skywalking.oap.meter.analyzer.dsl.counter;
 
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Maps;
 import io.vavr.Tuple;
 import io.vavr.Tuple2;
 import java.util.Map;
 import java.util.PriorityQueue;
 import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
 import lombok.AccessLevel;
 import lombok.EqualsAndHashCode;
 import lombok.RequiredArgsConstructor;
@@ -42,16 +42,12 @@ public class CounterWindow {
 
     public static final CounterWindow INSTANCE = new CounterWindow();
 
-    private final Map<ID, Tuple2<Long, Double>> lastElementMap = Maps.newHashMap();
-    private final Map<ID, Queue<Tuple2<Long, Double>>> windows = Maps.newHashMap();
+    private final Map<ID, Tuple2<Long, Double>> lastElementMap = new ConcurrentHashMap<>();
+    private final Map<ID, Queue<Tuple2<Long, Double>>> windows = new ConcurrentHashMap<>();
 
     public Tuple2<Long, Double> increase(String name, ImmutableMap<String, String> labels, Double value, long windowSize, long now) {
         ID id = new ID(name, labels);
-        if (!windows.containsKey(id)) {
-            windows.put(id, new PriorityQueue<>());
-        }
-
-        Queue<Tuple2<Long, Double>> window = windows.get(id);
+        Queue<Tuple2<Long, Double>> window = windows.computeIfAbsent(id, unused -> new PriorityQueue<>());
         window.offer(Tuple.of(now, value));
         long waterLevel = now - windowSize;
         Tuple2<Long, Double> peek = window.peek();
@@ -77,8 +73,7 @@ public class CounterWindow {
         ID id = new ID(name, labels);
 
         Tuple2<Long, Double> element = Tuple.of(now, value);
-        Tuple2<Long, Double> result = lastElementMap.get(id);
-        lastElementMap.put(id, element);
+        Tuple2<Long, Double> result = lastElementMap.put(id, element);
         if (result == null) {
             return element;
         }

oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/prometheus/rule/Rule.java

@@ -37,5 +37,6 @@ public class Rule implements MetricRuleConfig {
     private StaticConfig staticConfig;
     private String metricPrefix;
     private String expSuffix;
+    private String filter;
     private List<MetricsRule> metricsRules;
 }

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

@@ -87,6 +87,7 @@ public class AnalyzerTest {
     public void testSingle() {
         analyzer = Analyzer.build(
             "sum_service_instance",
+            null,
             "http_success_request.sum(['region', 'idc']).instance(['idc'] , ['region'])",
             meterSystem
         );
@@ -130,6 +131,7 @@ public class AnalyzerTest {
     public void testLabeled() {
         analyzer = Analyzer.build(
             "sum_service_instance_labels",
+            null,
             "http_success_request.sum(['region', 'idc' , 'instance']).instance(['idc'] , ['region'])",
             meterSystem
         );
@@ -178,6 +180,7 @@ public class AnalyzerTest {
     public void testHistogramPercentile() {
         analyzer = Analyzer.build(
             "instance_cpu_percentage",
+            null,
             "instance_cpu_percentage.sum(['le' , 'service' , 'instance']).histogram().histogram_percentile([75,99]).service(['service'])",
             meterSystem
         );

oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/FilterTest.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.meter.analyzer.dsl;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import static com.google.common.collect.ImmutableMap.of;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import com.google.common.collect.ImmutableMap;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@RunWith(Parameterized.class)
+public class FilterTest {
+    @Parameterized.Parameter
+    public String name;
+
+    @Parameterized.Parameter(1)
+    public ImmutableMap<String, SampleFamily> input;
+
+    @Parameterized.Parameter(2)
+    public String expression;
+
+    @Parameterized.Parameter(3)
+    public Result want;
+
+    @Parameterized.Parameters(name = "{index}: {0}")
+    public static Collection<Object[]> data() {
+        final SampleFamily sf =
+            SampleFamilyBuilder.newBuilder(
+                                   Sample.builder()
+                                         .value(1600592418480.0)
+                                         .labels(ImmutableMap.of("str", "val1"))
+                                         .name("instance_cpu_percentage")
+                                         .build(),
+                                   Sample.builder()
+                                         .value(1600592418480.0)
+                                         .labels(ImmutableMap.of("str", "val2"))
+                                         .name("instance_cpu_percentage")
+                                         .build())
+                               .build();
+        return Arrays.asList(new Object[][]{
+            {
+                "filter-string",
+                of("instance_cpu_percentage", sf),
+                "instance_cpu_percentage.filter({ tags -> tags.str == 'val1' })",
+                Result.success(SampleFamily.build(sf.context, sf.samples[0]))
+            },
+            {
+                "filter-none",
+                of("instance_cpu_percentage", sf),
+                "instance_cpu_percentage.filter({ tags -> tags.str == 'val2' })",
+                Result.success(SampleFamily.build(sf.context, sf.samples[1]))
+            },
+            {
+                "filter-not-equal",
+                of("instance_cpu_percentage", sf),
+                "instance_cpu_percentage.filter({ tags -> tags.str != 'val1' })",
+                Result.success(SampleFamily.build(sf.context, sf.samples[1]))
+            },
+            {
+                "filter-in",
+                of("instance_cpu_percentage", sf),
+                "instance_cpu_percentage.filter({ tags -> tags.str in [ 'val2' ] })",
+                Result.success(SampleFamily.build(sf.context, sf.samples[1]))
+            },
+            {
+                "filter-in",
+                of("instance_cpu_percentage", sf),
+                "instance_cpu_percentage.filter({ tags -> tags.str in [ 'val1', 'val2' ] })",
+                Result.success(sf)
+            },
+        });
+    }
+
+    @Test
+    public void test() {
+        Expression e = DSL.parse(expression);
+        Result r = e.run(input);
+        assertThat(r, is(want));
+    }
+}

oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/oc/OCMetricHandler.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.receiver.otel.oc;
 
+import com.google.common.base.Strings;
 import com.google.protobuf.Timestamp;
 import io.grpc.stub.StreamObserver;
 import io.opencensus.proto.agent.common.v1.Node;
@@ -75,6 +76,10 @@ public class OCMetricHandler extends MetricsServiceGrpc.MetricsServiceImplBase i
                             nodeLabels.put("node_identifier_pid", String.valueOf(node.getIdentifier().getPid()));
                         }
                     }
+                    final String name = node.getServiceInfo().getName();
+                    if (!Strings.isNullOrEmpty(name)) {
+                        nodeLabels.put("job_name", name);
+                    }
                 }
                 metrics.forEach(m -> m.toMeter(request.getMetricsList().stream()
                     .flatMap(metric -> metric.getTimeseriesList().stream().map(timeSeries ->

oap-server/server-receiver-plugin/skywalking-zabbix-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zabbix/provider/config/ZabbixConfig.java

@@ -28,6 +28,7 @@ public class ZabbixConfig implements MetricRuleConfig {
 
     private String metricPrefix;
     private String expSuffix;
+    private String filter;
     private Entities entities;
     private List<String> requiredZabbixItemKeys;
     private List<Metric> metrics;

oap-server/server-starter/src/main/resources/otel-oc-rules/istio-controlplane.yaml

@@ -28,6 +28,7 @@
 #    "-P6H3M"    -- parses as "-6 hours and -3 minutes"
 #    "-P-6H+3M"  -- parses as "+6 hours and -3 minutes"
 # </pre>
+filter: "{ tags -> tags.job_name == 'kubernetes-pods' }" # The OpenTelemetry job name
 expSuffix: tag({tags -> tags.cluster = 'istio-ctrl::' + tags.cluster}).service(['cluster', 'app'])
 metricPrefix: meter_istio
 metricsRules:

oap-server/server-starter/src/main/resources/otel-oc-rules/k8s-cluster.yaml

@@ -28,11 +28,10 @@
 #    "-P6H3M"    -- parses as "-6 hours and -3 minutes"
 #    "-P-6H+3M"  -- parses as "+6 hours and -3 minutes"
 # </pre>
+filter: "{ tags -> tags.job_name in [ 'kubernetes-cadvisor', 'kube-state-metrics' ] }" # The OpenTelemetry job name
 expSuffix: tag({tags -> tags.cluster = 'k8s-cluster::' + tags.cluster}).service(['cluster'])
 metricPrefix: k8s_cluster
 metricsRules:
-
-
   - name: cpu_cores
     exp: (kube_node_status_capacity * 1000).tagEqual('resource' , 'cpu').sum(['cluster'])
   - name: cpu_cores_allocatable

oap-server/server-starter/src/main/resources/otel-oc-rules/k8s-node.yaml

@@ -28,7 +28,7 @@
 #    "-P6H3M"    -- parses as "-6 hours and -3 minutes"
 #    "-P-6H+3M"  -- parses as "+6 hours and -3 minutes"
 # </pre>
-
+filter: "{ tags -> tags.job_name in [ 'kubernetes-cadvisor', 'kube-state-metrics' ] }" # The OpenTelemetry job name
 expSuffix: tag({tags -> tags.cluster = 'k8s-cluster::' + tags.cluster}).instance(['cluster'] , ['node'])
 metricPrefix: k8s_node
 metricsRules:

oap-server/server-starter/src/main/resources/otel-oc-rules/k8s-service.yaml

@@ -28,10 +28,10 @@
 #    "-P6H3M"    -- parses as "-6 hours and -3 minutes"
 #    "-P-6H+3M"  -- parses as "+6 hours and -3 minutes"
 # </pre>
+filter: "{ tags -> tags.job_name in [ 'kubernetes-cadvisor', 'kube-state-metrics' ] }" # The OpenTelemetry job name
 expSuffix: tag({tags -> tags.cluster = 'k8s-cluster::' + tags.cluster}).endpoint(['cluster'] , ['service'])
 metricPrefix: k8s_service
 metricsRules:
-
   - name: pod_total
     exp: kube_pod_info.retagByK8sMeta('service' , K8sRetagType.Pod2Service , 'pod' , 'namespace').tagNotEqual('service' , '').sum(['cluster' , 'service'])
 

oap-server/server-starter/src/main/resources/otel-oc-rules/oap.yaml

@@ -28,6 +28,7 @@
 #    "-P6H3M"    -- parses as "-6 hours and -3 minutes"
 #    "-P-6H+3M"  -- parses as "+6 hours and -3 minutes"
 # </pre>
+filter: "{ tags -> tags.job_name == 'skywalking-so11y' }" # The OpenTelemetry job name
 expSuffix: tag({tags -> tags.service = 'oap::' + tags.service}).instance(['service'], ['host_name'])
 metricPrefix: meter_oap
 metricsRules:

oap-server/server-starter/src/main/resources/otel-oc-rules/vm.yaml

@@ -28,6 +28,7 @@
 #    "-P6H3M"    -- parses as "-6 hours and -3 minutes"
 #    "-P-6H+3M"  -- parses as "+6 hours and -3 minutes"
 # </pre>
+filter: "{ tags -> tags.job_name == 'vm-monitoring' }" # The OpenTelemetry job name
 expSuffix: tag({tags -> tags.node_identifier_host_name = 'vm::' + tags.node_identifier_host_name}).service(['node_identifier_host_name'])
 metricPrefix: meter_vm
 metricsRules:

test/e2e-v2/cases/vm/prometheus-node-exporter/otel-collector-config.yaml

@@ -17,7 +17,7 @@ receivers:
   prometheus:
     config:
       scrape_configs:
-        - job_name: 'otel-collector'
+        - job_name: 'vm-monitoring' # make sure to use this in the vm.yaml to filter only VM metrics
           scrape_interval: 10s
           static_configs:
             - targets: [ 'vm-service:9100' ]