Add VirtualKafkaClusterReconciler (kroxylicious#1992)

* Add `VirtualKafkaClusterReconciler`
* Bump operator memory limit and request to avoid observed OOMKilled pods.

Signed-off-by: Tom Bentley <tbentley@redhat.com>

Author: tombentley

kroxylicious-operator/install/01.ClusterRole.kroxylicious-operator-watched.yaml

@@ -31,6 +31,7 @@ rules:
       - "kroxylicious.io"
     resources:
       - kafkaproxies/status
+      - virtualkafkaclusters/status
       - kafkaservices/status
       - kafkaproxyingresses/status
     verbs:

kroxylicious-operator/install/03.Deployment.kroxylicious-operator.yaml

@@ -32,10 +32,10 @@ spec:
           args: [ ]
           resources:
             limits:
-              memory: 300M
+              memory: 400M
               cpu: "1"
             requests:
-              memory: 300M
+              memory: 400M
               cpu: "1"
           ports:
             - containerPort: 8080

kroxylicious-operator/pom.xml

@@ -349,6 +349,9 @@
                         <io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.Conditions>
                             io.kroxylicious.kubernetes.api.common.Condition
                         </io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.Conditions>
+                        <io.kroxylicious.kubernetes.api.v1alpha1.virtualkafkaclusterstatus.Conditions>
+                            io.kroxylicious.kubernetes.api.common.Condition
+                        </io.kroxylicious.kubernetes.api.v1alpha1.virtualkafkaclusterstatus.Conditions>
                         <io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.clusters.Conditions>
                             io.kroxylicious.kubernetes.api.common.Condition
                         </io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.clusters.Conditions>

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

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

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

@@ -45,6 +45,7 @@
 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;
 
@@ -301,6 +302,22 @@ private static <R extends CustomResource<?, S>, S> R newStatus(R ingress,
         return result;
     }
 
+    @NonNull
+    static VirtualKafkaCluster patchWithCondition(VirtualKafkaCluster cluster, Condition condition) {
+        return newStatus(
+                cluster,
+                VirtualKafkaCluster::new,
+                VirtualKafkaClusterStatus::new,
+                VirtualKafkaClusterStatus::setConditions,
+                VirtualKafkaClusterStatus::setObservedGeneration,
+                maybeAddOrUpdateCondition(
+                        Optional.of(cluster)
+                                .map(VirtualKafkaCluster::getStatus)
+                                .map(VirtualKafkaClusterStatus::getConditions)
+                                .orElse(List.of()),
+                        condition));
+    }
+
     @NonNull
     static KafkaService patchWithCondition(KafkaService service, Condition condition) {
         return newStatus(

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

@@ -0,0 +1,91 @@
+/*
+ * 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.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+
+import io.kroxylicious.kubernetes.api.common.Condition;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
+import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
+
+import static io.kroxylicious.kubernetes.operator.ResourcesUtil.name;
+import static io.kroxylicious.kubernetes.operator.ResourcesUtil.namespace;
+
+public final class VirtualKafkaClusterReconciler implements
+        Reconciler<VirtualKafkaCluster> {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(VirtualKafkaClusterReconciler.class);
+    public static final String PROXY_EVENT_SOURCE_NAME = "proxy";
+
+    private final Clock clock;
+
+    public VirtualKafkaClusterReconciler(Clock clock) {
+        this.clock = clock;
+    }
+
+    @Override
+    public UpdateControl<VirtualKafkaCluster> reconcile(VirtualKafkaCluster cluster, Context<VirtualKafkaCluster> context) {
+        var proxyOpt = context.getSecondaryResource(KafkaProxy.class, PROXY_EVENT_SOURCE_NAME);
+        LOGGER.debug("spec.proxyRef.name resolves to: {}", proxyOpt);
+
+        Condition condition;
+        if (proxyOpt.isPresent()) {
+            condition = ResourcesUtil.newResolvedRefsTrue(clock, cluster);
+        }
+        else {
+            condition = ResourcesUtil.newResolvedRefsFalse(clock,
+                    cluster,
+                    "spec.proxyRef.name",
+                    "KafkaProxy not found");
+        }
+
+        UpdateControl<VirtualKafkaCluster> uc = UpdateControl.patchStatus(ResourcesUtil.patchWithCondition(cluster, condition));
+        if (LOGGER.isInfoEnabled()) {
+            LOGGER.info("Completed reconciliation of {}/{}", namespace(cluster), name(cluster));
+        }
+        return uc;
+    }
+
+    @Override
+    public List<EventSource<?, VirtualKafkaCluster>> prepareEventSources(EventSourceContext<VirtualKafkaCluster> context) {
+        InformerEventSourceConfiguration<KafkaProxy> configuration = InformerEventSourceConfiguration.from(
+                KafkaProxy.class,
+                VirtualKafkaCluster.class)
+                .withName(PROXY_EVENT_SOURCE_NAME)
+                .withPrimaryToSecondaryMapper((VirtualKafkaCluster cluster) -> ResourcesUtil.localRefAsResourceId(cluster, cluster.getSpec().getProxyRef()))
+                .withSecondaryToPrimaryMapper(proxy -> ResourcesUtil.findReferrers(context,
+                        proxy,
+                        VirtualKafkaCluster.class,
+                        cluster -> cluster.getSpec().getProxyRef()))
+                .build();
+        return List.of(new InformerEventSource<>(configuration, context));
+    }
+
+    @Override
+    public ErrorStatusUpdateControl<VirtualKafkaCluster> updateErrorStatus(VirtualKafkaCluster cluster, Context<VirtualKafkaCluster> context, Exception e) {
+        // ResolvedRefs to UNKNOWN
+        Condition condition = ResourcesUtil.newUnknownCondition(clock, cluster, Condition.Type.ResolvedRefs, e);
+        ErrorStatusUpdateControl<VirtualKafkaCluster> uc = ErrorStatusUpdateControl.patchStatus(ResourcesUtil.patchWithCondition(cluster, condition));
+        if (LOGGER.isInfoEnabled()) {
+            LOGGER.info("Completed reconciliation of {}/{} for error {}", namespace(cluster), name(cluster), e.toString());
+        }
+        return uc;
+    }
+}

kroxylicious-operator/src/main/resources/META-INF/fabric8/virtualkafkaclusters.kroxylicious.io-v1.yml

@@ -86,4 +86,51 @@ spec:
                       name:
                         maxLength: 253
                         minLength: 1
+                        type: string
+            status:
+              type: object
+              properties:
+                observedGeneration:
+                  description: |
+                    The metadata.generation that was observed during the last reconciliation by the operator.
+                  type: integer
+                conditions:
+                  # Mapped to Java type io.kroxylicious.kubernetes.api.common.Condition
+                  type: array
+                  items:
+                    type: object
+                    properties:
+                      lastTransitionTime:
+                        description: |
+                          lastTransitionTime is the last time the condition transitioned from one status to another. 
+                          This should be when the underlying condition changed. 
+                          If that is not known, then using the time when the API field changed is acceptable.
+                        type: string
+                        format: date-time
+                      message:
+                        description: |
+                          message is a human readable message indicating details about the transition. 
+                          This may be an empty string.
+                        type: string
+                      observedGeneration:
+                        description: |
+                          observedGeneration represents the .metadata.generation that the condition was set based upon. 
+                          For instance, if .metadata.generation is currently 12, but the 
+                          .status.conditions[x].observedGeneration is 9, the condition is out of date with 
+                          respect to the current state of the instance.
+                        type: integer
+                      reason:
+                        description: |
+                          reason contains a programmatic identifier indicating the reason for the condition's last transition. 
+                          Producers of specific condition types may define expected values and meanings for this field, 
+                          and whether the values are considered a guaranteed API. 
+                          The value should be a CamelCase string. 
+                          This field may not be empty.
+                        type: string
+                      status:
+                        description: status of the condition, one of True, False, Unknown.
+                        type: string
+                        enum: [ "True", "False", "Unknown" ]
+                      type:
+                        description: type of condition in CamelCase or in foo.example.com/CamelCase.
                         type: string

kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/KafkaServiceReconcilerIT.java

@@ -33,6 +33,7 @@ class KafkaServiceReconcilerIT {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaServiceReconcilerIT.class);
     private static final String UPDATED_BOOTSTRAP = "bar.bootstrap:9090";
+    public static final String SERVICE_A = "service-a";
 
     private KubernetesClient client;
     private static final ConditionFactory AWAIT = await().timeout(Duration.ofSeconds(60));
@@ -60,16 +61,15 @@ void stopOperator() {
     @Test
     void shouldAcceptKafkaService() {
         // Given
-        final KafkaService kafkaService = extension.create(new KafkaServiceBuilder().withNewMetadata().withName("mycoolkafkaservice").endMetadata().editOrNewSpec()
-                .withBootstrapServers("foo.bootstrap:9090").endSpec().build());
+        final KafkaService kafkaService = extension.create(kafkaService(SERVICE_A));
         final KafkaService updated = kafkaService.edit().editSpec().withBootstrapServers(UPDATED_BOOTSTRAP).endSpec().build();
 
         // When
         extension.replace(updated);
 
         // Then
         AWAIT.untilAsserted(() -> {
-            final KafkaService mycoolkafkaservice = extension.get(KafkaService.class, "mycoolkafkaservice");
+            final KafkaService mycoolkafkaservice = extension.get(KafkaService.class, SERVICE_A);
             Assertions.assertThat(mycoolkafkaservice.getSpec().getBootstrapServers()).isEqualTo(UPDATED_BOOTSTRAP);
             assertThat(mycoolkafkaservice.getStatus())
                     .isNotNull()
@@ -84,16 +84,29 @@ void shouldAcceptUpdateToKafkaService() {
         // Given
 
         // When
-        extension.create(new KafkaServiceBuilder().withNewMetadata().withName("mycoolkafkaservice").endMetadata().build());
+        extension.create(new KafkaServiceBuilder().withNewMetadata().withName(SERVICE_A).endMetadata().build());
 
         // Then
         AWAIT.untilAsserted(() -> {
-            final KafkaService mycoolkafkaservice = extension.get(KafkaService.class, "mycoolkafkaservice");
+            final KafkaService mycoolkafkaservice = extension.get(KafkaService.class, SERVICE_A);
             assertThat(mycoolkafkaservice.getStatus())
                     .isNotNull()
                     .singleCondition()
                     .hasObservedGenerationInSyncWithMetadataOf(mycoolkafkaservice)
                     .isAcceptedTrue();
         });
     }
+
+    private static KafkaService kafkaService(String name) {
+        // @formatter:off
+        return new KafkaServiceBuilder()
+                .withNewMetadata()
+                    .withName(name)
+                .endMetadata()
+                .editOrNewSpec()
+                    .withBootstrapServers("foo.bootstrap:9090")
+                .endSpec()
+                .build();
+        // @formatter:on
+    }
 }