Enhance Envoy metrics service analyzer by MAL (#6091)

.github/workflows/e2e.istio.yaml

@@ -129,3 +129,97 @@ jobs:
       - name: Clean up
         if: ${{ always() }}
         run: minikube delete
+
+  metrics-service:
+    runs-on: ubuntu-16.04
+    timeout-minutes: 60
+    name: MetricsService
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: true
+
+      - uses: actions/cache@v2
+        with:
+          path: ~/.m2/repository
+          key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
+          restore-keys: ${{ runner.os }}-maven-
+
+      - name: Build Docker Image
+        run: make docker
+
+      - name: Prepare envrionment
+        run: bash ${SCRIPTS_DIR}/pre.sh
+
+      - name: Install Minikube
+        run: bash ${SCRIPTS_DIR}/minikube.sh start
+
+      - name: Install Istio
+        run: |
+          bash ${SCRIPTS_DIR}/istio.sh \
+            --set profile=demo \
+            --set meshConfig.defaultConfig.envoyMetricsService.address=skywalking-oap.istio-system:11800 \
+            --set values.telemetry.v2.enabled=false # disable the metadata-exchange extension intentionally to make sure metrics service doesn't rely on it
+
+      - name: Install SkyWalking
+        run: |
+          git clone https://github.com/apache/skywalking-kubernetes.git
+          cd skywalking-kubernetes
+          git reset --hard dd749f25913830c47a97430618cefc4167612e75
+          cd chart
+          helm dep up skywalking
+          helm -n istio-system install skywalking skywalking \
+               --set fullnameOverride=skywalking \
+               --set elasticsearch.replicas=1 \
+               --set elasticsearch.minimumMasterNodes=1 \
+               --set elasticsearch.imageTag=7.5.1 \
+               --set oap.replicas=1 \
+               --set ui.image.repository=skywalking/ui \
+               --set ui.image.tag=$TAG \
+               --set oap.image.tag=$TAG \
+               --set oap.image.repository=skywalking/oap \
+               --set oap.storageType=elasticsearch7
+          kubectl -n istio-system get pods
+
+          sleep 3
+          kubectl -n istio-system wait --for=condition=available deployments/skywalking-oap --timeout=1200s
+          kubectl get pods -A -o wide --show-labels
+          kubectl get services -A -o wide
+
+      - name: Deploy demo services
+        run: bash ${SCRIPTS_DIR}/demo.sh
+
+      - name: Cluster Info
+        if: ${{ failure() }}
+        run: |
+          df -h
+          minikube logs
+          minikube status
+
+      - name: Set up Minikube tunnel
+        run: |
+          mkdir /tmp/minikube-tunnel
+          minikube tunnel > /tmp/minikube-tunnel/a.log &
+          export POD_NAME=$(kubectl get pods -n istio-system -l "app=skywalking,release=skywalking,component=ui" -o jsonpath="{.items[0].metadata.name}")
+          echo $POD_NAME
+          kubectl -n istio-system port-forward $POD_NAME 8080:8080 > /tmp/minikube-tunnel/b.log &
+
+      - name: Run E2E test
+        run: |
+          export GATEWAY_HOST=$(minikube ip)
+          export GATEWAY_PORT=$(kubectl -n istio-system get service istio-ingressgateway -o jsonpath='{.spec.ports[?(@.name=="http2")].nodePort}')
+          export WEBAPP_HOST=127.0.0.1
+          export WEBAPP_PORT=8080
+
+          ./mvnw --batch-mode -f test/e2e/pom.xml -am -DfailIfNoTests=false verify -Dit.test=org.apache.skywalking.e2e.mesh.MetricsServiceE2E
+
+      - name: Logs
+        if: ${{ failure() }}
+        run: |
+          kubectl -n istio-system logs --tail=10000 -l "app=skywalking,release=skywalking,component=ui"
+          kubectl -n istio-system logs --tail=10000 -l "app=skywalking,release=skywalking,component=oap"
+          cat /tmp/minikube-tunnel/*
+
+      - name: Clean up
+        if: ${{ always() }}
+        run: minikube delete

CHANGES.md

@@ -42,6 +42,7 @@ Release Notes.
 * Improve query performance in storage-influxdb-plugin.
 * Fix the uuid field in GRPCConfigWatcherRegister is not updated.
 * Support Envoy {AccessLog,Metrics}Service API V3.
+* Adopt the [MAL](docs/en/concepts-and-designs/mal.md) in Envoy metrics service analyzer.
 
 #### UI
 * Fix un-removed tags in trace query.

apm-dist-es7/src/main/assembly/binary-es7.xml

@@ -53,6 +53,7 @@
                 <include>endpoint-name-grouping.yml</include>
                 <include>oal/*.oal</include>
                 <include>fetcher-prom-rules/*.yaml</include>
+                <include>envoy-metrics-rules/*.yaml</include>
                 <include>meter-analyzer-config/*.yaml</include>
                 <include>otel-oc-rules/*</include>
                 <include>ui-initialized-templates/*</include>

apm-dist/src/main/assembly/binary.xml

@@ -53,6 +53,7 @@
                 <include>endpoint-name-grouping.yml</include>
                 <include>oal/*.oal</include>
                 <include>fetcher-prom-rules/*.yaml</include>
+                <include>envoy-metrics-rules/*.yaml</include>
                 <include>meter-analyzer-config/*.yaml</include>
                 <include>otel-oc-rules/*</include>
                 <include>ui-initialized-templates/*</include>

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

@@ -59,7 +59,7 @@ Between two scalars: they evaluate to another scalar that is the result of the o
 1 + 2
 ```
 
-Between a sample family and a scalar, the operator is applied to the value of every sample in the smaple family. For example:
+Between a sample family and a scalar, the operator is applied to the value of every sample in the sample family. For example:
 
 ```
 instance_trace_count + 2
@@ -110,9 +110,9 @@ Sample family supports the following aggregation operations that can be used to
 resulting in a new sample family of fewer samples(even single one) with aggregated values:
 
  - sum (calculate sum over dimensions)
- - min (select minimum over dimensions) (TODO)
- - max (select maximum over dimensions) (TODO)
- - avg (calculate the average over dimensions) (TODO)
+ - min (select minimum over dimensions)
+ - max (select maximum over dimensions)
+ - avg (calculate the average over dimensions)
  
 These operations can be used to aggregate over all label dimensions or preserve distinct dimensions by inputting `by` parameter. 
 

docs/en/setup/envoy/als_setting.md

@@ -23,7 +23,7 @@ You need three steps to open ALS.
 `k8s-mesh` uses the metadata from Kubernetes cluster, hence in this analyzer OAP needs access roles to `Pod`, `Service`, and `Endpoints`;
 `mx-mesh` uses the Envoy metadata exchange mechanism to get the service name, etc.,
 this analyzer requires Istio to enable the metadata exchange filter(you can enable it by
-`--set telemetry.v2.enabled=true`, or if you're using Istio 1.7+ and installing it with profile `demo`/`preview`,
+`--set values.telemetry.v2.enabled=true`, or if you're using Istio 1.7+ and installing it with profile `demo`/`preview`,
 it should be enabled then).
 Setting system env variable **SW_ENVOY_METRIC_ALS_HTTP_ANALYSIS** to activate the analyzer,
 such as `SW_ENVOY_METRIC_ALS_HTTP_ANALYSIS=k8s-mesh`.

docs/en/setup/envoy/config.yaml

@@ -39,6 +39,9 @@ node:
   metadata:
     skywalking: iscool
     envoy: isawesome
+    LABELS:
+      app: test-app
+    NAME: service-instance-name
 
 static_resources:
   listeners:

docs/en/setup/envoy/examples/metrics/docker-compose.yaml

@@ -17,7 +17,7 @@
 version: "3"
 services:
   envoy:
-    image: envoyproxy/envoy-alpine:latest
+    image: envoyproxy/envoy-alpine:v1.16.2
     command: /usr/local/bin/envoy -c /etc/envoy.yaml --service-cluster envoy-proxy
     ports:
     - 10000:10000

docs/en/setup/envoy/examples/metrics/envoy.yaml

@@ -39,6 +39,9 @@ node:
   metadata:
     skywalking: iscool
     envoy: isawesome
+    LABELS:
+      app: test-app
+    NAME: service-instance-name
 
 static_resources:
   listeners:
@@ -82,7 +85,7 @@ static_resources:
               - endpoint:
                   address:
                     socket_address:
-                      address: skywalking
+                      address: host.docker.internal
                       port_value: 11800
 
     - name: service_google

docs/en/setup/envoy/metrics_service_setting.md

@@ -39,6 +39,17 @@ A more complete static configuration, can be observed [here](config.yaml).
 
 Note that Envoy can also be configured dynamically through [xDS Protocol](https://github.com/envoyproxy/data-plane-api/blob/master/XDS_PROTOCOL.md).
 
+**Attention**: Only use this when Envoy is under Istio's control, because SkyWalking needs to parse the service name and service instance name from the metadata that is injected by Istio. However, if you want to use this without Istio, you need to inject the metadata yourself like this:
+
+```yaml
+node:
+  # ... other configs
+  metadata:
+    LABELS:
+      app: test-app
+    NAME: service-instance-name
+```
+
 # Metrics data
 
 Some of the Envoy statistics are listed in this [list](https://www.envoyproxy.io/docs/envoy/latest/configuration/statistics). A sample data that contains identifier can be found [here](identify.json), while the metrics only can be observed [here](metrics.json).

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

@@ -28,6 +28,7 @@ import groovy.lang.Closure;
 import io.vavr.Function2;
 import io.vavr.Tuple;
 import io.vavr.Tuple2;
+import java.util.function.DoubleBinaryOperator;
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
@@ -51,6 +52,7 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.mapping;
 import static java.util.stream.Collectors.toList;
@@ -161,22 +163,55 @@ public class SampleFamily {
 
     /* Aggregation operators */
     public SampleFamily sum(List<String> by) {
+        return aggregate(by, Double::sum);
+    }
+
+    public SampleFamily max(List<String> by) {
+        return aggregate(by, Double::max);
+    }
+
+    public SampleFamily min(List<String> by) {
+        return aggregate(by, Double::min);
+    }
+
+    public SampleFamily avg(List<String> by) {
+        ExpressionParsingContext.get().ifPresent(ctx -> ctx.aggregationLabels.addAll(by));
+        if (this == EMPTY) {
+            return EMPTY;
+        }
+        if (by == null) {
+            double result = Arrays.stream(samples).mapToDouble(s -> s.value).average().orElse(0.0D);
+            return SampleFamily.build(this.context, newSample(ImmutableMap.of(), samples[0].timestamp, result));
+        }
+        return SampleFamily.build(
+            this.context,
+            Arrays.stream(samples)
+                  .map(sample -> Tuple.of(by.stream()
+                                            .collect(toImmutableMap(labelKey -> labelKey, labelKey -> sample.labels.getOrDefault(labelKey, ""))), sample))
+                  .collect(groupingBy(Tuple2::_1, mapping(Tuple2::_2, toList())))
+                  .entrySet().stream()
+                  .map(entry -> newSample(entry.getKey(), entry.getValue().get(0).timestamp, entry.getValue().stream()
+                                                                                                  .mapToDouble(s -> s.value).average().orElse(0.0D)))
+                  .toArray(Sample[]::new)
+        );
+    }
+
+    protected SampleFamily aggregate(List<String> by, DoubleBinaryOperator aggregator) {
         ExpressionParsingContext.get().ifPresent(ctx -> ctx.aggregationLabels.addAll(by));
         if (this == EMPTY) {
             return EMPTY;
         }
         if (by == null) {
-            double result = Arrays.stream(samples).mapToDouble(s -> s.value).reduce(Double::sum).orElse(0.0D);
+            double result = Arrays.stream(samples).mapToDouble(s -> s.value).reduce(aggregator).orElse(0.0D);
             return SampleFamily.build(this.context, newSample(ImmutableMap.of(), samples[0].timestamp, result));
         }
         return SampleFamily.build(this.context, Arrays.stream(samples)
             .map(sample -> Tuple.of(by.stream()
-                .collect(ImmutableMap
-                    .toImmutableMap(labelKey -> labelKey, labelKey -> sample.labels.getOrDefault(labelKey, ""))), sample))
+                .collect(toImmutableMap(labelKey -> labelKey, labelKey -> sample.labels.getOrDefault(labelKey, ""))), sample))
             .collect(groupingBy(Tuple2::_1, mapping(Tuple2::_2, toList())))
             .entrySet().stream()
             .map(entry -> newSample(entry.getKey(), entry.getValue().get(0).timestamp, entry.getValue().stream()
-                .mapToDouble(s -> s.value).reduce(Double::sum).orElse(0.0D)))
+                .mapToDouble(s -> s.value).reduce(aggregator).orElse(0.0D)))
             .toArray(Sample[]::new));
     }
 

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

@@ -86,15 +86,15 @@ public class PrometheusMetricConverter {
     private Stream<Tuple2<String, SampleFamily>> convertMetric(Metric metric) {
         return Match(metric).of(
             Case($(instanceOf(Histogram.class)), t -> Stream.of(
-                Tuple.of(metric.getName() + "_count", SampleFamilyBuilder.newBuilder(Sample.builder().name(metric.getName() + "_count")
+                Tuple.of(escapedName(metric.getName() + "_count"), SampleFamilyBuilder.newBuilder(Sample.builder().name(escapedName(metric.getName() + "_count"))
                     .timestamp(metric.getTimestamp()).labels(ImmutableMap.copyOf(metric.getLabels())).value(((Histogram) metric).getSampleCount()).build()).build()),
-                Tuple.of(metric.getName() + "_sum", SampleFamilyBuilder.newBuilder(Sample.builder().name(metric.getName() + "_sum")
+                Tuple.of(escapedName(metric.getName() + "_sum"), SampleFamilyBuilder.newBuilder(Sample.builder().name(escapedName(metric.getName() + "_sum"))
                     .timestamp(metric.getTimestamp()).labels(ImmutableMap.copyOf(metric.getLabels())).value(((Histogram) metric).getSampleSum()).build()).build()),
                 convertToSample(metric).orElse(NIL))),
             Case($(instanceOf(Summary.class)), t -> Stream.of(
-                Tuple.of(metric.getName() + "_count", SampleFamilyBuilder.newBuilder(Sample.builder().name(metric.getName() + "_count")
+                Tuple.of(escapedName(metric.getName() + "_count"), SampleFamilyBuilder.newBuilder(Sample.builder().name(escapedName(metric.getName() + "_count"))
                     .timestamp(metric.getTimestamp()).labels(ImmutableMap.copyOf(metric.getLabels())).value(((Summary) metric).getSampleCount()).build()).build()),
-                Tuple.of(metric.getName() + "_sum", SampleFamilyBuilder.newBuilder(Sample.builder().name(metric.getName() + "_sum")
+                Tuple.of(escapedName(metric.getName() + "_sum"), SampleFamilyBuilder.newBuilder(Sample.builder().name(escapedName(metric.getName() + "_sum"))
                     .timestamp(metric.getTimestamp()).labels(ImmutableMap.copyOf(metric.getLabels())).value(((Summary) metric).getSampleSum()).build()).build()),
                 convertToSample(metric).orElse(NIL))),
             Case($(), t -> Stream.of(convertToSample(metric).orElse(NIL)))
@@ -104,13 +104,13 @@ public class PrometheusMetricConverter {
     private Optional<Tuple2<String, SampleFamily>> convertToSample(Metric metric) {
         Sample[] ss = Match(metric).of(
             Case($(instanceOf(Counter.class)), t -> Collections.singletonList(Sample.builder()
-                .name(t.getName())
+                .name(escapedName(t.getName()))
                 .labels(ImmutableMap.copyOf(t.getLabels()))
                 .timestamp(t.getTimestamp())
                 .value(t.getValue())
                 .build())),
             Case($(instanceOf(Gauge.class)), t -> Collections.singletonList(Sample.builder()
-                .name(t.getName())
+                .name(escapedName(t.getName()))
                 .labels(ImmutableMap.copyOf(t.getLabels()))
                 .timestamp(t.getTimestamp())
                 .value(t.getValue())
@@ -118,7 +118,7 @@ public class PrometheusMetricConverter {
             Case($(instanceOf(Histogram.class)), t -> t.getBuckets()
                 .entrySet().stream()
                 .map(b -> Sample.builder()
-                    .name(t.getName())
+                    .name(escapedName(t.getName()))
                     .labels(ImmutableMap.<String, String>builder()
                         .putAll(t.getLabels())
                         .put("le", b.getKey().toString())
@@ -129,7 +129,7 @@ public class PrometheusMetricConverter {
             Case($(instanceOf(Summary.class)),
                 t -> t.getQuantiles().entrySet().stream()
                     .map(b -> Sample.builder()
-                        .name(t.getName())
+                        .name(escapedName(t.getName()))
                         .labels(ImmutableMap.<String, String>builder()
                             .putAll(t.getLabels())
                             .put("quantile", b.getKey().toString())
@@ -141,6 +141,11 @@ public class PrometheusMetricConverter {
         if (ss.length < 1) {
             return Optional.empty();
         }
-        return Optional.of(Tuple.of(metric.getName(), SampleFamilyBuilder.newBuilder(ss).build()));
+        return Optional.of(Tuple.of(escapedName(metric.getName()), SampleFamilyBuilder.newBuilder(ss).build()));
+    }
+
+    // Returns the escaped name of the given one, with "." replaced by "_"
+    protected String escapedName(final String name) {
+        return name.replaceAll("\\.", "_");
     }
 }

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

@@ -81,6 +81,93 @@ public class AggregationTest {
                 ).build()),
                 false,
             },
+
+            {
+                "min",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t3")).value(100).build(),
+                    Sample.builder().labels(of("idc", "t1")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(3).build()
+                ).build()),
+                "http_success_request.min()",
+                Result.success(SampleFamilyBuilder.newBuilder(Sample.builder().labels(ImmutableMap.of()).value(3).build()).build()),
+                false,
+            },
+            {
+                "min-by",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn", "svc", "catalog")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us", "svc", "product")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us", "instance", "10.0.0.1")).value(100).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn", "instance", "10.0.0.1")).value(3).build()
+                ).build()),
+                "http_success_request.min(by = ['region', 'idc'])",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1", "region", "")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn")).value(3).build()
+                ).build()),
+                false,
+            },
+
+            {
+                "max",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t3")).value(100).build(),
+                    Sample.builder().labels(of("idc", "t1")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(3).build()
+                ).build()),
+                "http_success_request.max()",
+                Result.success(SampleFamilyBuilder.newBuilder(Sample.builder().labels(ImmutableMap.of()).value(100).build()).build()),
+                false,
+            },
+            {
+                "max-by",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn", "svc", "catalog")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us", "svc", "product")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us", "instance", "10.0.0.1")).value(100).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn", "instance", "10.0.0.1")).value(3).build()
+                ).build()),
+                "http_success_request.max(by = ['region', 'idc'])",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1", "region", "")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us")).value(100).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn")).value(50).build()
+                ).build()),
+                false,
+            },
+
+            {
+                "avg",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t3")).value(100).build(),
+                    Sample.builder().labels(of("idc", "t1")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(3).build()
+                ).build()),
+                "http_success_request.avg()",
+                Result.success(SampleFamilyBuilder.newBuilder(Sample.builder().labels(ImmutableMap.of()).value(51).build()).build()),
+                false,
+            },
+            {
+                "avg-by",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn", "svc", "catalog")).value(51).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us", "svc", "product")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us", "instance", "10.0.0.1")).value(100).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn", "instance", "10.0.0.1")).value(3).build()
+                ).build()),
+                "http_success_request.avg(by = ['region', 'idc'])",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1", "region", "")).value(50).build(),
+                    Sample.builder().labels(of("idc", "t1", "region", "us")).value(75).build(),
+                    Sample.builder().labels(of("idc", "t3", "region", "cn")).value(27).build()
+                ).build()),
+                false,
+            },
         });
     }
 
@@ -102,4 +189,4 @@ public class AggregationTest {
         }
         assertThat(r, is(want));
     }
-}
\ No newline at end of file
+}

oap-server/server-bootstrap/pom.xml

@@ -259,6 +259,7 @@
                         <exclude>endpoint-name-grouping.yml</exclude>
                         <exclude>oal/</exclude>
                         <exclude>fetcher-prom-rules/</exclude>
+                        <exclude>envoy-metrics-rules/</exclude>
                         <exclude>meter-analyzer-config/</exclude>
                         <exclude>otel-oc-rules/</exclude>
                         <exclude>ui-initialized-templates/</exclude>
@@ -267,4 +268,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
\ No newline at end of file
+</project>

oap-server/server-bootstrap/src/main/resources/envoy-metrics-rules/envoy.yaml

+# 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.
+
+# This will parse a textual representation of a duration. The formats
+# accepted are based on the ISO-8601 duration format {@code PnDTnHnMn.nS}
+# with days considered to be exactly 24 hours.
+# <p>
+# Examples:
+# <pre>
+#    "PT20.345S" -- parses as "20.345 seconds"
+#    "PT15M"     -- parses as "15 minutes" (where a minute is 60 seconds)
+#    "PT10H"     -- parses as "10 hours" (where an hour is 3600 seconds)
+#    "P2D"       -- parses as "2 days" (where a day is 24 hours or 86400 seconds)
+#    "P2DT3H4M"  -- parses as "2 days, 3 hours and 4 minutes"
+#    "P-6H3M"    -- parses as "-6 hours and +3 minutes"
+#    "-P6H3M"    -- parses as "-6 hours and -3 minutes"
+#    "-P-6H+3M"  -- parses as "+6 hours and -3 minutes"
+# </pre>
+
+expSuffix: tag({tags -> tags.cluster = 'istio-dp::' + tags.cluster}).instance(['cluster'], ['instance'])
+metricPrefix: envoy
+metricsRules:
+  - name: heap_memory_used
+    exp: server_memory_heap_size
+  - name: heap_memory_max_used
+    exp: server_memory_heap_size.max(['cluster', 'instance'])
+  - name: memory_allocated
+    exp: server_memory_allocated
+  - name: memory_allocated_max
+    exp: server_memory_allocated.max(['cluster', 'instance'])
+  - name: memory_physical_size
+    exp: server_memory_physical_size
+  - name: memory_physical_size_max
+    exp: server_memory_physical_size.max(['cluster', 'instance'])
+
+  - name: total_connections_used
+    exp: server_total_connections.max(['cluster', 'instance'])
+  - name: parent_connections_used
+    exp: server_parent_connections.max(['cluster', 'instance'])
+
+  - name: worker_threads
+    exp: server_concurrency
+  - name: worker_threads_max
+    exp: server_concurrency.max(['cluster', 'instance'])
+
+  - name: bug_failures
+    exp: server_envoy_bug_failures

oap-server/server-bootstrap/src/main/resources/oal/envoy.oal

-/*
- * 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.
- *
- */
-
-// Envoy instance metrics
-envoy_heap_memory_max_used = from(EnvoyInstanceMetric.value).filter(metricName == "server.memory_heap_size").maxDouble();
-envoy_total_connections_used = from(EnvoyInstanceMetric.value).filter(metricName == "server.total_connections").maxDouble();
-envoy_parent_connections_used = from(EnvoyInstanceMetric.value).filter(metricName == "server.parent_connections").maxDouble();
\ No newline at end of file

oap-server/server-bootstrap/src/main/resources/ui-initialized-templates/istio-dp.yml

+# 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.
+
+templates:
+  - name: "Istio Data Plane"
+    type: "DASHBOARD"
+    configuration: |-
+      [
+        {
+          "name": "Istio Data Plane",
+          "type": "service",
+          "serviceGroup": "istio-dp",
+          "children": [
+            {
+              "name": "Data Plane",
+              "children": [
+                {
+                  "width": "3",
+                  "title": "Heap Memory Used",
+                  "height": 350,
+                  "entityType": "ServiceInstance",
+                  "independentSelector": false,
+                  "metricType": "REGULAR_VALUE",
+                  "metricName": "envoy_heap_memory_max_used,envoy_heap_memory_used,envoy_memory_allocated_max,envoy_memory_allocated,envoy_memory_physical_size,envoy_memory_physical_size_max",
+                  "queryMetricType": "readMetricsValues",
+                  "chartType": "ChartLine",
+                  "unit": "MB",
+                  "aggregation": "/",
+                  "aggregationNum": "1048576"
+                },
+                {
+                  "width": "3",
+                  "title": "Connections Used",
+                  "height": 350,
+                  "entityType": "ServiceInstance",
+                  "independentSelector": false,
+                  "metricType": "REGULAR_VALUE",
+                  "metricName": "envoy_total_connections_used,envoy_parent_connections_used",
+                  "queryMetricType": "readMetricsValues",
+                  "chartType": "ChartLine"
+                },
+                {
+                  "width": "3",
+                  "title": "Concurrency",
+                  "height": 350,
+                  "entityType": "ServiceInstance",
+                  "independentSelector": false,
+                  "metricType": "REGULAR_VALUE",
+                  "metricName": "envoy_worker_threads,envoy_worker_threads_max",
+                  "queryMetricType": "readMetricsValues",
+                  "chartType": "ChartLine"
+                },
+                {
+                  "width": "3",
+                  "title": "Envoy Bug Failure",
+                  "height": 350,
+                  "entityType": "ServiceInstance",
+                  "independentSelector": false,
+                  "metricType": "REGULAR_VALUE",
+                  "metricName": "envoy_bug_failures",
+                  "queryMetricType": "readMetricsValues",
+                  "chartType": "ChartLine"
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    activated: true
+    disabled: false

oap-server/server-bootstrap/src/main/resources/ui-initialized-templates/istio.yml

@@ -172,4 +172,4 @@ templates:
     # False means providing a basic template, user needs to add it manually.
     activated: true
     # True means wouldn't show up on the dashboard. Only keeps the definition in the storage.
-    disabled: false
\ No newline at end of file
+    disabled: false

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/pom.xml

@@ -34,6 +34,11 @@
             <artifactId>receiver-proto</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>meter-analyzer</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.skywalking</groupId>
             <artifactId>skywalking-mesh-receiver-plugin</artifactId>

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/EnvoyMetricReceiverConfig.java

@@ -24,7 +24,10 @@ import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 import lombok.Getter;
+import org.apache.skywalking.oap.meter.analyzer.prometheus.rule.Rule;
+import org.apache.skywalking.oap.meter.analyzer.prometheus.rule.Rules;
 import org.apache.skywalking.oap.server.library.module.ModuleConfig;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 
 public class EnvoyMetricReceiverConfig extends ModuleConfig {
     @Getter
@@ -39,4 +42,8 @@ public class EnvoyMetricReceiverConfig extends ModuleConfig {
         }
         return Arrays.stream(alsHTTPAnalysis.trim().split(",")).map(String::trim).collect(Collectors.toList());
     }
+
+    public List<Rule> rules() throws ModuleStartException {
+        return Rules.loadRules("envoy-metrics-rules", Collections.singletonList("envoy"));
+    }
 }

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/EnvoyMetricReceiverProvider.java

@@ -20,19 +20,21 @@ package org.apache.skywalking.oap.server.receiver.envoy;
 
 import org.apache.skywalking.aop.server.receiver.mesh.MeshReceiverModule;
 import org.apache.skywalking.oap.server.core.CoreModule;
-import org.apache.skywalking.oap.server.core.oal.rt.OALEngineLoaderService;
 import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
 import org.apache.skywalking.oap.server.library.module.ModuleConfig;
 import org.apache.skywalking.oap.server.library.module.ModuleDefine;
 import org.apache.skywalking.oap.server.library.module.ModuleProvider;
 import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
+import org.apache.skywalking.oap.server.receiver.envoy.als.mx.FieldsHelper;
 import org.apache.skywalking.oap.server.receiver.sharing.server.SharingServerModule;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
 
 public class EnvoyMetricReceiverProvider extends ModuleProvider {
     private final EnvoyMetricReceiverConfig config;
 
+    protected String fieldMappingFile = "metadata-service-mapping.yaml";
+
     public EnvoyMetricReceiverProvider() {
         config = new EnvoyMetricReceiverConfig();
     }
@@ -54,7 +56,11 @@ public class EnvoyMetricReceiverProvider extends ModuleProvider {
 
     @Override
     public void prepare() throws ServiceNotProvidedException, ModuleStartException {
-
+        try {
+            FieldsHelper.SINGLETON.init(fieldMappingFile);
+        } catch (final Exception e) {
+            throw new ModuleStartException("Failed to load metadata-service-mapping.yaml", e);
+        }
     }
 
     @Override
@@ -63,12 +69,7 @@ public class EnvoyMetricReceiverProvider extends ModuleProvider {
                                                   .provider()
                                                   .getService(GRPCHandlerRegister.class);
         if (config.isAcceptMetricsService()) {
-            getManager().find(CoreModule.NAME)
-                        .provider()
-                        .getService(OALEngineLoaderService.class)
-                        .load(EnvoyOALDefine.INSTANCE);
-
-            final MetricServiceGRPCHandler handler = new MetricServiceGRPCHandler(getManager());
+            final MetricServiceGRPCHandler handler = new MetricServiceGRPCHandler(getManager(), config);
             service.addHandler(handler);
             service.addHandler(new MetricServiceGRPCHandlerV3(handler));
         }

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/EnvoyOALDefine.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.receiver.envoy;
-
-import org.apache.skywalking.oap.server.core.oal.rt.OALDefine;
-
-/**
- * Envoy OAl script includes the metrics related to Envoy only.
- */
-public class EnvoyOALDefine  extends OALDefine {
-    public static final EnvoyOALDefine INSTANCE = new EnvoyOALDefine();
-
-    private EnvoyOALDefine() {
-        super(
-            "oal/envoy.oal",
-            "org.apache.skywalking.oap.server.core.source"
-        );
-    }
-}
\ No newline at end of file

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/MetricServiceGRPCHandler.java

@@ -18,23 +18,31 @@
 
 package org.apache.skywalking.oap.server.receiver.envoy;
 
-import io.envoyproxy.envoy.config.core.v3.Node;
-import io.envoyproxy.envoy.service.metrics.v3.MetricsServiceGrpc;
+import io.envoyproxy.envoy.service.metrics.v2.MetricsServiceGrpc;
 import io.envoyproxy.envoy.service.metrics.v3.StreamMetricsMessage;
 import io.envoyproxy.envoy.service.metrics.v3.StreamMetricsResponse;
 import io.grpc.stub.StreamObserver;
 import io.prometheus.client.Metrics;
 import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.apm.util.StringUtil;
+import org.apache.skywalking.oap.meter.analyzer.prometheus.PrometheusMetricConverter;
 import org.apache.skywalking.oap.server.core.CoreModule;
 import org.apache.skywalking.oap.server.core.analysis.IDManager;
 import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
-import org.apache.skywalking.oap.server.core.source.EnvoyInstanceMetric;
+import org.apache.skywalking.oap.server.core.analysis.meter.MeterSystem;
 import org.apache.skywalking.oap.server.core.analysis.NodeType;
 import org.apache.skywalking.oap.server.core.source.ServiceInstanceUpdate;
 import org.apache.skywalking.oap.server.core.source.SourceReceiver;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Metric;
+import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
+import org.apache.skywalking.oap.server.receiver.envoy.als.mx.ServiceMetaInfoAdapter;
+import org.apache.skywalking.oap.server.receiver.envoy.metrics.adapters.ProtoMetricFamily2MetricsAdapter;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
 import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
@@ -44,10 +52,11 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 @Slf4j
 public class MetricServiceGRPCHandler extends MetricsServiceGrpc.MetricsServiceImplBase {
     private final SourceReceiver sourceReceiver;
-    private CounterMetrics counter;
-    private HistogramMetrics histogram;
+    private final CounterMetrics counter;
+    private final HistogramMetrics histogram;
+    private final List<PrometheusMetricConverter> converters;
 
-    public MetricServiceGRPCHandler(ModuleManager moduleManager) {
+    public MetricServiceGRPCHandler(final ModuleManager moduleManager, final EnvoyMetricReceiverConfig config) throws ModuleStartException {
         sourceReceiver = moduleManager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
         MetricsCreator metricsCreator = moduleManager.find(TelemetryModule.NAME)
                                                      .provider()
@@ -60,16 +69,23 @@ public class MetricServiceGRPCHandler extends MetricsServiceGrpc.MetricsServiceI
             "envoy_metric_in_latency", "The process latency of service metrics receiver", MetricsTag.EMPTY_KEY,
             MetricsTag.EMPTY_VALUE
         );
+
+        final MeterSystem meterSystem = moduleManager.find(CoreModule.NAME).provider().getService(MeterSystem.class);
+
+        converters = config.rules()
+                           .stream()
+                           .map(rule -> new PrometheusMetricConverter(rule, meterSystem))
+                           .collect(Collectors.toList());
     }
 
     @Override
     public StreamObserver<StreamMetricsMessage> streamMetrics(StreamObserver<StreamMetricsResponse> responseObserver) {
         return new StreamObserver<StreamMetricsMessage>() {
             private volatile boolean isFirst = true;
-            private String serviceName = null;
-            private String serviceInstanceName = null;
+            private ServiceMetaInfo service;
 
             @Override
+            @SneakyThrows
             public void onNext(StreamMetricsMessage message) {
                 if (log.isDebugEnabled()) {
                     log.debug("Received msg {}", message);
@@ -77,94 +93,41 @@ public class MetricServiceGRPCHandler extends MetricsServiceGrpc.MetricsServiceI
 
                 if (isFirst) {
                     isFirst = false;
-                    StreamMetricsMessage.Identifier identifier = message.getIdentifier();
-                    Node node = identifier.getNode();
-                    if (node != null) {
-                        String nodeId = node.getId();
-                        if (!StringUtil.isEmpty(nodeId)) {
-                            serviceInstanceName = nodeId;
-                        }
-                        String cluster = node.getCluster();
-                        if (!StringUtil.isEmpty(cluster)) {
-                            serviceName = cluster;
-                            if (serviceInstanceName == null) {
-                                serviceInstanceName = serviceName;
-                            }
-                        }
-                    }
-
-                    if (serviceName == null) {
-                        serviceName = serviceInstanceName;
-                    }
+                    service = new ServiceMetaInfoAdapter(message.getIdentifier().getNode().getMetadata());
                 }
 
                 if (log.isDebugEnabled()) {
-                    log.debug(
-                        "Envoy metrics reported from service[{}], service instance[{}]", serviceName,
-                        serviceInstanceName
-                    );
+                    log.debug("Envoy metrics reported from service[{}]", service);
                 }
 
-                if (StringUtil.isNotEmpty(serviceName) && StringUtil.isNotEmpty(serviceInstanceName)) {
+                if (service != null && StringUtil.isNotEmpty(service.getServiceName()) && StringUtil.isNotEmpty(service.getServiceInstanceName())) {
                     List<Metrics.MetricFamily> list = message.getEnvoyMetricsList();
                     boolean needHeartbeatUpdate = true;
-                    for (int i = 0; i < list.size(); i++) {
+
+                    for (final Metrics.MetricFamily metricFamily : list) {
                         counter.inc();
 
-                        final String serviceId = IDManager.ServiceID.buildId(serviceName, NodeType.Normal);
-                        final String serviceInstanceId = IDManager.ServiceInstanceID.buildId(
-                            serviceId, serviceInstanceName);
-
-                        HistogramMetrics.Timer timer = histogram.createTimer();
-                        try {
-                            Metrics.MetricFamily metricFamily = list.get(i);
-                            double value = 0;
-                            long timestamp = 0;
-                            switch (metricFamily.getType()) {
-                                case GAUGE:
-                                    for (Metrics.Metric metrics : metricFamily.getMetricList()) {
-                                        timestamp = metrics.getTimestampMs();
-                                        value = metrics.getGauge().getValue();
-
-                                        if (timestamp > 1000000000000000000L) {
-                                            /**
-                                             * Several versions of envoy in istio.deps send timestamp in nanoseconds,
-                                             * instead of milliseconds(protocol says).
-                                             *
-                                             * Sadly, but have to fix it forcedly.
-                                             *
-                                             * An example of timestamp is '1552303033488741055', clearly it is not in milliseconds.
-                                             *
-                                             * This should be removed in the future.
-                                             */
-                                            timestamp /= 1_000_000;
-                                        }
-
-                                        EnvoyInstanceMetric metricSource = new EnvoyInstanceMetric();
-                                        metricSource.setServiceId(serviceId);
-                                        metricSource.setServiceName(serviceName);
-                                        metricSource.setId(serviceInstanceId);
-                                        metricSource.setName(serviceInstanceName);
-                                        metricSource.setMetricName(metricFamily.getName());
-                                        metricSource.setValue(value);
-                                        metricSource.setTimeBucket(TimeBucket.getMinuteTimeBucket(timestamp));
-                                        sourceReceiver.receive(metricSource);
-                                    }
-                                    break;
-                                default:
-                                    continue;
-                            }
-                            if (needHeartbeatUpdate) {
+                        final String serviceId = IDManager.ServiceID.buildId(service.getServiceName(), NodeType.Normal);
+
+                        try (final HistogramMetrics.Timer ignored = histogram.createTimer()) {
+                            final ProtoMetricFamily2MetricsAdapter adapter = new ProtoMetricFamily2MetricsAdapter(metricFamily);
+                            final Stream<Metric> metrics = adapter.adapt().peek(it -> {
+                                it.getLabels().putIfAbsent("cluster", service.getServiceName());
+                                it.getLabels().putIfAbsent("instance", service.getServiceInstanceName());
+                            });
+                            converters.forEach(converter -> converter.toMeter(metrics));
+
+                            if (needHeartbeatUpdate && list.get(0).getMetricCount() > 0) {
+                                final long timestamp = adapter.adaptTimestamp(list.get(0).getMetric(0));
+
                                 // Send heartbeat
                                 ServiceInstanceUpdate serviceInstanceUpdate = new ServiceInstanceUpdate();
-                                serviceInstanceUpdate.setName(serviceInstanceName);
+                                serviceInstanceUpdate.setName(service.getServiceInstanceName());
                                 serviceInstanceUpdate.setServiceId(serviceId);
                                 serviceInstanceUpdate.setTimeBucket(TimeBucket.getMinuteTimeBucket(timestamp));
                                 sourceReceiver.receive(serviceInstanceUpdate);
                                 needHeartbeatUpdate = false;
                             }
-                        } finally {
-                            timer.finish();
                         }
                     }
                 }

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/metrics/adapters/ProtoMetricFamily2MetricsAdapter.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.receiver.envoy.metrics.adapters;
+
+import io.prometheus.client.Metrics;
+import java.util.Map;
+import java.util.stream.Stream;
+import lombok.RequiredArgsConstructor;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Gauge;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Metric;
+
+import static java.util.stream.Collectors.toMap;
+
+@RequiredArgsConstructor
+public class ProtoMetricFamily2MetricsAdapter {
+    protected final Metrics.MetricFamily metricFamily;
+
+    public Stream<Metric> adapt() {
+        switch (metricFamily.getType()) {
+            case GAUGE:
+                return metricFamily.getMetricList()
+                                   .stream()
+                                   .map(it -> Gauge.builder()
+                                                   .name(adaptMetricsName(it))
+                                                   .value(adaptValue(it))
+                                                   .timestamp(adaptTimestamp(it))
+                                                   .labels(adaptLabels(it))
+                                                   .build());
+            default:
+                return Stream.of();
+        }
+    }
+
+    @SuppressWarnings("unused")
+    public String adaptMetricsName(final Metrics.Metric metric) {
+        return metricFamily.getName();
+    }
+
+    public double adaptValue(final Metrics.Metric it) {
+        return it.getGauge().getValue();
+    }
+
+    public Map<String, String> adaptLabels(final Metrics.Metric metric) {
+        return metric.getLabelList()
+                     .stream()
+                     .collect(toMap(Metrics.LabelPair::getName, Metrics.LabelPair::getValue));
+    }
+
+    public long adaptTimestamp(final Metrics.Metric metric) {
+        long timestamp = metric.getTimestampMs();
+
+        if (timestamp > 1000000000000000000L) {
+            /*
+             * Several versions of envoy in istio.deps send timestamp in nanoseconds,
+             * instead of milliseconds(protocol says).
+             *
+             * Sadly, but have to fix it forcefully.
+             *
+             * An example of timestamp is '1552303033488741055', clearly it is not in milliseconds.
+             *
+             * This should be removed in the future.
+             */
+            timestamp /= 1_000_000;
+        }
+
+        return timestamp;
+    }
+}

test/e2e/e2e-common/src/main/java/org/apache/skywalking/e2e/utils/Yamls.java

@@ -38,6 +38,10 @@ public final class Yamls {
         <T> T as(final Class<T> klass);
     }
 
+    public static boolean exists(final String file) {
+        return new ClassPathResource(Envs.resolve(file)).exists();
+    }
+
     public static AsTypeBuilder load(final String file) throws IOException {
         final InputStream inputStream = new ClassPathResource(Envs.resolve(file)).getInputStream();
 

test/e2e/e2e-data/src/main/java/org/apache/skywalking/e2e/metrics/MetricsQuery.java

@@ -120,6 +120,18 @@ public class MetricsQuery extends AbstractQuery<MetricsQuery> {
         METER_INSTANCE_PERSISTENCE_EXECUTE_COUNT 
     };
 
+    public static String[] ALL_ENVOY_LINER_METRICS = {
+        "envoy_heap_memory_used",
+        "envoy_heap_memory_max_used",
+        "envoy_memory_allocated",
+        "envoy_memory_allocated_max",
+        "envoy_memory_physical_size",
+        "envoy_memory_physical_size_max",
+        "envoy_total_connections_used",
+        "envoy_worker_threads",
+        "envoy_worker_threads_max"
+    };
+
     public static String METER_INSTANCE_PERSISTENCE_EXECUTE_PERCENTILE = "meter_oap_instance_persistence_execute_percentile";
 
     public static String[] ALL_SO11Y_LABELED_METRICS = {

test/e2e/e2e-test/src/test/java/org/apache/skywalking/e2e/mesh/MetricsServiceE2E.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.e2e.mesh;
+
+import com.google.common.base.Strings;
+import java.net.URL;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.e2e.base.SkyWalkingTestAdapter;
+import org.apache.skywalking.e2e.base.TrafficController;
+import org.apache.skywalking.e2e.common.HostAndPort;
+import org.apache.skywalking.e2e.metrics.AtLeastOneOfMetricsMatcher;
+import org.apache.skywalking.e2e.metrics.MetricsValueMatcher;
+import org.apache.skywalking.e2e.metrics.ReadMetrics;
+import org.apache.skywalking.e2e.metrics.ReadMetricsQuery;
+import org.apache.skywalking.e2e.retryable.RetryableTest;
+import org.apache.skywalking.e2e.service.Service;
+import org.apache.skywalking.e2e.service.ServicesMatcher;
+import org.apache.skywalking.e2e.service.ServicesQuery;
+import org.apache.skywalking.e2e.service.instance.Instance;
+import org.apache.skywalking.e2e.service.instance.Instances;
+import org.apache.skywalking.e2e.service.instance.InstancesMatcher;
+import org.apache.skywalking.e2e.service.instance.InstancesQuery;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestInstance;
+
+import static org.apache.skywalking.e2e.metrics.MetricsQuery.ALL_ENVOY_LINER_METRICS;
+import static org.apache.skywalking.e2e.utils.Times.now;
+import static org.apache.skywalking.e2e.utils.Yamls.exists;
+import static org.apache.skywalking.e2e.utils.Yamls.load;
+
+@Slf4j
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class MetricsServiceE2E extends SkyWalkingTestAdapter {
+    private final String swWebappHost = Optional.ofNullable(Strings.emptyToNull(System.getenv("WEBAPP_HOST"))).orElse("127.0.0.1");
+
+    private final String swWebappPort = Optional.ofNullable(Strings.emptyToNull(System.getenv("WEBAPP_PORT"))).orElse("12800");
+
+    protected HostAndPort swWebappHostPort = HostAndPort.builder()
+                                                        .host(swWebappHost)
+                                                        .port(Integer.parseInt(swWebappPort))
+                                                        .build();
+
+    @BeforeAll
+    public void setUp() throws Exception {
+        LOGGER.info("set up");
+
+        queryClient(swWebappHostPort);
+
+        String gatewayHost = Strings.isNullOrEmpty(System.getenv("GATEWAY_HOST")) ? "127.0.0.1" : System.getenv("GATEWAY_HOST");
+        String gatewayPort = Strings.isNullOrEmpty(System.getenv("GATEWAY_PORT")) ? "80" : System.getenv("GATEWAY_PORT");
+
+        HostAndPort serviceHostPort = HostAndPort.builder()
+                                                 .host(gatewayHost)
+                                                 .port(Integer.parseInt(gatewayPort))
+                                                 .build();
+
+        final URL url = new URL("http", serviceHostPort.host(), serviceHostPort.port(), "/productpage");
+
+        trafficController =
+            TrafficController.builder()
+                             .logResult(false)
+                             .sender(() -> restTemplate.getForEntity(url.toURI(), String.class))
+                             .build()
+                             .start();
+
+        LOGGER.info("set up done");
+    }
+
+    @RetryableTest
+    void test() throws Exception {
+        List<Service> services = graphql.services(new ServicesQuery().start(startTime).end(now()));
+
+        services = services.stream().filter(s -> s.getLabel().startsWith("istio-dp::")).collect(Collectors.toList());
+        LOGGER.info("services: {}", services);
+        load("expected/metricsservice/services.yml").as(ServicesMatcher.class).verify(services);
+        for (final Service service : services) {
+            if (service.getLabel().contains("egressgateway")) {
+                continue;
+            }
+
+            final Instances instances = graphql.instances(
+                new InstancesQuery().serviceId(service.getKey()).start(startTime).end(now())
+            );
+
+            LOGGER.info("instances: {}", instances);
+
+            String instancesFile = "expected/metricsservice/instances-" + service.getLabel() + ".yml";
+            instancesFile = instancesFile.replaceAll("::", "-");
+            if (!exists(instancesFile)) {
+                instancesFile = "expected/metricsservice/instances.yml";
+            }
+            load(instancesFile).as(InstancesMatcher.class).verify(instances);
+            for (Instance instance : instances.getInstances()) {
+                for (String metricsName : ALL_ENVOY_LINER_METRICS) {
+                    LOGGER.info("verifying service instance: {}", instance);
+                    final ReadMetrics instanceMetrics = graphql.readMetrics(
+                        new ReadMetricsQuery().stepByMinute().metricsName(metricsName)
+                                              .serviceName(service.getLabel()).instanceName(instance.getLabel())
+                    );
+
+                    LOGGER.info("{}: {}", metricsName, instanceMetrics);
+                    final AtLeastOneOfMetricsMatcher instanceRespTimeMatcher = new AtLeastOneOfMetricsMatcher();
+                    final MetricsValueMatcher greaterThanZero = new MetricsValueMatcher();
+                    greaterThanZero.setValue("gt 0");
+                    instanceRespTimeMatcher.setValue(greaterThanZero);
+                    instanceRespTimeMatcher.verify(instanceMetrics.getValues());
+                }
+            }
+        }
+    }
+}

test/e2e/e2e-test/src/test/resources/expected/metricsservice/instances-istio-dp-reviews.yml

+# 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.
+
+instances:
+  - key: not null
+    label: not null
+  - key: not null
+    label: not null
+  - key: not null
+    label: not null

test/e2e/e2e-test/src/test/resources/expected/metricsservice/instances.yml

+# 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.
+
+instances:
+  - key: not null
+    label: not null

test/e2e/e2e-test/src/test/resources/expected/metricsservice/services.yml

+# 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.
+
+services:
+  - key: not null
+    label: istio-dp::ratings
+  - key: not null
+    label: istio-dp::reviews
+  - key: not null
+    label: istio-dp::productpage
+  - key: not null
+    label: istio-dp::details
+  - key: not null
+    label: istio-dp::istio-ingressgateway
+  - key: not null
+    label: istio-dp::istio-egressgateway