Merge pull request kroxylicious#1970 from SamBarker/filter-reconciler

Stub `KafkaServiceReconciler`

Author: SamBarker

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

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

kroxylicious-operator/pom.xml

@@ -355,6 +355,9 @@
                         <io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxyingressstatus.Conditions>
                             io.kroxylicious.kubernetes.api.common.Condition
                         </io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxyingressstatus.Conditions>
+                        <io.kroxylicious.kubernetes.api.v1alpha1.kafkaservicestatus.Conditions>
+                            io.kroxylicious.kubernetes.api.common.Condition
+                        </io.kroxylicious.kubernetes.api.v1alpha1.kafkaservicestatus.Conditions>
                         <io.kroxylicious.kubernetes.filter.api.v1alpha1.kafkaprotocolfilterspec.ConfigTemplate>
                             java.lang.Object
                         </io.kroxylicious.kubernetes.filter.api.v1alpha1.kafkaprotocolfilterspec.ConfigTemplate>

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

@@ -27,6 +27,7 @@
 import io.micrometer.prometheusmetrics.PrometheusMeterRegistry;
 import io.prometheus.metrics.exporter.httpserver.MetricsHandler;
 
+import io.kroxylicious.kubernetes.operator.kafkaservice.KafkaServiceReconciler;
 import io.kroxylicious.kubernetes.operator.management.UnsupportedHttpMethodFilter;
 import io.kroxylicious.proxy.service.HostPort;
 import io.kroxylicious.proxy.tag.VisibleForTesting;
@@ -81,6 +82,7 @@ void start() {
         operator.installShutdownHook(Duration.ofSeconds(10));
         operator.register(new KafkaProxyReconciler());
         operator.register(new KafkaProxyIngressReconciler(Clock.systemUTC()));
+        operator.register(new KafkaServiceReconciler(Clock.systemUTC()));
         addHttpGetHandler("/", () -> 404);
         managementServer.start();
         operator.start();

kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/kafkaservice/KafkaServiceReconciler.java

@@ -0,0 +1,58 @@
+/*
+ * 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.kafkaservice;
+
+import java.time.Clock;
+import java.util.List;
+
+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.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+
+import io.kroxylicious.kubernetes.api.common.Condition;
+import io.kroxylicious.kubernetes.api.common.ConditionBuilder;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaService;
+
+public final class KafkaServiceReconciler implements
+        io.javaoperatorsdk.operator.api.reconciler.Reconciler<KafkaService> {
+
+    private final Clock clock;
+
+    public KafkaServiceReconciler(Clock clock) {
+        this.clock = clock;
+    }
+
+    @Override
+    public UpdateControl<KafkaService> reconcile(KafkaService resource, Context<KafkaService> context) {
+        final Condition acceptedCondition = new ConditionBuilder()
+                .withType(Condition.Type.Accepted)
+                .withLastTransitionTime(clock.instant().atZone(clock.getZone()))
+                .withObservedGeneration(resource.getMetadata().getGeneration())
+                .withStatus(Condition.Status.TRUE)
+                .build();
+        final KafkaService amended = resource
+                .edit()
+                .withNewStatus()
+                .withObservedGeneration(resource.getMetadata().getGeneration())
+                .withConditions(acceptedCondition)
+                .endStatus()
+                .build();
+        return UpdateControl.patchStatus(amended);
+    }
+
+    @Override
+    public List<EventSource<?, KafkaService>> prepareEventSources(EventSourceContext<KafkaService> context) {
+        return io.javaoperatorsdk.operator.api.reconciler.Reconciler.super.prepareEventSources(context);
+    }
+
+    @Override
+    public ErrorStatusUpdateControl<KafkaService> updateErrorStatus(KafkaService resource, Context<KafkaService> context, Exception e) {
+        return io.javaoperatorsdk.operator.api.reconciler.Reconciler.super.updateErrorStatus(resource, context, e);
+    }
+}

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

@@ -21,7 +21,7 @@ spec:
     singular: kafkaservice
     kind: KafkaService
     shortNames:
-      - kcr
+      - ks
   # list of versions supported by this CustomResourceDefinition
   versions:
     - name: v1alpha1
@@ -44,3 +44,49 @@ spec:
                   type: string
                   minLength: 1
                   pattern: ^(([^:]+:[0-9]{1,5}),)*([^:]+:[0-9]{1,5})$
+            status:
+              type: object
+              properties:
+                observedGeneration:
+                  description: |
+                    The metadata.generation that was observed during the last reconciliation by the operator.
+                  type: integer
+                conditions:
+                  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

@@ -0,0 +1,100 @@
+/*
+ * 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.time.Duration;
+
+import org.assertj.core.api.Assertions;
+import org.awaitility.core.ConditionFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIf;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaService;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaServiceBuilder;
+import io.kroxylicious.kubernetes.operator.kafkaservice.KafkaServiceReconciler;
+
+import static io.kroxylicious.kubernetes.operator.assertj.OperatorAssertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+@EnabledIf(value = "io.kroxylicious.kubernetes.operator.OperatorTestUtils#isKubeClientAvailable", disabledReason = "no viable kube client available")
+class KafkaServiceReconcilerIT {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaServiceReconcilerIT.class);
+    private static final String UPDATED_BOOTSTRAP = "bar.bootstrap:9090";
+
+    private KubernetesClient client;
+    private static final ConditionFactory AWAIT = await().timeout(Duration.ofSeconds(60));
+
+    @BeforeEach
+    void beforeEach() {
+        client = OperatorTestUtils.kubeClient();
+    }
+
+    @SuppressWarnings("JUnitMalformedDeclaration") // The beforeAll and beforeEach have the same effect so we can use it as an instance field.
+    @RegisterExtension
+    LocallyRunOperatorExtension extension = LocallyRunOperatorExtension.builder()
+            .withReconciler(new KafkaServiceReconciler(Clock.systemUTC()))
+            .withKubernetesClient(client)
+            .waitForNamespaceDeletion(true)
+            .withConfigurationService(x -> x.withCloseClientOnStop(false))
+            .build();
+
+    @AfterEach
+    void stopOperator() {
+        extension.getOperator().stop();
+        LOGGER.atInfo().log("Test finished");
+    }
+
+    @Test
+    void shouldAcceptKafkaService() {
+        // Given
+        final KafkaService kafkaService = extension.create(new KafkaServiceBuilder().withNewMetadata().withName("mycoolkafkaservice").endMetadata().editOrNewSpec()
+                .withBootstrapServers("foo.bootstrap:9090").endSpec().build());
+        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");
+            Assertions.assertThat(mycoolkafkaservice.getSpec().getBootstrapServers()).isEqualTo(UPDATED_BOOTSTRAP);
+            assertThat(mycoolkafkaservice.getStatus())
+                    .isNotNull()
+                    .singleCondition()
+                    .hasObservedGenerationInSyncWithMetadataOf(mycoolkafkaservice)
+                    .isAcceptedTrue();
+        });
+    }
+
+    @Test
+    void shouldAcceptUpdateToKafkaService() {
+        // Given
+
+        // When
+        extension.create(new KafkaServiceBuilder().withNewMetadata().withName("mycoolkafkaservice").endMetadata().build());
+
+        // Then
+        AWAIT.untilAsserted(() -> {
+            final KafkaService mycoolkafkaservice = extension.get(KafkaService.class, "mycoolkafkaservice");
+            assertThat(mycoolkafkaservice.getStatus())
+                    .isNotNull()
+                    .singleCondition()
+                    .hasObservedGenerationInSyncWithMetadataOf(mycoolkafkaservice)
+                    .isAcceptedTrue();
+        });
+    }
+}

kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/assertj/KafkaProxyStatusAssert.java

@@ -6,10 +6,12 @@
 
 package io.kroxylicious.kubernetes.operator.assertj;
 
+import org.assertj.core.api.AbstractLongAssert;
 import org.assertj.core.api.Assertions;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.assertj.core.api.ListAssert;
 
+import io.kroxylicious.kubernetes.api.common.Condition;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyStatus;
 import io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.Clusters;
 
@@ -25,6 +27,22 @@ public static KafkaProxyStatusAssert assertThat(KafkaProxyStatus actual) {
         return new KafkaProxyStatusAssert(actual);
     }
 
+    @Override
+    public AbstractLongAssert<?> observedGeneration() {
+        return Assertions.assertThat(actual.getObservedGeneration());
+    }
+
+    @Override
+    public ListAssert<Condition.Status> conditions() {
+        return Assertions.assertThat(actual.getConditions())
+                .asInstanceOf(InstanceOfAssertFactories.list(Condition.Status.class));
+    }
+
+    @Override
+    public ConditionAssert singleCondition() {
+        return conditions().singleElement(AssertFactory.condition());
+    }
+
     public ListAssert<Clusters> clusters() {
         return Assertions.assertThat(actual.getClusters())
                 .asInstanceOf(InstanceOfAssertFactories.list(io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.Clusters.class));

kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/assertj/KafkaServiceStatusAssert.java

@@ -0,0 +1,41 @@
+/*
+ * 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.assertj;
+
+import org.assertj.core.api.AbstractLongAssert;
+import org.assertj.core.api.AbstractObjectAssert;
+import org.assertj.core.api.Assertions;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.assertj.core.api.ListAssert;
+
+import io.kroxylicious.kubernetes.api.common.Condition;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaServiceStatus;
+
+public class KafkaServiceStatusAssert extends AbstractObjectAssert<KafkaServiceStatusAssert, KafkaServiceStatus> {
+    protected KafkaServiceStatusAssert(
+                                       KafkaServiceStatus o) {
+        super(o, KafkaServiceStatusAssert.class);
+    }
+
+    public static KafkaServiceStatusAssert assertThat(KafkaServiceStatus actual) {
+        return new KafkaServiceStatusAssert(actual);
+    }
+
+    public AbstractLongAssert<?> observedGeneration() {
+        return Assertions.assertThat(actual.getObservedGeneration());
+    }
+
+    public ListAssert<Condition.Status> conditions() {
+        return Assertions.assertThat(actual.getConditions())
+                .asInstanceOf(InstanceOfAssertFactories.list(Condition.Status.class));
+    }
+
+    public ConditionAssert singleCondition() {
+        return conditions().singleElement(AssertFactory.condition());
+    }
+
+}

kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/assertj/OperatorAssertions.java

@@ -8,13 +8,18 @@
 
 import io.kroxylicious.kubernetes.api.common.Condition;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyStatus;
+import io.kroxylicious.kubernetes.api.v1alpha1.KafkaServiceStatus;
 import io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxystatus.Clusters;
 
 public class OperatorAssertions {
     public static KafkaProxyStatusAssert assertThat(KafkaProxyStatus actual) {
         return KafkaProxyStatusAssert.assertThat(actual);
     }
 
+    public static KafkaServiceStatusAssert assertThat(KafkaServiceStatus actual) {
+        return KafkaServiceStatusAssert.assertThat(actual);
+    }
+
     public static ClusterAssert assertThat(Clusters actual) {
         return ClusterAssert.assertThat(actual);
     }