Repository: apache/skywalking (language: python)

Issue:
[Feature] adapt Istio ALS new peer metadata attributes
### Search before asking

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


### Description

https://github.com/istio/istio/pull/53274/files

### Use case

_No response_

### Related issues

_No response_

### Are you willing to submit a pull request to implement this on your own?

- [X] Yes I am willing to submit a pull request on my own!

### Code of Conduct

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



Discussion / hints:
Just tried out https://github.com/istio/istio/releases/tag/1.24.0-rc.0 and the filter_state becomes empty always, according to


> Peer metadata is now stored
>     in `filter_state.downstream_peer` and `filter_state.upstream_peer` instead
>     of `filter_state["wasm.downstream_peer"]` and
>     `filter_state["wasm.upstream_peer"]`.

it should have downstream_peer upstream_peer but not, after downgrading to 1.22, `filter_state["wasm.downstream_peer”]` and
     `filter_state["wasm.upstream_peer”]`  appear again so it’s probably not an issue of my setup



Tracked by https://github.com/istio/istio/issues/53769
a demo iop file as following:
```yaml
apiVersion: install.istio.io/v1alpha1
kind: IstioOperator
spec:
  meshConfig:
    accessLogFormat: |
      %FILTER_STATE(upstream_peer)% %FILTER_STATE(downstream_peer)% [%START_TIME%] "%REQ(:METHOD)% %REQ(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL%" %RESPONSE_CODE% %RESPONSE_FLAGS% %RESPONSE_CODE_DETAILS% %CONNECTION_TERMINATION_DETAILS% "%UPSTREAM_TRANSPORT_FAILURE_REASON%" %BYTES_RECEIVED% %BYTES_SENT% %DURATION% %RESP(X-ENVOY-UPSTREAM-SERVICE-TIME)% "%REQ(X-FORWARDED-FOR)%" "%REQ(USER-AGENT)%" "%REQ(X-REQUEST-ID)%" "%REQ(:AUTHORITY)%" "%UPSTREAM_HOST%" %UPSTREAM_CLUSTER% %UPSTREAM_LOCAL_ADDRESS% %DOWNSTREAM_LOCAL_ADDRESS% %DOWNSTREAM_REMOTE_ADDRESS% %REQUESTED_SERVER_NAME% %ROUTE_NAME% requestSize=%BYTES_RECEIVED% responseSize=%BYTES_SENT% requestPath=%REQ(X-ENVOY-ORIGINAL-PATH?:PATH)%
    extensionProviders:
      - name: grpc-als
        envoyHttpAls:
          service: envoy-als.monitoring.svc.cluster.local
          port: 8080
    defaultProviders:
      accessLogging:
        - grpc-als
        - envoy
```
```console
{"namespace":"default","service":"httpbin","revision":"v1","workload":"httpbin","name":"httpbin-7c9544f45c-h9vs8","cluster":"Kubernetes","version":"v1","type":"deployment","app":"httpbin"} - [2024-11-05T14:11:49.246Z] "GET /get HTTP/1.1" 200 - via_upstream - "-" 0 641 143 135 "-" "curl/8.10.1" "a55da549-1fc1-4167-9465-5ab575a4964a" "httpbin:8000" "10.244.0.46:8080" outbound|8000||httpbin.default.svc.cluster.local; 10.244.0.45:43480 10.96.51.241:8000 10.244.0.45:50248 - default requestSize=0 responseSize=641 requestPath=/get
```

```console
filter_state_objects:{key:"upstream_peer" value:{[type.googleapis.com/google.protobuf.Struct]:{fields:{key:"app" value:{string_value:"httpbin"}} fields:{key:"cluster" value:{string_value:"Kubernetes"}} fields:{key:"name" value:{string_value:"httpbin-7c9544f45c-h9vs8"}} fields:{key:"namespace" value:{string_value:"default"}} fields:{key:"revision" value:{string_value:"v1"}} fields:{key:"service" value:{string_value:"httpbin"}} fields:{key:"type" value:{string_value:"deployment"}} fields:{key:"version" value:{string_value:"v1"}} fields:{key:"workload" value:{string_value:"httpbin"}}}}}
```

```console
- {"namespace":"default","revision":"latest","app":"sleep","name":"sleep-646c4b65d-ktmrt","service":"sleep","workload":"sleep","cluster":"Kubernetes","type":"deployment"} [2024-11-05T14:11:49.264Z] "GET /get HTTP/1.1" 200 - via_upstream - "-" 0 641 7 1 "-" "curl/8.10.1" "a55da549-1fc1-4167-9465-5ab575a4964a" "httpbin:8000" "10.244.0.46:8080" inbound|8080||; 127.0.0.6:41867 10.244.0.46:8080 10.244.0.45:43480 invalid:outbound_.8000_._.httpbin.default.svc.cluster.local default requestSize=0 responseSize=641 requestPath=/get
```

```console
filter_state_objects:{key:"downstream_peer" value:{[type.googleapis.com/google.protobuf.Struct]:{fields:{key:"app" value:{string_value:"sleep"}} fields:{key:"cluster" value:{string_value:"Kubernetes"}} fields:{key:"name" value:{string_value:"sleep-646c4b65d-ktmrt"}} fields:{key:"namespace" value:{string_value:"default"}} fields:{key:"revision" value:{string_value:"latest"}} fields:{key:"service" value:{string_value:"sleep"}} fields:{key:"type" value:{string_value:"deployment"}} fields:{key:"workload" value:{string_value:"sleep"}}}}}
```

Fix patch:
diff --git a/.github/workflows/skywalking.yaml b/.github/workflows/skywalking.yaml
index 855a3e3c1793..8edb000dc007 100644
--- a/.github/workflows/skywalking.yaml
+++ b/.github/workflows/skywalking.yaml
@@ -555,26 +555,31 @@ jobs:
           - name: Kafka Log
             config: test/e2e-v2/cases/kafka/log/e2e.yaml
 
-          - name: Istio Metrics Service 1.15.0
+          - name: Istio Metrics Service 1.20.0
             config: test/e2e-v2/cases/istio/metrics/e2e.yaml
             env: |
-              ISTIO_VERSION=1.15.0
-              KUBERNETES_VERSION=25
-          - name: Istio Metrics Service 1.16.0
+              ISTIO_VERSION=1.20.0
+              KUBERNETES_VERSION=28
+          - name: Istio Metrics Service 1.21.0
             config: test/e2e-v2/cases/istio/metrics/e2e.yaml
             env: |
-              ISTIO_VERSION=1.16.0
-              KUBERNETES_VERSION=25
-          - name: Istio Metrics Service 1.17.0
+              ISTIO_VERSION=1.21.0
+              KUBERNETES_VERSION=28
+          - name: Istio Metrics Service 1.22.0
             config: test/e2e-v2/cases/istio/metrics/e2e.yaml
             env: |
-              ISTIO_VERSION=1.17.0
-              KUBERNETES_VERSION=25
-          - name: Istio Metrics Service 1.18.0
+              ISTIO_VERSION=1.22.0
+              KUBERNETES_VERSION=28
+          - name: Istio Metrics Service 1.23.0
             config: test/e2e-v2/cases/istio/metrics/e2e.yaml
             env: |
-              ISTIO_VERSION=1.18.0
-              KUBERNETES_VERSION=25
+              ISTIO_VERSION=1.23.0
+              KUBERNETES_VERSION=28
+          - name: Istio Metrics Service 1.24.0
+            config: test/e2e-v2/cases/istio/metrics/e2e.yaml
+            env: |
+              ISTIO_VERSION=1.24.0
+              KUBERNETES_VERSION=28
 
           - name: Rover with Istio Process 1.15.0
             config: test/e2e-v2/cases/rover/process/istio/e2e.yaml
@@ -769,14 +774,16 @@ jobs:
       matrix:
         analyzer: [k8s-mesh, mx-mesh]
         versions:
-          - istio: 1.15.0
-            kubernetes: 25
-          - istio: 1.16.0
-            kubernetes: 25
-          - istio: 1.17.0
-            kubernetes: 25
-          - istio: 1.18.0
-            kubernetes: 25
+          - istio: 1.20.0
+            kubernetes: 28
+          - istio: 1.21.0
+            kubernetes: 28
+          - istio: 1.22.0
+            kubernetes: 28
+          - istio: 1.23.0
+            kubernetes: 28
+          - istio: 1.24.0
+            kubernetes: 28
 
     steps:
       - uses: actions/checkout@v3
diff --git a/docker/oap/log4j2.xml b/docker/oap/log4j2.xml
index 6152fc75c2b2..e5869e2914e2 100644
--- a/docker/oap/log4j2.xml
+++ b/docker/oap/log4j2.xml
@@ -28,7 +28,7 @@
         </Console>
     </Appenders>
     <Loggers>
-        <Root level="INFO">
+        <Root level="DEBUG">
             <AppenderRef ref="Console"/>
         </Root>
     </Loggers>
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 0617ec035e67..77b1ffb9dca3 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -19,6 +19,7 @@
 * Add max query size settings to BanyanDB.
 * Fix "BanyanDBTraceQueryDAO.queryBasicTraces" doesn't support querying by "trace_id".
 * Polish mesh data dispatcher: don't generate Instance/Endpoint metrics if they are empty.
+* Adapt the new metadata standardization in Istio 1.24.
 
 #### UI
 
diff --git a/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/FieldsHelper.java b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/FieldsHelper.java
index d5dcf38252f4..48fd9aabbfa8 100644
--- a/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/FieldsHelper.java
+++ b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/FieldsHelper.java
@@ -38,6 +38,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.BiConsumer;
 import java.util.regex.Matcher;
@@ -166,7 +167,7 @@ public void init(final InputStream inputStream) {
     public void inflate(final Struct metadata, final Object target) {
         final Value empty = Value.newBuilder().setStringValue("-").build();
         final Value root = Value.newBuilder().setStructValue(metadata).build();
-        for (final Map.Entry<String, FieldFormat> entry : fieldNameMapping.entrySet()) {
+        for (final var entry : fieldNameMapping.entrySet()) {
             final FieldFormat fieldFormat = entry.getValue();
             final Object[] values = new String[fieldFormat.properties.size()];
             for (int i = 0; i < fieldFormat.properties.size(); i++) {
@@ -175,7 +176,9 @@ public void inflate(final Struct metadata, final Object target) {
                 for (final Field field : property) {
                     Value value = root;
                     for (final String segment : field.dsvSegments) {
-                        value = value.getStructValue().getFieldsOrDefault(segment, empty);
+                        final var fieldMaps = new TreeMap<String, Value>(String.CASE_INSENSITIVE_ORDER);
+                        fieldMaps.putAll(value.getStructValue().getFieldsMap());
+                        value = fieldMaps.getOrDefault(segment, empty);
                     }
                     if (Strings.isNullOrEmpty(value.getStringValue()) || "-".equals(value.getStringValue())) {
                         continue;
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java
index c8b03e112e5d..e8ad193c6572 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java
@@ -18,24 +18,30 @@
 
 package org.apache.skywalking.oap.server.receiver.envoy.als.mx;
 
-import com.google.protobuf.Any;
-import com.google.protobuf.TextFormat;
-import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
-import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
-import lombok.extern.slf4j.Slf4j;
+import static org.apache.skywalking.oap.server.core.Const.TLS_MODE.NON_TLS;
+
+import java.util.Base64;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics.Builder;
 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.FieldsHelper;
 import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
 import org.apache.skywalking.oap.server.receiver.envoy.als.AbstractALSAnalyzer;
+import org.apache.skywalking.oap.server.receiver.envoy.als.AccessLogAnalyzer.Result.ResultBuilder;
 import org.apache.skywalking.oap.server.receiver.envoy.als.Role;
 import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
 
-import java.util.Base64;
-import java.util.concurrent.atomic.AtomicBoolean;
+import com.google.protobuf.Any;
+import com.google.protobuf.Struct;
+import com.google.protobuf.TextFormat;
 
-import static org.apache.skywalking.oap.server.core.Const.TLS_MODE.NON_TLS;
+import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
+import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer {
@@ -44,6 +50,10 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer {
 
     public static final String DOWNSTREAM_KEY = "wasm.downstream_peer";
 
+    public static final String UPSTREAM_PEER = "upstream_peer";
+
+    public static final String DOWNSTREAM_PEER = "downstream_peer";
+
     protected String fieldMappingFile = "metadata-service-mapping.yaml";
 
     protected EnvoyMetricReceiverConfig config;
@@ -86,6 +96,9 @@ public Result analysis(
         final var properties = entry.getCommonProperties();
         final var stateMap = properties.getFilterStateObjectsMap();
         final var result = previousResult.toBuilder();
+        if (log.isDebugEnabled()) {
+            log.debug("Filter state object map: {}", stateMap);
+        }
         if (stateMap.isEmpty()) {
             return result.service(currSvc).build();
         }
@@ -93,6 +106,24 @@ public Result analysis(
         final var previousMetrics = previousResult.getMetrics();
         final var httpMetrics = previousMetrics.getHttpMetricsBuilder();
         final var downstreamExists = new AtomicBoolean();
+        parseFilterObject(previousResult, entry, role, currSvc, stateMap, result, httpMetrics, downstreamExists);
+        parseFilterObjectPrior124(previousResult, entry, role, currSvc, stateMap, result, httpMetrics, downstreamExists);
+        if (role.equals(Role.PROXY) && !downstreamExists.get()) {
+            final var metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics();
+            if (log.isDebugEnabled()) {
+                log.debug("Transformed a {} inbound mesh metric {}", role, TextFormat.shortDebugString(metric));
+            }
+            httpMetrics.addMetrics(metric);
+            result.hasDownstreamMetrics(true);
+        }
+        return result.metrics(previousMetrics.setHttpMetrics(httpMetrics)).service(currSvc).build();
+    }
+
+    // TODO: remove this when 1.24.0 is our minimum supported version.
+    @Deprecated(forRemoval = true)
+    private void parseFilterObjectPrior124(final Result previousResult, final HTTPAccessLogEntry entry, final Role role,
+            final ServiceMetaInfo currSvc, final Map<String, Any> stateMap, final ResultBuilder result,
+            final Builder httpMetrics, final AtomicBoolean downstreamExists) {
         stateMap.forEach((key, value) -> {
             if (!key.equals(UPSTREAM_KEY) && !key.equals(DOWNSTREAM_KEY)) {
                 return;
@@ -131,21 +162,63 @@ public Result analysis(
                     break;
             }
         });
-        if (role.equals(Role.PROXY) && !downstreamExists.get()) {
-            final var metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics();
+    }
+
+    private void parseFilterObject(final Result previousResult, final HTTPAccessLogEntry entry, final Role role,
+            final ServiceMetaInfo currSvc, final Map<String, Any> stateMap, final ResultBuilder result,
+            final Builder httpMetrics, final AtomicBoolean downstreamExists) {
+        stateMap.forEach((key, value) -> {
+            if (!key.equals(UPSTREAM_PEER) && !key.equals(DOWNSTREAM_PEER)) {
+                return;
+            }
             if (log.isDebugEnabled()) {
-                log.debug("Transformed a {} inbound mesh metric {}", role, TextFormat.shortDebugString(metric));
+                log.debug("Filter state object key: {}, value: {}", key, value);
             }
-            httpMetrics.addMetrics(metric);
-            result.hasDownstreamMetrics(true);
-        }
-        return result.metrics(previousMetrics.setHttpMetrics(httpMetrics)).service(currSvc).build();
+            final ServiceMetaInfo svc;
+            try {
+                log.debug("Filter state object value map: {}", value.unpack(Struct.class).getFieldsMap());
+                svc = adaptToServiceMetaInfo(value.unpack(Struct.class));
+            } catch (Exception e) {
+                log.error("Fail to parse metadata {} to FlatNode", Base64.getEncoder().encode(value.toByteArray()));
+                return;
+            }
+            final HTTPServiceMeshMetric.Builder metrics;
+            switch (key) {
+                case UPSTREAM_PEER:
+                    if (previousResult.hasUpstreamMetrics()) {
+                        break;
+                    }
+                    metrics = newAdapter(entry, currSvc, svc).adaptToUpstreamMetrics().setTlsMode(NON_TLS);
+                    if (log.isDebugEnabled()) {
+                        log.debug("Transformed a {} outbound mesh metrics {}", role, TextFormat.shortDebugString(metrics));
+                    }
+                    httpMetrics.addMetrics(metrics);
+                    result.hasUpstreamMetrics(true);
+                    break;
+                case DOWNSTREAM_PEER:
+                    if (previousResult.hasDownstreamMetrics()) {
+                        break;
+                    }
+                    metrics = newAdapter(entry, svc, currSvc).adaptToDownstreamMetrics();
+                    if (log.isDebugEnabled()) {
+                        log.debug("Transformed a {} inbound mesh metrics {}", role, TextFormat.shortDebugString(metrics));
+                    }
+                    httpMetrics.addMetrics(metrics);
+                    downstreamExists.set(true);
+                    result.hasDownstreamMetrics(true);
+                    break;
+            }
+        });
     }
 
     protected ServiceMetaInfo adaptToServiceMetaInfo(final Any value) throws Exception {
         return new ServiceMetaInfoAdapter(value);
     }
 
+    protected ServiceMetaInfo adaptToServiceMetaInfo(final Struct struct) throws Exception {
+        return config.serviceMetaInfoFactory().fromStruct(struct);
+    }
+
     protected ServiceMetaInfo adaptToServiceMetaInfo(final StreamAccessLogsMessage.Identifier identifier) throws Exception {
         return config.serviceMetaInfoFactory().fromStruct(identifier.getNode().getMetadata());
     }
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/ServiceMetaInfoAdapter.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/ServiceMetaInfoAdapter.java
index 902b78a6ca17..c3d4184daf69 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/ServiceMetaInfoAdapter.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/ServiceMetaInfoAdapter.java
@@ -29,6 +29,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -65,6 +66,9 @@ public ServiceMetaInfoAdapter(final ByteString bv) throws Exception {
         }
 
         final Struct metadata = requireNonNull(extractStructFromNodeFlatBuffer(flatNode));
+        if (log.isDebugEnabled()) {
+            log.debug("Node metadata: {}", metadata);
+        }
         FieldsHelper.forClass(this.getClass().getSuperclass()).inflate(metadata, this);
         appendTags(metadata);
     }
@@ -111,10 +115,17 @@ public static Struct extractStructFromNodeFlatBuffer(final FlatNode node) {
     public ServiceMetaInfoAdapter(final Struct metadata) {
         FieldsHelper.forClass(this.getClass().getSuperclass()).inflate(requireNonNull(metadata), this);
         appendTags(requireNonNull(metadata));
+        if (log.isDebugEnabled()) {
+            log.info("Metadata is converted to: {}", this);
+        }
     }
 
     private void appendTags(Struct metadata) {
-        final Map<String, Value> fieldsMap = metadata.getFieldsMap();
+        final Map<String, Value> fieldsMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+        fieldsMap.putAll(metadata.getFieldsMap());
+        if (log.isDebugEnabled()) {
+            log.debug("Metadata field map: {}", fieldsMap);
+        }
         final List<KeyValue> tags = new ArrayList<>();
         if (fieldsMap.containsKey("NAME")) {
             tags.add(new KeyValue("pod", fieldsMap.get("NAME").getStringValue()));
@@ -122,6 +133,9 @@ private void appendTags(Struct metadata) {
         if (fieldsMap.containsKey("NAMESPACE")) {
             tags.add(new KeyValue("namespace", fieldsMap.get("NAMESPACE").getStringValue()));
         }
+        if (log.isDebugEnabled()) {
+            log.debug("Converted tags: {}", tags);
+        }
         if (!tags.isEmpty()) {
             this.setTags(tags);
         }
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java
index 5d2e67835274..469f49e50780 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java
@@ -19,16 +19,19 @@
 package org.apache.skywalking.oap.server.receiver.envoy.als.tcp.mx;
 
 import com.google.protobuf.Any;
+import com.google.protobuf.Struct;
 import com.google.protobuf.TextFormat;
 import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
 import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
 import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetrics.Builder;
 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.FieldsHelper;
 import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
+import org.apache.skywalking.oap.server.receiver.envoy.als.AccessLogAnalyzer.Result.ResultBuilder;
 import org.apache.skywalking.oap.server.receiver.envoy.als.Role;
 import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
 import org.apache.skywalking.oap.server.receiver.envoy.als.mx.ServiceMetaInfoAdapter;
@@ -40,7 +43,9 @@
 
 import static org.apache.skywalking.oap.server.core.Const.TLS_MODE.NON_TLS;
 import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.DOWNSTREAM_KEY;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.DOWNSTREAM_PEER;
 import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.UPSTREAM_KEY;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.UPSTREAM_PEER;
 
 @Slf4j
 public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyzer {
@@ -93,6 +98,23 @@ public Result analysis(
 
         final var tcpMetrics = previousMetrics.getTcpMetricsBuilder();
         final var downstreamExists = new AtomicBoolean();
+        parseFilterObject(previousResult, entry, role, currSvc, stateMap, newResult, tcpMetrics, downstreamExists);
+        parseFilterObjectPrior124(previousResult, entry, role, currSvc, stateMap, newResult, tcpMetrics, downstreamExists);
+        if (role.equals(Role.PROXY) && !downstreamExists.get()) {
+            final TCPServiceMeshMetric.Builder metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics();
+            if (log.isDebugEnabled()) {
+                log.debug("Transformed a {} inbound mesh metric {}", role, TextFormat.shortDebugString(metric));
+            }
+            tcpMetrics.addMetrics(metric);
+        }
+        return newResult.metrics(previousMetrics.setTcpMetrics(tcpMetrics)).service(currSvc).build();
+    }
+
+    // TODO: remove this when 1.24.0 is our minimum supported version.
+    @Deprecated(forRemoval = true)
+    private void parseFilterObjectPrior124(final Result previousResult, final TCPAccessLogEntry entry, final Role role,
+            final ServiceMetaInfo currSvc, final Map<String, Any> stateMap, final ResultBuilder newResult,
+            final Builder tcpMetrics, final AtomicBoolean downstreamExists) {
         stateMap.forEach((key, value) -> {
             if (!key.equals(UPSTREAM_KEY) && !key.equals(DOWNSTREAM_KEY)) {
                 return;
@@ -131,20 +153,63 @@ public Result analysis(
                     break;
             }
         });
-        if (role.equals(Role.PROXY) && !downstreamExists.get()) {
-            final TCPServiceMeshMetric.Builder metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics();
+    }
+
+    private void parseFilterObject(final Result previousResult, final TCPAccessLogEntry entry, final Role role,
+            final ServiceMetaInfo currSvc, final Map<String, Any> stateMap, final ResultBuilder newResult,
+            final Builder tcpMetrics, final AtomicBoolean downstreamExists) {
+        stateMap.forEach((key, value) -> {
+            if (!key.equals(UPSTREAM_PEER) && !key.equals(DOWNSTREAM_PEER)) {
+                return;
+            }
             if (log.isDebugEnabled()) {
-                log.debug("Transformed a {} inbound mesh metric {}", role, TextFormat.shortDebugString(metric));
+                log.debug("Filter state object key: {}, value: {}", key, value);
             }
-            tcpMetrics.addMetrics(metric);
-        }
-        return newResult.metrics(previousMetrics.setTcpMetrics(tcpMetrics)).service(currSvc).build();
+            final ServiceMetaInfo svc;
+            try {
+                log.debug("Filter state object value map: {}", value.unpack(Struct.class).getFieldsMap());
+                svc = adaptToServiceMetaInfo(value.unpack(Struct.class));
+            } catch (Exception e) {
+                log.error("Fail to parse metadata {} to FlatNode", Base64.getEncoder().encode(value.toByteArray()));
+                return;
+            }
+            final TCPServiceMeshMetric.Builder metrics;
+            switch (key) {
+                case UPSTREAM_PEER:
+                    if (previousResult.hasUpstreamMetrics()) {
+                        break;
+                    }
+                    metrics = newAdapter(entry, currSvc, svc).adaptToUpstreamMetrics().setTlsMode(NON_TLS);
+                    if (log.isDebugEnabled()) {
+                        log.debug("Transformed a {} outbound mesh metrics {}", role, TextFormat.shortDebugString(metrics));
+                    }
+                    tcpMetrics.addMetrics(metrics);
+                    newResult.hasUpstreamMetrics(true);
+                    break;
+                case DOWNSTREAM_PEER:
+                    if (previousResult.hasDownstreamMetrics()) {
+                        break;
+                    }
+                    metrics = newAdapter(entry, svc, currSvc).adaptToDownstreamMetrics();
+                    if (log.isDebugEnabled()) {
+                        log.debug("Transformed a {} inbound mesh metrics {}", role, TextFormat.shortDebugString(metrics));
+                    }
+                    tcpMetrics.addMetrics(metrics);
+                    downstreamExists.set(true);
+                    newResult.hasDownstreamMetrics(true);
+                    break;
+            }
+        });
     }
 
     protected ServiceMetaInfo adaptToServiceMetaInfo(final Any value) throws Exception {
         return new ServiceMetaInfoAdapter(value);
     }
 
+    protected ServiceMetaInfo adaptToServiceMetaInfo(final Struct struct) throws Exception {
+        return config.serviceMetaInfoFactory().fromStruct(struct);
+    }
+
     protected ServiceMetaInfo adaptToServiceMetaInfo(final StreamAccessLogsMessage.Identifier identifier) throws Exception {
         return config.serviceMetaInfoFactory().fromStruct(identifier.getNode().getMetadata());
     }
diff --git a/oap-server/server-starter/src/main/resources/cilium-rules/metadata-service-mapping.yaml b/oap-server/server-starter/src/main/resources/cilium-rules/metadata-service-mapping.yaml
index 6246e987fc14..a201089f37e4 100644
--- a/oap-server/server-starter/src/main/resources/cilium-rules/metadata-service-mapping.yaml
+++ b/oap-server/server-starter/src/main/resources/cilium-rules/metadata-service-mapping.yaml
@@ -13,5 +13,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-serviceName: ${LABELS."service.istio.io/canonical-name",LABELS."app.kubernetes.io/name",LABELS.component,LABELS.app,LABELS.k8s-app}.${NAMESPACE}
+serviceName: ${LABELS."service.istio.io/canonical-name",LABELS."app.kubernetes.io/name",LABELS.component,LABELS.app,LABELS.k8s-app,SERVICE}.${NAMESPACE}
 serviceInstanceName: ${NAME}
diff --git a/oap-server/server-starter/src/main/resources/metadata-service-mapping.yaml b/oap-server/server-starter/src/main/resources/metadata-service-mapping.yaml
index cf9522b15041..93b324948eba 100644
--- a/oap-server/server-starter/src/main/resources/metadata-service-mapping.yaml
+++ b/oap-server/server-starter/src/main/resources/metadata-service-mapping.yaml
@@ -13,5 +13,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-serviceName: ${LABELS."service.istio.io/canonical-name",LABELS."app.kubernetes.io/name",LABELS.app,LABELS.k8s-app}.${NAMESPACE}
+serviceName: ${LABELS."service.istio.io/canonical-name",LABELS."app.kubernetes.io/name",LABELS.app,LABELS.k8s-app,SERVICE}.${NAMESPACE}
 serviceInstanceName: ${NAME}


Test patch:
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/resources/metadata-service-mapping.yaml b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/resources/metadata-service-mapping.yaml
index acb4d430a3db..40461718dde4 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/resources/metadata-service-mapping.yaml
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/resources/metadata-service-mapping.yaml
@@ -13,5 +13,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-serviceName: ${LABELS."service.istio.io/canonical-revision"}.${LABELS."service.istio.io/canonical-name",LABELS."app.kubernetes.io/name",LABELS.app}.${NAMESPACE}
+serviceName: ${LABELS."service.istio.io/canonical-revision",REVISION}.${LABELS."service.istio.io/canonical-name",LABELS."app.kubernetes.io/name",LABELS.app,SERVICE}.${NAMESPACE}
 serviceInstanceName: ${NAME}
diff --git a/test/e2e-v2/cases/cilium/traffic-gen.yaml b/test/e2e-v2/cases/cilium/traffic-gen.yaml
index 4e3b05021672..590ff4d98fa9 100644
--- a/test/e2e-v2/cases/cilium/traffic-gen.yaml
+++ b/test/e2e-v2/cases/cilium/traffic-gen.yaml
@@ -33,8 +33,15 @@ spec:
     spec:
       containers:
         - name: trafficgenerator
-          image: elswork/wrk
-          command: ["wrk", "-t1", "-c1", "-d20m", "http://productpage:9080/productpage"]
+          image: alpine/curl:8.9.1
+          command: ["/bin/sh", "-c", "--"]
+          args:
+            - |
+              while true; do
+                curl -s -o /dev/null -w "%{http_code}\n" http://productpage:9080/productpage
+                sleep 1
+              done
           resources:
             requests:
-              cpu: 0.1
\ No newline at end of file
+              cpu: 0.1
+
diff --git a/test/e2e-v2/cases/istio/kind.k28.yaml b/test/e2e-v2/cases/istio/kind.k28.yaml
new file mode 100644
index 000000000000..6bb82dfce82e
--- /dev/null
+++ b/test/e2e-v2/cases/istio/kind.k28.yaml
@@ -0,0 +1,20 @@
+# 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.
+
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+nodes:
+  - role: control-plane
+    image: kindest/node:v1.28.15@sha256:a7c05c7ae043a0b8c818f5a06188bc2c4098f6cb59ca7d1856df00375d839251
diff --git a/test/e2e-v2/cases/istio/log4j2.xml b/test/e2e-v2/cases/istio/log4j2.xml
index 9e132a8d8bfe..2565d7f70a46 100644
--- a/test/e2e-v2/cases/istio/log4j2.xml
+++ b/test/e2e-v2/cases/istio/log4j2.xml
@@ -34,7 +34,7 @@
         <logger name="org.apache.skywalking.oap.server.storage.plugin.elasticsearch" level="DEBUG"/>
         <logger name="org.apache.skywalking.oap.server.core.storage.ttl" level="DEBUG"/>
         <logger name="org.apache.skywalking.library.elasticsearch" level="DEBUG"/>
-        <Root level="INFO">
+        <Root level="DEBUG">
             <AppenderRef ref="Console"/>
         </Root>
     </Loggers>
diff --git a/test/e2e-v2/cases/istio/traffic-gen.yaml b/test/e2e-v2/cases/istio/traffic-gen.yaml
index 549bcfde6595..0b365ce2eb07 100644
--- a/test/e2e-v2/cases/istio/traffic-gen.yaml
+++ b/test/e2e-v2/cases/istio/traffic-gen.yaml
@@ -33,8 +33,15 @@ spec:
     spec:
       containers:
         - name: trafficgenerator
-          image: williamyeh/wrk
-          command: ["wrk", "-t1", "-c1", "-d20m", "http://istio-ingressgateway.istio-system:80/productpage"]
+          image: alpine/curl:8.9.1
+          command: ["/bin/sh", "-c", "--"]
+          args:
+            - |
+              while true; do
+                curl -s -o /dev/null -w "%{http_code}\n" http://istio-ingressgateway.istio-system:80/productpage
+                sleep 1
+              done
           resources:
             requests:
-              cpu: 0.1
\ No newline at end of file
+              cpu: 0.1
+
diff --git a/test/e2e-v2/cases/istio/values.yaml b/test/e2e-v2/cases/istio/values.yaml
index 8b42ad4765f1..27426bbd75f0 100644
--- a/test/e2e-v2/cases/istio/values.yaml
+++ b/test/e2e-v2/cases/istio/values.yaml
@@ -16,7 +16,7 @@
 oap:
   config:
     metadata-service-mapping.yaml: |
-      serviceName: e2e::${LABELS."service.istio.io/canonical-name"}
+      serviceName: e2e::${LABELS."service.istio.io/canonical-name",SERVICE}
       serviceInstanceName: ${NAME}
 
 elasticsearch:
diff --git a/test/e2e-v2/cases/profiling/ebpf/access_log/traffic-gen.yaml b/test/e2e-v2/cases/profiling/ebpf/access_log/traffic-gen.yaml
index 4e3b05021672..a1d08fef93f1 100644
--- a/test/e2e-v2/cases/profiling/ebpf/access_log/traffic-gen.yaml
+++ b/test/e2e-v2/cases/profiling/ebpf/access_log/traffic-gen.yaml
@@ -33,8 +33,14 @@ spec:
     spec:
       containers:
         - name: trafficgenerator
-          image: elswork/wrk
-          command: ["wrk", "-t1", "-c1", "-d20m", "http://productpage:9080/productpage"]
+          image: alpine/curl:8.9.1
+          command: ["/bin/sh", "-c", "--"]
+          args:
+            - |
+              while true; do
+                curl -s -o /dev/null -w "%{http_code}\n" http://productpage:9080/productpage
+                sleep 1
+              done
           resources:
             requests:
-              cpu: 0.1
\ No newline at end of file
+              cpu: 0.1