Rework error reporting (kroxylicious#2007)

* Remove the `KafkaProxy.status.clusters` conditions, using the `VirtualKafkaCluster.status.conditions`
* Use `KafkaProxyContext` to manage the items in the context.managedWorkflowAndDependentResourceContext() (removing `SharedKafkaProxyContext` for more consistent naming).
  Initialize this in `KafkaProxyReconciler#initContext()` so that all the dependent resources see the same sets of referenced resources.
* `ProxyConfigConfigMap` now writes a list of conditions for each cluster to the Proxy CM
* VirtualKafkaClusterReconciler now has the proxy ConfigMap as a secondary resource, and uses the list of conditions written by the `ProxyConfigConfigMap` (during the KafkaProxyReconciliation) to update the `VirtualKafkaCluster.status.conditions` for one cluster.

Also some other changes
* Have `KafkaProxyReconciler`'s constructor accept a `Clock`, like the other reconcilers
* Factor out `ProxyConfigData` for encapsulating the contents of the Proxy ConfigMap's data section.
* Alloc ResourceUtil's `maybeAddOrUpdateCondition` to take a list of conditions, rather than just one.
* ClusterCondition and InvalidClusterException gone, plus related tests.


Signed-off-by: Tom Bentley <[email protected]>

Author: tombentley

kroxylicious-operator/pom.xml

@@ -126,6 +126,10 @@
             <artifactId>jackson-databind</artifactId>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.slf4j</groupId>

kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/api/common/Condition.java

@@ -10,7 +10,7 @@
  * A common Condition type, used in CR statuses.
  */
 @com.fasterxml.jackson.annotation.JsonInclude(com.fasterxml.jackson.annotation.JsonInclude.Include.NON_NULL)
-@com.fasterxml.jackson.annotation.JsonPropertyOrder({ "lastTransitionTime", "message", "observedGeneration", "reason", "status", "type" })
+@com.fasterxml.jackson.annotation.JsonPropertyOrder({ "observedGeneration", "type", "status", "lastTransitionTime", "reason", "message" })
 @com.fasterxml.jackson.databind.annotation.JsonDeserialize(using = com.fasterxml.jackson.databind.JsonDeserializer.None.class)
 @javax.annotation.processing.Generated("io.fabric8.java.generator.CRGeneratorRunner")
 @lombok.EqualsAndHashCode()

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

@@ -18,8 +18,6 @@
 
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
-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;
@@ -50,10 +48,10 @@ static String serviceName(VirtualKafkaCluster cluster) {
     public Map<String, Service> desiredResources(
                                                  KafkaProxy primary,
                                                  Context<KafkaProxy> context) {
-        ProxyModelBuilder proxyModelBuilder = ProxyModelBuilder.contextBuilder(context);
-        ProxyModel model = proxyModelBuilder.build(primary, context);
+        KafkaProxyContext kafkaProxyContext = KafkaProxyContext.proxyContext(context);
+        var model = kafkaProxyContext.model();
         Stream<Service> serviceStream = model.clustersWithValidIngresses().stream()
-                .filter(cluster -> !SharedKafkaProxyContext.isBroken(context, cluster))
+                .filter(cluster -> !kafkaProxyContext.isBroken(cluster))
                 .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/ClusterService.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.kubernetes.operator;
+
+import java.time.Clock;
+import java.util.stream.Collectors;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
+import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
+import io.kroxylicious.kubernetes.operator.model.ProxyModel;
+
+public record KafkaProxyContext(Clock clock,
+                                SecureConfigInterpolator secureConfigInterpolator,
+                                ProxyModel model) {
+
+    private static final String KEY_CTX = KafkaProxyContext.class.getName();
+
+    static void init(Context<KafkaProxy> context,
+                     Clock clock,
+                     SecureConfigInterpolator secureConfigInterpolator,
+                     ProxyModel model) {
+        var rc = context.managedWorkflowAndDependentResourceContext();
+
+        rc.put(KEY_CTX,
+                new KafkaProxyContext(
+                        Clock.fixed(clock.instant(), clock.getZone()),
+                        secureConfigInterpolator,
+                        model));
+    }
+
+    static KafkaProxyContext proxyContext(Context<KafkaProxy> context) {
+        return context.managedWorkflowAndDependentResourceContext().getMandatory(KEY_CTX, KafkaProxyContext.class);
+    }
+
+    boolean isBroken(VirtualKafkaCluster cluster) {
+        var fullyResolved = model().resolutionResult().fullyResolvedClustersInNameOrder().stream().map(ResourcesUtil::name).collect(Collectors.toSet());
+        return !fullyResolved.contains(ResourcesUtil.name(cluster))
+                || !model().ingressModel().clusterIngressModel(cluster).stream()
+                        .allMatch(ingressModel -> ingressModel.ingressExceptions().isEmpty());
+
+    }
+
+}

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

@@ -5,6 +5,7 @@
  */
 package io.kroxylicious.kubernetes.operator;
 
+import java.time.Clock;
 import java.time.Duration;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
@@ -45,6 +46,8 @@
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaClusterSpec;
 import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilter;
+import io.kroxylicious.kubernetes.operator.model.ProxyModel;
+import io.kroxylicious.kubernetes.operator.model.ProxyModelBuilder;
 import io.kroxylicious.proxy.tag.VisibleForTesting;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -83,22 +86,22 @@ public class KafkaProxyReconciler implements
     public static final String CONFIG_DEP = "config";
     public static final String DEPLOYMENT_DEP = "deployment";
     public static final String CLUSTERS_DEP = "clusters";
-    static final String SEC = "sec";
+
+    private final Clock clock;
     private final SecureConfigInterpolator secureConfigInterpolator;
 
-    public KafkaProxyReconciler(SecureConfigInterpolator secureConfigInterpolator) {
+    public KafkaProxyReconciler(Clock clock, SecureConfigInterpolator secureConfigInterpolator) {
+        this.clock = clock;
         this.secureConfigInterpolator = secureConfigInterpolator;
     }
 
-    static SecureConfigInterpolator secureConfigInterpolator(Context<KafkaProxy> context) {
-        return context.managedWorkflowAndDependentResourceContext().getMandatory(SEC, SecureConfigInterpolator.class);
-    }
-
     @Override
     public void initContext(
-                            KafkaProxy primary,
+                            KafkaProxy proxy,
                             Context<KafkaProxy> context) {
-        context.managedWorkflowAndDependentResourceContext().put(SEC, secureConfigInterpolator);
+        ProxyModelBuilder proxyModelBuilder = ProxyModelBuilder.contextBuilder();
+        ProxyModel model = proxyModelBuilder.build(proxy, context);
+        KafkaProxyContext.init(context, clock, secureConfigInterpolator, model);
     }
 
     /**
@@ -111,7 +114,7 @@ public UpdateControl<KafkaProxy> reconcile(KafkaProxy primary,
             LOGGER.info("Completed reconciliation of {}/{}", namespace(primary), name(primary));
         }
         return UpdateControl.patchStatus(
-                buildStatus(primary, context, null));
+                buildStatus(primary, null));
     }
 
     /**
@@ -123,7 +126,7 @@ public ErrorStatusUpdateControl<KafkaProxy> updateErrorStatus(KafkaProxy primary
                                                                   Exception exception) {
         // Post-condition: status.conditions should be in a canonical order (to avoid non-terminating reconciliations)
         // Post-condition: There is only one Ready condition
-        var control = ErrorStatusUpdateControl.patchStatus(buildStatus(primary, context, exception));
+        var control = ErrorStatusUpdateControl.patchStatus(buildStatus(primary, exception));
         if (exception instanceof InvalidResourceException) {
             control.withNoRetry();
         }
@@ -135,7 +138,6 @@ public ErrorStatusUpdateControl<KafkaProxy> updateErrorStatus(KafkaProxy primary
     }
 
     private static KafkaProxy buildStatus(KafkaProxy primary,
-                                          Context<KafkaProxy> context,
                                           @Nullable Exception exception) {
         if (exception instanceof AggregatedOperatorException aoe && aoe.getAggregatedExceptions().size() == 1) {
             exception = aoe.getAggregatedExceptions().values().iterator().next();
@@ -150,30 +152,17 @@ private static KafkaProxy buildStatus(KafkaProxy primary,
                 .endMetadata()
                 .withNewStatus()
                     .withObservedGeneration(generation(primary))
-                    .withConditions(effectiveReadyCondition(now, primary, exception ))
-                    .withClusters(clusterConditions(now, primary, context ))
+                    .withConditions(effectiveReadyCondition(now, primary, exception))
                 .endStatus()
             .build();
         // @formatter:on
     }
 
-    private static List<io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.Clusters> clusterConditions(ZonedDateTime now,
-                                                                                                             KafkaProxy primary,
-                                                                                                             Context<KafkaProxy> context) {
-        return ResourcesUtil.clustersInNameOrder(context).map(cluster -> {
-            ClusterCondition clusterCondition = SharedKafkaProxyContext.clusterCondition(context, cluster);
-            var conditions = newClusterCondition(now, primary, clusterCondition);
-            return new io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.ClustersBuilder()
-                    .withName(name(cluster))
-                    .withConditions(conditions).build();
-        }).toList();
-    }
-
     /**
      * Determines whether the {@code Ready} condition has had a state transition,
      * and returns an appropriate new {@code Ready} condition.
      *
-     * @param now
+     * @param now The current time
      * @param primary The primary.
      * @param exception An exception, or null if the reconciliation was successful.
      * @return The {@code Ready} condition to use in {@code status.conditions}.
@@ -224,7 +213,7 @@ else if (exception instanceof InvalidResourceException) {
     }
 
     /**
-     * @param now
+     * @param now The current time
      * @param primary The primary.
      * @param exception An exception, or null if the reconciliation was successful.
      * @return The {@code Ready} condition to use in {@code status.conditions}
@@ -245,20 +234,6 @@ private static Condition newCondition(
                 .build();
     }
 
-    private static Condition newClusterCondition(
-                                                 ZonedDateTime now,
-                                                 KafkaProxy primary,
-                                                 ClusterCondition clusterCondition) {
-        return new ConditionBuilder()
-                .withLastTransitionTime(now)
-                .withMessage(clusterCondition.message())
-                .withObservedGeneration(generation(primary))
-                .withReason(clusterCondition.reason())
-                .withStatus(clusterCondition.status())
-                .withType(clusterCondition.type())
-                .build();
-    }
-
     private static boolean isTransition(@Nullable Condition oldReady, @Nullable Exception exception) {
         if (oldReady == null) {
             return true;

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

@@ -79,7 +79,7 @@ public static void main(String[] args) {
      */
     void start() {
         operator.installShutdownHook(Duration.ofSeconds(10));
-        operator.register(new KafkaProxyReconciler(SecureConfigInterpolator.DEFAULT_INTERPOLATOR));
+        operator.register(new KafkaProxyReconciler(Clock.systemUTC(), SecureConfigInterpolator.DEFAULT_INTERPOLATOR));
         operator.register(new VirtualKafkaClusterReconciler(Clock.systemUTC()));
         operator.register(new KafkaProxyIngressReconciler(Clock.systemUTC()));
         operator.register(new KafkaServiceReconciler(Clock.systemUTC()));