KubernetesCoordinator: make self instance return real pod IP address instead of 127.0.0.1. (#13577)

Author: wankai123

docs/en/changes/changes.md

@@ -4,6 +4,8 @@
 
 #### OAP Server
 
+* KubernetesCoordinator: make self instance return real pod IP address instead of `127.0.0.1`.
+
 #### UI
 
 #### Documentation

oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.cluster.plugin.kubernetes;
 
+import com.linecorp.armeria.client.Endpoint;
 import io.fabric8.kubernetes.client.KubernetesClientBuilder;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.oap.server.core.CoreModule;
@@ -152,7 +153,16 @@ public void start() {
                     .collect(Collectors.toList());
 
             // The endpoint group will never include itself, add it.
-            final var selfInstance = new RemoteInstance(new Address("127.0.0.1", port, true));
+            Endpoint selfEndpoint = null;
+            if (endpointGroup instanceof KubernetesLabelSelectorEndpointGroup) {
+                selfEndpoint = ((KubernetesLabelSelectorEndpointGroup) endpointGroup).getSelfEndpoint();
+            }
+            final RemoteInstance selfInstance;
+            if (selfEndpoint == null) {
+                selfInstance = new RemoteInstance(new Address("127.0.0.1", port, true));
+            } else {
+                selfInstance = new RemoteInstance(new Address(selfEndpoint.host(), selfEndpoint.port(), true));
+            }
             instances.add(selfInstance);
 
             if (log.isDebugEnabled()) {

oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesLabelSelectorEndpointGroup.java

@@ -28,15 +28,17 @@
 import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
 import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
 import io.fabric8.kubernetes.client.informers.cache.Lister;
+import java.util.ArrayList;
+import java.util.List;
 import lombok.Data;
+import lombok.Getter;
 import lombok.experimental.Accessors;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
 
 @Slf4j
 public class KubernetesLabelSelectorEndpointGroup extends DynamicEndpointGroup {
@@ -48,6 +50,8 @@ public class KubernetesLabelSelectorEndpointGroup extends DynamicEndpointGroup {
     private final String portName;
     private final SharedIndexInformer<Pod> podInformer;
     private final String selfUid;
+    @Getter
+    private volatile Endpoint selfEndpoint;
 
     private KubernetesLabelSelectorEndpointGroup(Builder builder) {
         super(builder.selectionStrategy);
@@ -92,14 +96,26 @@ private void updateEndpoints() {
             }
             final var podLister = new Lister<>(podInformer.getIndexer());
             final var pods = podLister.namespace(namespace).list();
-
-            final var newEndpoints = pods.stream()
-                    .filter(this::isPodReady)
-                    .filter(pod -> StringUtil.isNotBlank(pod.getStatus().getPodIP()))
-                    .filter(pod -> !pod.getMetadata().getUid().equals(selfUid))
-                    .map(this::createEndpoint)
-                    .filter(endpoint -> endpoint != null)
-                    .collect(Collectors.toList());
+            final List<Endpoint> newEndpoints = new ArrayList<>();
+            for (Pod pod : pods) {
+                if (!isPodReady(pod)) {
+                    continue;
+                }
+                if (StringUtil.isBlank(pod.getStatus().getPodIP())) {
+                    continue;
+                }
+                if (pod.getMetadata().getUid().equals(selfUid)) {
+                    Endpoint endpoint = createEndpoint(pod);
+                    if (endpoint != null) {
+                        selfEndpoint = endpoint;
+                    }
+                    continue;
+                }
+                Endpoint endpoint = createEndpoint(pod);
+                if (endpoint != null) {
+                    newEndpoints.add(endpoint);
+                }
+            }
 
             log.debug("Updating endpoints to: {}", newEndpoints);
             setEndpoints(newEndpoints);