You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/03/11 09:51:11 UTC
(skywalking) branch fix/k8s-coordinator-self-endpoint-race created (now 3127f1df2e)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a change to branch fix/k8s-coordinator-self-endpoint-race
in repository https://gitbox.apache.org/repos/asf/skywalking.git
at 3127f1df2e Fix K8s coordinator self-endpoint race condition (#13739)
This branch includes the following new commits:
new 3127f1df2e Fix K8s coordinator self-endpoint race condition (#13739)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking) 01/01: Fix K8s coordinator self-endpoint race condition (#13739)
Posted by wu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch fix/k8s-coordinator-self-endpoint-race
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 3127f1df2e7f6392494a42f96d07784841fb925f
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Wed Mar 11 17:50:56 2026 +0800
Fix K8s coordinator self-endpoint race condition (#13739)
Include self pod in the DynamicEndpointGroup endpoint list so that
when the informer syncs self after initial startup, the list actually
changes and the listener re-fires with the correct self IP.
Previously self was excluded from the endpoint list, so DynamicEndpointGroup
deduplicated and never re-notified the listener, leaving self permanently
stuck at 127.0.0.1.
---
docs/en/changes/changes.md | 1 +
.../plugin/kubernetes/KubernetesCoordinator.java | 32 ++--
.../KubernetesLabelSelectorEndpointGroup.java | 10 +-
.../plugin/kubernetes/SelfEndpointAccessor.java | 29 +++
.../kubernetes/KubernetesCoordinatorTest.java | 210 +++++++++++++++++++++
5 files changed, 268 insertions(+), 14 deletions(-)
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 50ba551a53..a8465dc26a 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -95,6 +95,7 @@
#### OAP Server
* KubernetesCoordinator: make self instance return real pod IP address instead of `127.0.0.1`.
+* Fix KubernetesCoordinator self-endpoint race condition: include self in the endpoint list so DynamicEndpointGroup re-fires the listener when the self pod appears in the informer after initial sync.
* Enhance the alarm kernel with recovered status notification capability
* Fix BrowserWebVitalsPerfData `clsTime` to `cls` and make it double type.
* Init `log-mal-rules` at module provider start stage to avoid re-init for every LAL.
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java
index e2d9ec6b29..d3871deb61 100644
--- a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java
@@ -68,7 +68,7 @@ public class KubernetesCoordinator extends ClusterCoordinator {
}
}
- private EndpointGroup createEndpointGroup() {
+ EndpointGroup createEndpointGroup() {
if (port == -1) {
port = manager.find(CoreModule.NAME).provider().getService(ConfigService.class).getGRPCPort();
}
@@ -148,22 +148,28 @@ public class KubernetesCoordinator extends ClusterCoordinator {
log.debug("[kubernetes cluster endpoints]: {}", endpoints);
}
+ // The endpoint group now includes self in the list.
+ // Identify self by matching against getSelfEndpoint().
+ Endpoint selfEndpoint = null;
+ if (endpointGroup instanceof SelfEndpointAccessor) {
+ selfEndpoint = ((SelfEndpointAccessor) endpointGroup).getSelfEndpoint();
+ }
+
+ final Endpoint finalSelfEndpoint = selfEndpoint;
final var instances = endpoints.stream()
- .map(endpoint -> new RemoteInstance(new Address(endpoint.host(), endpoint.port(), false)))
+ .map(endpoint -> {
+ final boolean isSelf = finalSelfEndpoint != null
+ && endpoint.host().equals(finalSelfEndpoint.host())
+ && endpoint.port() == finalSelfEndpoint.port();
+ return new RemoteInstance(new Address(endpoint.host(), endpoint.port(), isSelf));
+ })
.collect(Collectors.toList());
- // The endpoint group will never include itself, add it.
- 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));
+ // If self was not found in the endpoint list (informer hasn't synced self yet),
+ // add a fallback self instance so the cluster always has a self node.
+ if (instances.stream().noneMatch(i -> i.getAddress().isSelf())) {
+ instances.add(new RemoteInstance(new Address("127.0.0.1", port, true)));
}
- instances.add(selfInstance);
if (log.isDebugEnabled()) {
instances.forEach(instance -> log.debug("kubernetes cluster instance: {}", instance));
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesLabelSelectorEndpointGroup.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesLabelSelectorEndpointGroup.java
index 63f6f9f411..5744bbd009 100644
--- a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesLabelSelectorEndpointGroup.java
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesLabelSelectorEndpointGroup.java
@@ -41,7 +41,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@Slf4j
-public class KubernetesLabelSelectorEndpointGroup extends DynamicEndpointGroup {
+public class KubernetesLabelSelectorEndpointGroup extends DynamicEndpointGroup implements SelfEndpointAccessor {
private final KubernetesClient kubernetesClient;
private final String namespace;
@@ -108,6 +108,14 @@ public class KubernetesLabelSelectorEndpointGroup extends DynamicEndpointGroup {
Endpoint endpoint = createEndpoint(pod);
if (endpoint != null) {
selfEndpoint = endpoint;
+ // Include self in the endpoint list so that when self pod appears
+ // in the informer (after initial sync), the endpoint list changes
+ // and DynamicEndpointGroup fires the listener again.
+ // Previously self was excluded here, so the endpoint list stayed
+ // the same when self appeared — DynamicEndpointGroup deduplicated
+ // and skipped listener notification, leaving the coordinator stuck
+ // with the 127.0.0.1 fallback forever. See #13739.
+ newEndpoints.add(endpoint);
}
continue;
}
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/SelfEndpointAccessor.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/SelfEndpointAccessor.java
new file mode 100644
index 0000000000..8f421d2a19
--- /dev/null
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/SelfEndpointAccessor.java
@@ -0,0 +1,29 @@
+/*
+ * 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.cluster.plugin.kubernetes;
+
+import com.linecorp.armeria.client.Endpoint;
+
+/**
+ * Provides access to the self endpoint for the Kubernetes coordinator.
+ * Implemented by {@link KubernetesLabelSelectorEndpointGroup} and test doubles.
+ */
+interface SelfEndpointAccessor {
+ Endpoint getSelfEndpoint();
+}
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinatorTest.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinatorTest.java
new file mode 100644
index 0000000000..3601c4b18e
--- /dev/null
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinatorTest.java
@@ -0,0 +1,210 @@
+/*
+ * 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.cluster.plugin.kubernetes;
+
+import com.linecorp.armeria.client.Endpoint;
+import com.linecorp.armeria.client.endpoint.DynamicEndpointGroup;
+import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
+import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics;
+import org.apache.skywalking.oap.server.testing.util.ReflectUtil;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
+
+/**
+ * Tests for {@link KubernetesCoordinator} listener behavior, especially the race condition
+ * where the K8s informer hasn't synced the self pod when the first endpoint update fires.
+ * See <a href="https://github.com/apache/skywalking/issues/13739">#13739</a>.
+ */
+@ExtendWith(MockitoExtension.class)
+public class KubernetesCoordinatorTest {
+
+ private static final int GRPC_PORT = 11800;
+ private static final String REMOTE_POD_IP = "10.116.2.203";
+ private static final String SELF_POD_IP = "10.116.2.100";
+
+ @Mock
+ private HealthCheckMetrics healthChecker;
+
+ private KubernetesCoordinator coordinator;
+ private TestEndpointGroup testEndpointGroup;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ final var config = new ClusterModuleKubernetesConfig();
+ config.setLabelSelector("app=oap");
+
+ testEndpointGroup = new TestEndpointGroup();
+
+ coordinator = spy(new KubernetesCoordinator(null, config));
+ ReflectUtil.setInternalState(coordinator, "port", GRPC_PORT);
+ ReflectUtil.setInternalState(coordinator, "healthChecker", healthChecker);
+ doReturn(testEndpointGroup).when(coordinator).createEndpointGroup();
+
+ coordinator.start();
+ }
+
+ /**
+ * Simulates the race condition: informer has NOT synced the self pod yet.
+ * Only the remote pod is in the endpoint list, and getSelfEndpoint() returns null.
+ * The coordinator should fall back to 127.0.0.1 for self.
+ */
+ @Test
+ public void shouldFallbackTo127WhenSelfNotSynced() {
+ // selfEndpoint is null (not synced), only remote pod in the list
+ testEndpointGroup.fireEndpoints(
+ List.of(Endpoint.of(REMOTE_POD_IP, GRPC_PORT)));
+
+ final List<RemoteInstance> instances = coordinator.queryRemoteNodes();
+ assertEquals(2, instances.size());
+
+ final RemoteInstance remote = findByHost(instances, REMOTE_POD_IP);
+ assertFalse(remote.getAddress().isSelf());
+
+ final RemoteInstance self = findByHost(instances, "127.0.0.1");
+ assertTrue(self.getAddress().isSelf());
+ }
+
+ /**
+ * Simulates the fix for #13739: after the informer syncs the self pod,
+ * the endpoint list now includes self (list changes from 1 to 2 entries),
+ * so DynamicEndpointGroup fires the listener again. This time
+ * getSelfEndpoint() returns the real IP and self is correctly identified.
+ *
+ * Previously, self was excluded from the endpoint list, so the list
+ * didn't change when self appeared — DynamicEndpointGroup deduplicated
+ * and never re-fired the listener, leaving self stuck at 127.0.0.1.
+ */
+ @Test
+ public void shouldResolveSelfAfterInformerSync() {
+ // Phase 1: informer hasn't synced self yet
+ testEndpointGroup.fireEndpoints(
+ List.of(Endpoint.of(REMOTE_POD_IP, GRPC_PORT)));
+
+ List<RemoteInstance> instances = coordinator.queryRemoteNodes();
+ assertEquals(2, instances.size());
+ assertTrue(findByHost(instances, "127.0.0.1").getAddress().isSelf(),
+ "Before sync, self should be 127.0.0.1 fallback");
+
+ // Phase 2: informer syncs self pod — endpoint list now includes both pods
+ final Endpoint selfEndpoint = Endpoint.of(SELF_POD_IP, GRPC_PORT);
+ testEndpointGroup.setSelfEndpoint(selfEndpoint);
+ testEndpointGroup.fireEndpoints(List.of(
+ Endpoint.of(REMOTE_POD_IP, GRPC_PORT),
+ selfEndpoint
+ ));
+
+ instances = coordinator.queryRemoteNodes();
+ assertEquals(2, instances.size());
+
+ final RemoteInstance self = findByHost(instances, SELF_POD_IP);
+ assertTrue(self.getAddress().isSelf(),
+ "After sync, self should use real pod IP " + SELF_POD_IP);
+
+ final RemoteInstance remote = findByHost(instances, REMOTE_POD_IP);
+ assertFalse(remote.getAddress().isSelf());
+
+ assertTrue(instances.stream().noneMatch(
+ i -> i.getAddress().getHost().equals("127.0.0.1")),
+ "127.0.0.1 fallback should be gone after self is resolved");
+ }
+
+ /**
+ * Verifies that TTL leader election works correctly after self is resolved.
+ * Sorted instance list should have deterministic ordering with real IPs,
+ * and exactly one node should be self.
+ */
+ @Test
+ public void shouldElectTTLLeaderCorrectlyWithRealIPs() {
+ final Endpoint selfEndpoint = Endpoint.of(SELF_POD_IP, GRPC_PORT);
+ testEndpointGroup.setSelfEndpoint(selfEndpoint);
+
+ testEndpointGroup.fireEndpoints(List.of(
+ Endpoint.of(REMOTE_POD_IP, GRPC_PORT),
+ selfEndpoint
+ ));
+
+ final List<RemoteInstance> instances = coordinator.queryRemoteNodes();
+ Collections.sort(instances);
+
+ assertEquals(2, instances.size());
+
+ final long selfCount = instances.stream()
+ .filter(i -> i.getAddress().isSelf())
+ .count();
+ assertEquals(1, selfCount, "Exactly one instance should be self");
+
+ final RemoteInstance selfInstance = instances.stream()
+ .filter(i -> i.getAddress().isSelf())
+ .findFirst()
+ .orElseThrow();
+ assertEquals(SELF_POD_IP, selfInstance.getAddress().getHost());
+ }
+
+ private RemoteInstance findByHost(List<RemoteInstance> instances, String host) {
+ return instances.stream()
+ .filter(i -> i.getAddress().getHost().equals(host))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError(
+ "No instance with host: " + host + ", found: " + instances));
+ }
+
+ /**
+ * A controllable endpoint group for testing. Implements {@link SelfEndpointAccessor}
+ * so the coordinator can resolve self endpoint without requiring a real K8s client.
+ *
+ * Simulates the K8s informer behavior:
+ * - Initially selfEndpoint is null (informer hasn't synced self pod yet)
+ * - After calling {@link #setSelfEndpoint}, the next {@link #fireEndpoints} call
+ * simulates the informer adding the self pod to its cache
+ */
+ static class TestEndpointGroup extends DynamicEndpointGroup implements SelfEndpointAccessor {
+ private volatile Endpoint selfEndpoint;
+
+ @Override
+ public Endpoint getSelfEndpoint() {
+ return selfEndpoint;
+ }
+
+ void setSelfEndpoint(Endpoint selfEndpoint) {
+ this.selfEndpoint = selfEndpoint;
+ }
+
+ void fireEndpoints(List<Endpoint> endpoints) {
+ setEndpoints(endpoints);
+ }
+
+ @Override
+ protected void doCloseAsync(CompletableFuture<?> future) {
+ future.complete(null);
+ }
+ }
+}