Merge pull request kroxylicious#2067 from robobario/proxy-reconciler-flip-out-on-other-referents-stale-status

KafkaProxyReconciler error on stale Referent status

Author: robobario

kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/KafkaProxyReconciler.java

@@ -312,16 +312,21 @@ private static InformerEventSource<KafkaService, KafkaProxy> buildKafkaServiceEv
 
     @VisibleForTesting
     static SecondaryToPrimaryMapper<KafkaService> kafkaServiceRefToProxyMapper(EventSourceContext<KafkaProxy> context) {
-        return kafkaServiceRef -> {
+        return kafkaService -> {
+            // we do not want to trigger reconciliation of any proxy if the ingress has not been reconciled
+            if (!ResourcesUtil.isStatusFresh(kafkaService)) {
+                LOGGER.debug("Ignoring event from KafkaService with stale status: {}", ResourcesUtil.toLocalRef(kafkaService));
+                return Set.of();
+            }
             // find all virtual clusters that reference this kafkaServiceRef
 
-            Set<? extends LocalRef<KafkaProxy>> proxyRefs = ResourcesUtil.resourcesInSameNamespace(context, kafkaServiceRef, VirtualKafkaCluster.class)
-                    .filter(vkc -> vkc.getSpec().getTargetKafkaServiceRef().equals(ResourcesUtil.toLocalRef(kafkaServiceRef)))
+            Set<? extends LocalRef<KafkaProxy>> proxyRefs = ResourcesUtil.resourcesInSameNamespace(context, kafkaService, VirtualKafkaCluster.class)
+                    .filter(vkc -> vkc.getSpec().getTargetKafkaServiceRef().equals(ResourcesUtil.toLocalRef(kafkaService)))
                     .map(VirtualKafkaCluster::getSpec)
                     .map(VirtualKafkaClusterSpec::getProxyRef)
                     .collect(Collectors.toSet());
 
-            Set<ResourceID> proxyIds = ResourcesUtil.filteredResourceIdsInSameNamespace(context, kafkaServiceRef, KafkaProxy.class,
+            Set<ResourceID> proxyIds = ResourcesUtil.filteredResourceIdsInSameNamespace(context, kafkaService, KafkaProxy.class,
                     proxy -> proxyRefs.contains(toLocalRef(proxy)));
             LOGGER.debug("Event source KafkaService SecondaryToPrimaryMapper got {}", proxyIds);
             return proxyIds;
@@ -404,6 +409,11 @@ static SecondaryToPrimaryMapper<VirtualKafkaCluster> clusterToProxyMapper(EventS
     @VisibleForTesting
     static SecondaryToPrimaryMapper<KafkaProxyIngress> ingressToProxyMapper(EventSourceContext<KafkaProxy> context) {
         return ingress -> {
+            // we do not want to trigger reconciliation of any proxy if the ingress has not been reconciled
+            if (!ResourcesUtil.isStatusFresh(ingress)) {
+                LOGGER.debug("Ignoring event from ingress with stale status: {}", ResourcesUtil.toLocalRef(ingress));
+                return Set.of();
+            }
             // we need to reconcile all proxies when a kafka proxy ingress changes in case the proxyRef is updated, we need to update
             // the previously referenced proxy too.
             Set<ResourceID> proxyIds = ResourcesUtil.filteredResourceIdsInSameNamespace(context, ingress, KafkaProxy.class, proxy -> true);
@@ -425,6 +435,11 @@ static PrimaryToSecondaryMapper<KafkaProxy> proxyToIngressMapper(EventSourceCont
     @VisibleForTesting
     static SecondaryToPrimaryMapper<KafkaProtocolFilter> filterToProxy(EventSourceContext<KafkaProxy> context) {
         return (KafkaProtocolFilter filter) -> {
+            // we do not want to trigger reconciliation of any proxy if the filter has not been reconciled
+            if (!ResourcesUtil.isStatusFresh(filter)) {
+                LOGGER.debug("Ignoring event from filter with stale status: {}", ResourcesUtil.toLocalRef(filter));
+                return Set.of();
+            }
             // filters don't point to a proxy, but must be in the same namespace as the proxy/proxies which reference the,
             // so when a filter changes we reconcile all the proxies in the same namespace
             Set<ResourceID> proxiesInFilterNamespace = ResourcesUtil.filteredResourceIdsInSameNamespace(context, filter, KafkaProxy.class, proxy -> true);

kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/ResourcesUtil.java

@@ -28,8 +28,14 @@
 
 import io.kroxylicious.kubernetes.api.common.AnyLocalRefBuilder;
 import io.kroxylicious.kubernetes.api.common.LocalRef;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngress;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngressStatus;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaService;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaServiceStatus;
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaClusterStatus;
+import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilter;
+import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilterStatus;
 
 public class ResourcesUtil {
 
@@ -280,6 +286,36 @@ public static boolean isStatusFresh(VirtualKafkaCluster cluster) {
         return isStatusFresh(cluster, c -> Optional.ofNullable(c.getStatus()).map(VirtualKafkaClusterStatus::getObservedGeneration).orElse(null));
     }
 
+    /**
+     * Checks that the status observedGeneration is equal to the metadata generation. Indicating
+     * that the current {@code spec} of the resource has been reconciled.
+     * @param ingress ingress
+     * @return true if status observedGeneration is equal to the metadata generation
+     */
+    public static boolean isStatusFresh(KafkaProxyIngress ingress) {
+        return isStatusFresh(ingress, i -> Optional.ofNullable(i.getStatus()).map(KafkaProxyIngressStatus::getObservedGeneration).orElse(null));
+    }
+
+    /**
+     * Checks that the status observedGeneration is equal to the metadata generation. Indicating
+     * that the current {@code spec} of the resource has been reconciled.
+     * @param service service
+     * @return true if status observedGeneration is equal to the metadata generation
+     */
+    public static boolean isStatusFresh(KafkaService service) {
+        return isStatusFresh(service, i -> Optional.ofNullable(i.getStatus()).map(KafkaServiceStatus::getObservedGeneration).orElse(null));
+    }
+
+    /**
+     * Checks that the status observedGeneration is equal to the metadata generation. Indicating
+     * that the current {@code spec} of the resource has been reconciled.
+     * @param filter filter
+     * @return true if status observedGeneration is equal to the metadata generation
+     */
+    public static boolean isStatusFresh(KafkaProtocolFilter filter) {
+        return isStatusFresh(filter, i -> Optional.ofNullable(i.getStatus()).map(KafkaProtocolFilterStatus::getObservedGeneration).orElse(null));
+    }
+
     private static <T extends HasMetadata> boolean isStatusFresh(T resource, Function<T, Long> observedGenerationFunc) {
         Long observedGeneration = observedGenerationFunc.apply(resource);
         Long generation = resource.getMetadata().getGeneration();

kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/resolver/DependencyResolver.java

@@ -142,7 +142,20 @@ private ClusterResolutionResult discoverProblemsAndBuildResolutionResult(Virtual
                                                                              Set<KafkaProxy> proxies) {
         Set<DanglingReference> danglingReferences = determineDanglingRefs(result, proxies, cluster);
         Set<LocalRef<?>> resolvedRefsFalse = determineResolvedRefsFalse(result);
-        return new ClusterResolutionResult(cluster, danglingReferences, resolvedRefsFalse, Set.of());
+        Set<LocalRef<?>> referentsWithStaleStatus = determineReferentsWithStaleStatus(result);
+        return new ClusterResolutionResult(cluster, danglingReferences, resolvedRefsFalse, referentsWithStaleStatus);
+    }
+
+    private static Set<LocalRef<?>> determineReferentsWithStaleStatus(CommonDependencies cluster) {
+        Stream<LocalRef<?>> ingressesWithStaleStatus = cluster.ingresses().entrySet().stream().filter(e -> !ResourcesUtil.isStatusFresh(e.getValue()))
+                .map(Map.Entry::getKey);
+        Stream<LocalRef<?>> servicesWithStaleStatus = cluster.kafkaServices().entrySet().stream().filter(e -> !ResourcesUtil.isStatusFresh(e.getValue()))
+                .map(Map.Entry::getKey);
+        Stream<LocalRef<?>> filtersWithStaleStatus = cluster.filters().entrySet().stream().filter(e -> !ResourcesUtil.isStatusFresh(e.getValue()))
+                .map(Map.Entry::getKey);
+        return Stream.of(ingressesWithStaleStatus, servicesWithStaleStatus, filtersWithStaleStatus)
+                .flatMap(Function.identity())
+                .collect(Collectors.toSet());
     }
 
     private static Set<LocalRef<?>> determineReferentsWithStaleStatus(VirtualKafkaCluster cluster) {