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 2dacf6de4bde..c16846955f6c 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 @@ private List 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 subsets = v1Endpoints.getSubsets(); if (Objects.isNull(subsets) || CollectionUtils.isEmpty(subsets)) { LOG.info("Endpoints {} do not have subsets", serviceName); @@ -171,7 +174,7 @@ private List 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 getDefaultDubboRouteConfig(final V1IngressBackend de } } } + if (dubboUpstreamList.isEmpty()) { + throw new EndpointsUnavailableException(namespace, serviceName); + } } return dubboUpstreamList; } @@ -343,6 +349,9 @@ private List 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 subsets = v1Endpoints.getSubsets(); String[] protocols = null; if (Objects.nonNull(annotations) && annotations.containsKey(IngressConstants.UPSTREAMS_PROTOCOL_ANNOTATION_KEY) @@ -361,7 +370,7 @@ private List 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 parseUpstream(final V1HTTPIngressPath path, final St } } } + 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 000000000000..455a1cb7ecca --- /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 bebba6fc947a..a0998e1ce5e1 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 @@ -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 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 shenyuMemoryConfigList = ingressParser.parse(ingressCopy, apiClient); - String pluginName = getPluginName(ingressCopy); + private void addNewIngressConfigToShenyu(final V1Ingress v1Ingress, final List 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 0507d71b6e40..50220f60ab7f 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 void testReconcile() { verify(shenyuCacheRepository).saveOrUpdateMetaData(any()); } + @Test + public void testIngressBeforeEndpointsRecoversOnRetry() { + Indexer 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 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 6907b28777f3..43b8c6baf4b2 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.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 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 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(