Adapt the new metadata standardization in Istio 1.24 (#12768)

Author: kezhenxu94

.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

docker/oap/log4j2.xml

@@ -28,7 +28,7 @@
         </Console>
     </Appenders>
     <Loggers>
-        <Root level="INFO">
+        <Root level="DEBUG">
             <AppenderRef ref="Console"/>
         </Root>
     </Loggers>

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
 

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;

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,13 +96,34 @@ 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();
         }
 
         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());
     }

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,17 +115,27 @@ 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()));
         }
         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);
         }