Merge pull request kroxylicious#1924 from robobario/refactor-early-resolution

operator: report unresolved ingresses

Author: robobario

kroxylicious-operator/pom.xml

@@ -216,6 +216,11 @@
             <artifactId>kubernetes-model-apiextensions</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

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

@@ -11,7 +11,7 @@
 import java.util.Set;
 
 import io.kroxylicious.kubernetes.api.v1alpha1.virtualkafkaclusterspec.TargetCluster;
-import io.kroxylicious.kubernetes.operator.ingress.IngressConflictException;
+import io.kroxylicious.kubernetes.operator.model.ingress.IngressConflictException;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
@@ -36,11 +36,16 @@ static ClusterCondition filterInvalid(String cluster, String filterName, String
                 String.format("Filter \"%s\" is invalid: %s", filterName, detail));
     }
 
-    static ClusterCondition filterNotFound(String cluster, String filterName) {
+    public static ClusterCondition filterNotFound(String cluster, String filterName) {
         return new ClusterCondition(cluster, ConditionType.Accepted, Status.FALSE, INVALID,
                 String.format("Filter \"%s\" does not exist.", filterName));
     }
 
+    public static ClusterCondition ingressNotFound(String cluster, String ingressName) {
+        return new ClusterCondition(cluster, ConditionType.Accepted, Status.FALSE, INVALID,
+                String.format("KafkaProxyIngress \"%s\" does not exist.", ingressName));
+    }
+
     public static ClusterCondition ingressConflict(String cluster, Set<IngressConflictException> ingressConflictExceptions) {
         String ingresses = ingressConflictExceptions.stream()
                 .map(IngressConflictException::getIngressName)
@@ -49,7 +54,7 @@ public static ClusterCondition ingressConflict(String cluster, Set<IngressConfli
                 String.format("Ingress(es) [%s] of cluster conflicts with another ingress", ingresses));
     }
 
-    static ClusterCondition targetClusterRefNotFound(String cluster, TargetCluster targetCluster) {
+    public static ClusterCondition targetClusterRefNotFound(String cluster, TargetCluster targetCluster) {
         return new ClusterCondition(cluster, ConditionType.Accepted, Status.FALSE, INVALID,
                 String.format("Target Cluster \"%s\" does not exist.", kubeName(targetCluster).orElse("<unknown>")));
     }

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

@@ -5,7 +5,6 @@
  */
 package io.kroxylicious.kubernetes.operator;
 
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -18,10 +17,10 @@
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
 
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
-import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngress;
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
-import io.kroxylicious.kubernetes.operator.ingress.IngressAllocator;
-import io.kroxylicious.kubernetes.operator.ingress.ProxyIngressModel;
+import io.kroxylicious.kubernetes.operator.model.ProxyModel;
+import io.kroxylicious.kubernetes.operator.model.ProxyModelBuilder;
+import io.kroxylicious.kubernetes.operator.model.ingress.ProxyIngressModel;
 
 import static io.kroxylicious.kubernetes.operator.ResourcesUtil.toByNameMap;
 
@@ -51,12 +50,11 @@ static String serviceName(VirtualKafkaCluster cluster) {
     public Map<String, Service> desiredResources(
                                                  KafkaProxy primary,
                                                  Context<KafkaProxy> context) {
-        List<VirtualKafkaCluster> clusters = ResourcesUtil.clustersInNameOrder(context).toList();
-        Set<KafkaProxyIngress> ingresses = context.getSecondaryResources(KafkaProxyIngress.class);
-        ProxyIngressModel ingressModel = IngressAllocator.allocateProxyIngressModel(primary, clusters, ingresses);
-        Stream<Service> serviceStream = clusters.stream()
+        ProxyModelBuilder proxyModelBuilder = ProxyModelBuilder.contextBuilder(context);
+        ProxyModel model = proxyModelBuilder.build(primary, context);
+        Stream<Service> serviceStream = model.clustersWithValidIngresses().stream()
                 .filter(cluster -> !SharedKafkaProxyContext.isBroken(context, cluster))
-                .flatMap(cluster -> ingressModel.clusterIngressModel(cluster).map(ProxyIngressModel.VirtualClusterIngressModel::services).orElse(Stream.empty()));
+                .flatMap(cluster -> model.ingressModel().clusterIngressModel(cluster).map(ProxyIngressModel.VirtualClusterIngressModel::services).orElse(Stream.empty()));
         return serviceStream.collect(toByNameMap());
     }
 

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

@@ -8,38 +8,35 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
-import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
 import io.fabric8.kubernetes.api.model.Volume;
 import io.fabric8.kubernetes.api.model.VolumeMount;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.CRUDKubernetesDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaClusterRef;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
-import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngress;
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
-import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaClusterSpec;
 import io.kroxylicious.kubernetes.api.v1alpha1.virtualkafkaclusterspec.Filters;
-import io.kroxylicious.kubernetes.operator.ingress.IngressAllocator;
-import io.kroxylicious.kubernetes.operator.ingress.IngressConflictException;
-import io.kroxylicious.kubernetes.operator.ingress.ProxyIngressModel;
+import io.kroxylicious.kubernetes.operator.model.ProxyModel;
+import io.kroxylicious.kubernetes.operator.model.ProxyModelBuilder;
+import io.kroxylicious.kubernetes.operator.model.ingress.ProxyIngressModel;
+import io.kroxylicious.kubernetes.operator.resolver.ResolutionResult;
 import io.kroxylicious.proxy.config.ConfigParser;
 import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.config.NamedFilterDefinition;
@@ -52,7 +49,6 @@
 import edu.umd.cs.findbugs.annotations.NonNull;
 
 import static io.kroxylicious.kubernetes.operator.Labels.standardLabels;
-import static io.kroxylicious.kubernetes.operator.ResourcesUtil.clusterRefs;
 import static io.kroxylicious.kubernetes.operator.ResourcesUtil.namespace;
 
 /**
@@ -132,24 +128,19 @@ protected ConfigMap desired(KafkaProxy primary,
 
     String generateProxyConfig(KafkaProxy primary,
                                Context<KafkaProxy> context) {
+        ProxyModelBuilder proxyModelBuilder = ProxyModelBuilder.contextBuilder(context);
+        ProxyModel model = proxyModelBuilder.build(primary, context);
+        List<NamedFilterDefinition> allFilterDefinitions = buildFilterDefinitions(context, model);
+        Map<String, NamedFilterDefinition> namedDefinitions = allFilterDefinitions.stream().collect(Collectors.toMap(NamedFilterDefinition::name, f -> f));
 
-        List<VirtualKafkaCluster> virtualKafkaClusters = ResourcesUtil.clustersInNameOrder(context).toList();
-        Set<KafkaProxyIngress> ingresses = context.getSecondaryResources(KafkaProxyIngress.class);
-        ProxyIngressModel ingressModel = getProxyIngressModel(primary, context, virtualKafkaClusters, ingresses);
+        var virtualClusters = buildVirtualClusters(namedDefinitions.keySet(), model);
 
-        Map<ResourceID, KafkaClusterRef> clusterRefs = clusterRefs(context);
-
-        // TODO fix this double invocation of buildFilterDefinitions which is a workaround for https://github.com/kroxylicious/kroxylicious/issues/1916
-        // first invocation rejects some virtual clusters with unresolved refs
-        buildFilterDefinitions(context, virtualKafkaClusters);
-
-        var virtualClusters = buildVirtualClusters(context, virtualKafkaClusters, clusterRefs, ingressModel);
-
-        // second invocation excludes filters that belong to clusters broken in `buildVirtualClusters`
-        List<NamedFilterDefinition> filterDefinitions = buildFilterDefinitions(context, virtualKafkaClusters);
+        List<NamedFilterDefinition> referencedFilters = virtualClusters.stream().flatMap(c -> Optional.ofNullable(c.filters()).stream().flatMap(Collection::stream))
+                .distinct()
+                .map(namedDefinitions::get).toList();
 
         Configuration configuration = new Configuration(
-                new ManagementConfiguration(null, null, new EndpointsConfiguration(new PrometheusMetricsConfig())), filterDefinitions,
+                new ManagementConfiguration(null, null, new EndpointsConfiguration(new PrometheusMetricsConfig())), referencedFilters,
                 null, // no defaultFilters <= each of the virtualClusters specifies its own
                 virtualClusters,
                 List.of(), false,
@@ -159,60 +150,31 @@ String generateProxyConfig(KafkaProxy primary,
         return toYaml(configuration);
     }
 
-    private static @NonNull ProxyIngressModel getProxyIngressModel(KafkaProxy primary, Context<KafkaProxy> context, List<VirtualKafkaCluster> virtualKafkaClusters,
-                                                                   Set<KafkaProxyIngress> ingresses) {
-        ProxyIngressModel ingressModel = IngressAllocator.allocateProxyIngressModel(primary, virtualKafkaClusters, ingresses);
-        for (ProxyIngressModel.VirtualClusterIngressModel virtualClusterIngressModel : ingressModel.clusters()) {
-            Set<IngressConflictException> exceptions = virtualClusterIngressModel.ingressExceptions();
-            if (!exceptions.isEmpty()) {
-                VirtualKafkaCluster cluster = virtualClusterIngressModel.cluster();
-                SharedKafkaProxyContext.addClusterCondition(context, cluster, ClusterCondition.ingressConflict(ResourcesUtil.name(cluster), exceptions));
-            }
-        }
-        return ingressModel;
-    }
-
     @NonNull
-    private static List<VirtualCluster> buildVirtualClusters(Context<KafkaProxy> context,
-                                                             List<VirtualKafkaCluster> clusters,
-                                                             Map<ResourceID, KafkaClusterRef> clusterRefs,
-                                                             ProxyIngressModel ingressModel) {
-        Map<VirtualKafkaCluster, Optional<KafkaClusterRef>> clusterToRefMap = clusters.stream()
-                .collect(Collectors.toMap(
-                        Function.identity(),
-                        cluster -> clusterTargetClusterResourceID(cluster).map(clusterRefs::get)));
-
-        clusterToRefMap.entrySet().stream()
-                .filter(e -> e.getValue().isEmpty())
-                .forEach(e -> {
-                    var cluster = e.getKey();
-                    SharedKafkaProxyContext.addClusterCondition(context, cluster, targetClusterResourceNotFound(cluster).accepted());
-                });
-
-        return clusters.stream()
-                .filter(cluster -> !SharedKafkaProxyContext.isBroken(context, cluster))
-                .map(cluster -> getVirtualCluster(cluster, clusterToRefMap.get(cluster).get(), ingressModel))
+    private static List<VirtualCluster> buildVirtualClusters(Set<String> successfullyBuiltFilterNames, ProxyModel model) {
+        return model.clustersWithValidIngresses().stream()
+                .filter(cluster -> Optional.ofNullable(cluster.getSpec().getFilters()).stream().flatMap(Collection::stream).allMatch(
+                        filters -> successfullyBuiltFilterNames.contains(filterDefinitionName(filters))))
+                .map(cluster -> getVirtualCluster(cluster, model.resolutionResult().kafkaClusterRef(cluster).orElseThrow(), model.ingressModel()))
                 .toList();
     }
 
     @NonNull
 
-    private List<NamedFilterDefinition> buildFilterDefinitions(Context<KafkaProxy> context, List<VirtualKafkaCluster> clusters) {
+    private List<NamedFilterDefinition> buildFilterDefinitions(Context<KafkaProxy> context,
+                                                               ProxyModel model) {
         List<NamedFilterDefinition> filterDefinitions = new ArrayList<>();
         Set<NamedFilterDefinition> uniqueValues = new HashSet<>();
-        for (VirtualKafkaCluster cluster1 : clusters) {
-            if (SharedKafkaProxyContext.isBroken(context, cluster1)) {
-                continue;
-            }
+        for (VirtualKafkaCluster cluster : model.clustersWithValidIngresses()) {
             try {
-                for (NamedFilterDefinition namedFilterDefinition : filterDefinitions(context, cluster1)) {
+                for (NamedFilterDefinition namedFilterDefinition : filterDefinitions(context, cluster, model.resolutionResult())) {
                     if (uniqueValues.add(namedFilterDefinition)) {
                         filterDefinitions.add(namedFilterDefinition);
                     }
                 }
             }
             catch (InvalidClusterException e) {
-                SharedKafkaProxyContext.addClusterCondition(context, cluster1, e.accepted());
+                SharedKafkaProxyContext.addClusterCondition(context, cluster, e.accepted());
             }
         }
         filterDefinitions.sort(Comparator.comparing(NamedFilterDefinition::name));
@@ -233,14 +195,14 @@ private static String filterDefinitionName(Filters filterCrRef) {
     }
 
     @NonNull
-    private List<NamedFilterDefinition> filterDefinitions(Context<KafkaProxy> context, VirtualKafkaCluster cluster)
+    private List<NamedFilterDefinition> filterDefinitions(Context<KafkaProxy> context, VirtualKafkaCluster cluster, ResolutionResult resolutionResult)
             throws InvalidClusterException {
 
         return Optional.ofNullable(cluster.getSpec().getFilters()).orElse(List.of()).stream().map(filterCrRef -> {
 
             String filterDefinitionName = filterDefinitionName(filterCrRef);
 
-            var filterCr = filterResourceFromRef(cluster, context, filterCrRef);
+            var filterCr = resolutionResult.filter(filterCrRef).orElseThrow();
             if (filterCr.getAdditionalProperties().get("spec") instanceof Map<?, ?> spec) {
                 String type = (String) spec.get("type");
                 SecureConfigInterpolator.InterpolationResult interpolationResult = interpolateConfig(spec);
@@ -266,7 +228,7 @@ private static <T> void putOrMerged(ManagedWorkflowAndDependentResourceContext c
         }
     }
 
-    private @NonNull SecureConfigInterpolator.InterpolationResult interpolateConfig(Map<?, ?> spec) {
+    private SecureConfigInterpolator.InterpolationResult interpolateConfig(Map<?, ?> spec) {
         SecureConfigInterpolator.InterpolationResult result;
         Object configTemplate = spec.get("configTemplate");
         if (configTemplate != null) {
@@ -278,34 +240,6 @@ private static <T> void putOrMerged(ManagedWorkflowAndDependentResourceContext c
         return result;
     }
 
-    @NonNull
-    private static InvalidClusterException filterResourceNotFound(VirtualKafkaCluster cluster, Filters filterRef) {
-        return new InvalidClusterException(ClusterCondition.filterNotFound(ResourcesUtil.name(cluster), filterRef.getName()));
-    }
-
-    @NonNull
-    private static InvalidClusterException targetClusterResourceNotFound(VirtualKafkaCluster cluster) {
-        return new InvalidClusterException(ClusterCondition.targetClusterRefNotFound(ResourcesUtil.name(cluster), cluster.getSpec().getTargetCluster()));
-    }
-
-    /**
-     * Look up a Filter CR from the group, kind and name given in the cluster.
-     */
-    @NonNull
-    private static GenericKubernetesResource filterResourceFromRef(VirtualKafkaCluster cluster, Context<KafkaProxy> context, Filters filterRef)
-            throws InvalidClusterException {
-        return context.getSecondaryResources(GenericKubernetesResource.class).stream()
-                .filter(filterResource -> {
-                    String apiVersion = filterResource.getApiVersion();
-                    var filterResourceGroup = apiVersion.substring(0, apiVersion.indexOf("/"));
-                    return filterResourceGroup.equals(filterRef.getGroup())
-                            && filterResource.getKind().equals(filterRef.getKind())
-                            && ResourcesUtil.name(filterResource).equals(filterRef.getName());
-                })
-                .findFirst()
-                .orElseThrow(() -> filterResourceNotFound(cluster, filterRef));
-    }
-
     private static VirtualCluster getVirtualCluster(VirtualKafkaCluster cluster,
                                                     KafkaClusterRef kafkaClusterRef,
                                                     ProxyIngressModel ingressModel) {
@@ -321,10 +255,4 @@ private static VirtualCluster getVirtualCluster(VirtualKafkaCluster cluster,
                 filterNamesForCluster(cluster));
     }
 
-    private static Optional<ResourceID> clusterTargetClusterResourceID(VirtualKafkaCluster cluster) {
-        return Optional.ofNullable(cluster.getSpec())
-                .map(VirtualKafkaClusterSpec::getTargetCluster)
-                .map(io.kroxylicious.kubernetes.api.v1alpha1.virtualkafkaclusterspec.TargetCluster::getClusterRef)
-                .map(r -> new ResourceID(r.getName(), namespace(cluster)));
-    }
 }