This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 01766dcf54 Fixes #6484: Guard missing Dubbo ingress endpoints (#7220)
01766dcf54 is described below
commit 01766dcf544d63169f41cbbf042b4fb24ed4dca3
Author: BobSong <[email protected]>
AuthorDate: Sat Oct 10 09:45:58 2026 +0800
Fixes #6484: Guard missing Dubbo ingress endpoints (#7220)
* fix(kubernetes-controller): guard missing Dubbo ingress endpoints
* fix(k8s): retry Dubbo ingress until endpoints are ready
---------
Co-authored-by: BobSong-dev <[email protected]>
---
.../shenyu/k8s/parser/DubboIngressParser.java | 16 ++++++-
.../k8s/parser/EndpointsUnavailableException.java | 36 ++++++++++++++++
.../shenyu/k8s/reconciler/IngressReconciler.java | 32 +++++++-------
.../org/apache/shenyu/k8s/DubboReconcilerTest.java | 38 ++++++++++++++++-
.../shenyu/k8s/parser/DubboIngressParserTest.java | 49 ++++++++++++++++++++++
5 files changed, 151 insertions(+), 20 deletions(-)
diff --git
a/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/parser/DubboIngressParser.java
b/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/parser/DubboIngressParser.java
index 7554451d88..b4afe5419b 100644
---
a/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/parser/DubboIngressParser.java
+++
b/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/parser/DubboIngressParser.java
@@ -159,6 +159,9 @@ public class DubboIngressParser implements
K8sResourceParser<V1Ingress> {
String serviceName = defaultBackend.getService().getName();
// shenyu routes directly to the container
V1Endpoints v1Endpoints =
endpointsLister.namespace(namespace).get(serviceName);
+ if (Objects.isNull(v1Endpoints)) {
+ throw new EndpointsUnavailableException(namespace,
serviceName);
+ }
List<V1EndpointSubset> subsets = v1Endpoints.getSubsets();
if (Objects.isNull(subsets) || CollectionUtils.isEmpty(subsets)) {
LOG.info("Endpoints {} do not have subsets", serviceName);
@@ -171,7 +174,7 @@ public class DubboIngressParser implements
K8sResourceParser<V1Ingress> {
for (V1EndpointAddress address : addresses) {
String upstreamIp = address.getIp();
String defaultPort =
parsePort(defaultBackend.getService());
- if (Objects.nonNull(defaultPort)) {
+ if (StringUtils.isNotBlank(upstreamIp) &&
Objects.nonNull(defaultPort)) {
DubboUpstream upstream = DubboUpstream.builder()
.upstreamUrl(upstreamIp + ":" +
defaultPort)
.weight(100)
@@ -185,6 +188,9 @@ public class DubboIngressParser implements
K8sResourceParser<V1Ingress> {
}
}
}
+ if (dubboUpstreamList.isEmpty()) {
+ throw new EndpointsUnavailableException(namespace,
serviceName);
+ }
}
return dubboUpstreamList;
}
@@ -346,6 +352,9 @@ public class DubboIngressParser implements
K8sResourceParser<V1Ingress> {
&& Objects.nonNull(path.getBackend().getService()) &&
Objects.nonNull(path.getBackend().getService().getName())) {
String serviceName = path.getBackend().getService().getName();
V1Endpoints v1Endpoints =
endpointsLister.namespace(namespace).get(serviceName);
+ if (Objects.isNull(v1Endpoints)) {
+ throw new EndpointsUnavailableException(namespace,
serviceName);
+ }
List<V1EndpointSubset> subsets = v1Endpoints.getSubsets();
String[] protocols = null;
if (Objects.nonNull(annotations) &&
annotations.containsKey(IngressConstants.UPSTREAMS_PROTOCOL_ANNOTATION_KEY)
@@ -364,7 +373,7 @@ public class DubboIngressParser implements
K8sResourceParser<V1Ingress> {
V1EndpointAddress address = addresses.get(i);
String upstreamIp = address.getIp();
String defaultPort =
parsePort(path.getBackend().getService());
- if (Objects.nonNull(defaultPort)) {
+ if (StringUtils.isNotBlank(upstreamIp) &&
Objects.nonNull(defaultPort)) {
String upstreamProtocol =
Objects.isNull(protocols) || i >= protocols.length ? "dubbo://" : protocols[i];
DubboUpstream upstream = DubboUpstream.builder()
.upstreamUrl(upstreamIp + ":" +
defaultPort)
@@ -379,6 +388,9 @@ public class DubboIngressParser implements
K8sResourceParser<V1Ingress> {
}
}
}
+ if (upstreamList.isEmpty()) {
+ throw new EndpointsUnavailableException(namespace,
serviceName);
+ }
}
return upstreamList;
}
diff --git
a/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/parser/EndpointsUnavailableException.java
b/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/parser/EndpointsUnavailableException.java
new file mode 100644
index 0000000000..455a1cb7ec
--- /dev/null
+++
b/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/parser/EndpointsUnavailableException.java
@@ -0,0 +1,36 @@
+/*
+ * 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.shenyu.k8s.parser;
+
+/**
+ * Indicates that an ingress must be retried when its backend becomes
available.
+ */
+public class EndpointsUnavailableException extends RuntimeException {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Create a retryable backend availability failure.
+ *
+ * @param namespace backend namespace
+ * @param serviceName backend service
+ */
+ public EndpointsUnavailableException(final String namespace, final String
serviceName) {
+ super("No ready Dubbo endpoints for " + namespace + "/" + serviceName);
+ }
+}
diff --git
a/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/reconciler/IngressReconciler.java
b/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/reconciler/IngressReconciler.java
index fb51ab847c..07b44eb4c8 100644
---
a/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/reconciler/IngressReconciler.java
+++
b/shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/reconciler/IngressReconciler.java
@@ -62,6 +62,7 @@ import org.apache.shenyu.k8s.common.ServiceIngressRelation;
import org.apache.shenyu.k8s.common.ShenyuMemoryConfig;
import org.apache.shenyu.k8s.parser.IngressPluginDefinition;
import org.apache.shenyu.k8s.parser.IngressParser;
+import org.apache.shenyu.k8s.parser.EndpointsUnavailableException;
import org.apache.shenyu.k8s.repository.ShenyuCacheRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -175,22 +176,21 @@ public class IngressReconciler implements Reconciler {
return new Result(false);
}
- if (Objects.isNull(oldIngress)) {
+ if (Objects.isNull(oldIngress) || needUpdate(oldIngress, v1Ingress)) {
try {
- addNewIngressConfigToShenyu(v1Ingress, new
CoreV1Api(apiClient));
- } catch (IOException e) {
- LOG.error("add new ingress config error", e);
- }
- } else if (needUpdate(oldIngress, v1Ingress)) {
- // Update logic
- // 1. clean old config
- doDeleteConfigByIngress(request, oldIngress);
-
- // 2. add new config
- try {
- addNewIngressConfigToShenyu(v1Ingress, new
CoreV1Api(apiClient));
+ // Resolve all backends before replacing the last successfully
applied configuration.
+ V1Ingress ingressCopy = new
V1IngressBuilder(v1Ingress).build();
+ List<ShenyuMemoryConfig> configs =
ingressParser.parse(ingressCopy, new CoreV1Api(apiClient));
+ if (Objects.nonNull(oldIngress)) {
+ doDeleteConfigByIngress(request, oldIngress);
+ }
+ addNewIngressConfigToShenyu(v1Ingress, configs);
+ } catch (EndpointsUnavailableException e) {
+ LOG.info("Retry ingress {}: {}", request, e.getMessage());
+ return new Result(true);
} catch (IOException e) {
LOG.error("add new ingress config error", e);
+ return new Result(true);
}
}
IngressCache.getInstance().put(request.getNamespace(),
request.getName(), v1Ingress);
@@ -469,10 +469,8 @@ public class IngressReconciler implements Reconciler {
return !oldIngress.equals(currentIngress);
}
- private void addNewIngressConfigToShenyu(final V1Ingress v1Ingress, final
CoreV1Api apiClient) throws IOException {
- V1Ingress ingressCopy = new V1IngressBuilder(v1Ingress).build();
- List<ShenyuMemoryConfig> shenyuMemoryConfigList =
ingressParser.parse(ingressCopy, apiClient);
- String pluginName = getPluginName(ingressCopy);
+ private void addNewIngressConfigToShenyu(final V1Ingress v1Ingress, final
List<ShenyuMemoryConfig> shenyuMemoryConfigList) throws IOException {
+ String pluginName = getPluginName(v1Ingress);
for (ShenyuMemoryConfig shenyuMemoryConfig : shenyuMemoryConfigList) {
if (Objects.nonNull(shenyuMemoryConfig)) {
diff --git
a/shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/DubboReconcilerTest.java
b/shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/DubboReconcilerTest.java
index 0507d71b6e..50220f60ab 100644
---
a/shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/DubboReconcilerTest.java
+++
b/shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/DubboReconcilerTest.java
@@ -82,7 +82,7 @@ public final class DubboReconcilerTest {
@BeforeEach
public void init() {
- IngressCache.getInstance().remove("mockedNamespace", "mockedIngress");
+ IngressCache.getInstance().remove(NAMESPACE, INGRESS_NAME);
ingressInformer = mock(SharedIndexInformer.class);
secretInformer = mock(SharedIndexInformer.class);
shenyuCacheRepository = mock(ShenyuCacheRepository.class);
@@ -163,6 +163,42 @@ public final class DubboReconcilerTest {
verify(shenyuCacheRepository).saveOrUpdateMetaData(any());
}
+ @Test
+ public void testIngressBeforeEndpointsRecoversOnRetry() {
+ Indexer<V1Endpoints> indexer = endpointsInformer.getIndexer();
+ final V1Endpoints ready = indexer.getByKey(NAMESPACE + "/" +
SERVICE_NAME);
+ when(indexer.getByKey(NAMESPACE + "/" +
SERVICE_NAME)).thenReturn(null);
+ Request request = new Request(NAMESPACE, INGRESS_NAME);
+
+ Assertions.assertEquals(new Result(true),
ingressReconciler.reconcile(request));
+ org.mockito.Mockito.verify(shenyuCacheRepository,
org.mockito.Mockito.never()).saveOrUpdateSelectorData(any());
+ Assertions.assertNull(IngressCache.getInstance().get(NAMESPACE,
INGRESS_NAME));
+
+ when(indexer.getByKey(NAMESPACE + "/" +
SERVICE_NAME)).thenReturn(ready);
+ Assertions.assertEquals(new Result(false),
ingressReconciler.reconcile(request));
+
verify(shenyuCacheRepository).saveOrUpdateSelectorData(org.mockito.ArgumentMatchers.argThat(selector
->
+ selector.getHandle().contains("127.0.0.1:20888")));
+ verify(shenyuCacheRepository).saveOrUpdateRuleData(any());
+ verify(shenyuCacheRepository).saveOrUpdateMetaData(any());
+ }
+
+ @Test
+ public void testUnavailableUpdatePreservesAppliedIngress() {
+ Request request = new Request(NAMESPACE, INGRESS_NAME);
+ Assertions.assertEquals(new Result(false),
ingressReconciler.reconcile(request));
+ final V1Ingress applied = IngressCache.getInstance().get(NAMESPACE,
INGRESS_NAME);
+ V1Ingress updated = new V1IngressBuilder(applied).build();
+ updated.getMetadata().setResourceVersion("2");
+ when(ingressInformer.getIndexer().getByKey(NAMESPACE + "/" +
INGRESS_NAME)).thenReturn(updated);
+ when(endpointsInformer.getIndexer().getByKey(NAMESPACE + "/" +
SERVICE_NAME)).thenReturn(null);
+ org.mockito.Mockito.clearInvocations(shenyuCacheRepository);
+
+ Assertions.assertEquals(new Result(true),
ingressReconciler.reconcile(request));
+ Assertions.assertSame(applied,
IngressCache.getInstance().get(NAMESPACE, INGRESS_NAME));
+ org.mockito.Mockito.verify(shenyuCacheRepository,
org.mockito.Mockito.never()).deleteSelectorData(any(), any());
+ org.mockito.Mockito.verify(shenyuCacheRepository,
org.mockito.Mockito.never()).saveOrUpdateSelectorData(any());
+ }
+
@Test
public void testParseSkipsContextPathWhenDubboEnabled() {
Map<String, String> annotations = new HashMap<>();
diff --git
a/shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/parser/DubboIngressParserTest.java
b/shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/parser/DubboIngressParserTest.java
index 6907b28777..43b8c6baf4 100644
---
a/shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/parser/DubboIngressParserTest.java
+++
b/shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/parser/DubboIngressParserTest.java
@@ -25,9 +25,12 @@ import io.kubernetes.client.openapi.models.V1Endpoints;
import io.kubernetes.client.openapi.models.V1EndpointsBuilder;
import io.kubernetes.client.openapi.models.V1HTTPIngressPathBuilder;
import io.kubernetes.client.openapi.models.V1Ingress;
+import io.kubernetes.client.openapi.models.V1IngressBackend;
import io.kubernetes.client.openapi.models.V1IngressBuilder;
import io.kubernetes.client.openapi.models.V1IngressRuleBuilder;
+import io.kubernetes.client.openapi.models.V1IngressServiceBackend;
import io.kubernetes.client.openapi.models.V1Service;
+import io.kubernetes.client.openapi.models.V1ServiceBackendPort;
import org.apache.shenyu.common.dto.convert.selector.DubboUpstream;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.k8s.common.IngressConstants;
@@ -160,6 +163,52 @@ public class DubboIngressParserTest {
assertNotNull(upstreams);
}
+ @Test
+ public void shouldRetryPathWhenEndpointsAreMissing() {
+ Indexer<V1Endpoints> missingEndpointsIndexer = mock(Indexer.class);
+ DubboIngressParser parser = new DubboIngressParser(serviceLister, new
Lister<>(missingEndpointsIndexer));
+
+ Assertions.assertThrows(EndpointsUnavailableException.class, () ->
parser.parse(
+ createIngress(null, Collections.emptyMap(), true), null));
+ }
+
+ @Test
+ public void shouldRetryDefaultBackendWhenEndpointsAreMissing() {
+ Indexer<V1Endpoints> missingEndpointsIndexer = mock(Indexer.class);
+ DubboIngressParser parser = new DubboIngressParser(serviceLister, new
Lister<>(missingEndpointsIndexer));
+ V1Ingress ingress = new
V1IngressBuilder().withNewMetadata().withName("test-ingress").withNamespace(NAMESPACE)
+
.withAnnotations(Collections.emptyMap()).withLabels(Collections.emptyMap()).endMetadata()
+ .withNewSpec().withDefaultBackend(new V1IngressBackend()
+ .service(new
V1IngressServiceBackend().name(SERVICE_NAME)
+ .port(new
V1ServiceBackendPort().number(8080))))
+ .endSpec().build();
+
+ Assertions.assertThrows(EndpointsUnavailableException.class, () ->
parser.parse(ingress, null));
+ }
+
+ @Test
+ public void shouldNotInheritDefaultWhenPathEndpointsAreUnavailable() {
+ V1Ingress ingress = createIngress(null, Collections.emptyMap(), true);
+ ingress.getSpec().setDefaultBackend(new V1IngressBackend().service(new
V1IngressServiceBackend()
+ .name("ready-default").port(new
V1ServiceBackendPort().number(8080))));
+ when(endpointsIndexer.getByKey(NAMESPACE +
"/ready-default")).thenReturn(new V1EndpointsBuilder()
+ .withSubsets(new V1EndpointSubsetBuilder().withAddresses(new
V1EndpointAddress().ip("127.0.0.2")).build()).build());
+ when(endpointsIndexer.getByKey(NAMESPACE + "/" +
SERVICE_NAME)).thenReturn(null);
+ DubboIngressParser parser = new DubboIngressParser(serviceLister,
endpointsLister);
+ Assertions.assertThrows(EndpointsUnavailableException.class, () ->
parser.parse(ingress, null));
+ }
+
+ @Test
+ public void shouldRetryUntilEndpointsHaveReadyAddresses() {
+ V1Ingress ingress = createIngress(null, Collections.emptyMap(), true);
+ DubboIngressParser parser = new DubboIngressParser(serviceLister,
endpointsLister);
+ when(endpointsIndexer.getByKey(NAMESPACE + "/" +
SERVICE_NAME)).thenReturn(new V1Endpoints());
+ Assertions.assertThrows(EndpointsUnavailableException.class, () ->
parser.parse(ingress, null));
+ when(endpointsIndexer.getByKey(NAMESPACE + "/" +
SERVICE_NAME)).thenReturn(new V1EndpointsBuilder()
+ .withSubsets(new
V1EndpointSubsetBuilder().withNotReadyAddresses(new
V1EndpointAddress().ip("127.0.0.1")).build()).build());
+ Assertions.assertThrows(EndpointsUnavailableException.class, () ->
parser.parse(ingress, null));
+ }
+
@Test
public void shouldParseIngressWhenAnnotationsAreNull() {
ShenyuMemoryConfig config = assertDoesNotThrow(() ->
createParser().parse(
