Fixes #6484: Guard missing Dubbo ingress endpoints

shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/parser/DubboIngressParser.java

@@ -159,6 +159,9 @@ private List<DubboUpstream> getDefaultDubboRouteConfig(final V1IngressBackend de
             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 @@ private List<DubboUpstream> getDefaultDubboRouteConfig(final V1IngressBackend de
                     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 @@ private List<DubboUpstream> getDefaultDubboRouteConfig(final V1IngressBackend de
                     }
                 }
             }
+            if (dubboUpstreamList.isEmpty()) {
+                throw new EndpointsUnavailableException(namespace, serviceName);
+            }
         }
         return dubboUpstreamList;
     }
@@ -343,6 +349,9 @@ private List<DubboUpstream> parseUpstream(final V1HTTPIngressPath path, final St
                 && 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)
@@ -361,7 +370,7 @@ private List<DubboUpstream> parseUpstream(final V1HTTPIngressPath path, final St
                         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)
@@ -376,6 +385,9 @@ private List<DubboUpstream> parseUpstream(final V1HTTPIngressPath path, final St
                     }
                 }
             }
+            if (upstreamList.isEmpty()) {
+                throw new EndpointsUnavailableException(namespace, serviceName);
+            }
         }
         return upstreamList;
     }

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);
+    }
+}

shenyu-kubernetes-controller/src/main/java/org/apache/shenyu/k8s/reconciler/IngressReconciler.java

@@ -60,6 +60,7 @@
 import org.apache.shenyu.k8s.common.ServiceIngressRelation;
 import org.apache.shenyu.k8s.common.ShenyuMemoryConfig;
 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;
@@ -172,22 +173,21 @@ public Result reconcile(final Request request) {
             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);
@@ -439,10 +439,8 @@ private boolean needUpdate(final V1Ingress oldIngress, final V1Ingress currentIn
         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)) {

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 void testReconcile() {
         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<>();

shenyu-kubernetes-controller/src/test/java/org/apache/shenyu/k8s/parser/DubboIngressParserTest.java

@@ -25,9 +25,12 @@
 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 void testEmptyProtocolAnnotation() {
         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(