DBZ-8234 ability to enable DS API

debezium-operator-api/src/main/java/io/debezium/operator/api/model/CommonLabels.java

@@ -8,6 +8,8 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
 public final class CommonLabels {
 
     public static final String KEY_APP_K8_INSTANCE = "app.kubernetes.io/instance";
@@ -52,4 +54,11 @@ public Map<String, String> getMap() {
         return new HashMap<>(map);
     }
 
+    public static boolean hasLabel(HasMetadata resource, String key, String expectedValue) {
+        var labels = resource.getMetadata().getLabels();
+        var actualValue = labels.get(key);
+
+        return actualValue != null && actualValue.equals(expectedValue);
+    }
+
 }

debezium-operator-api/src/main/java/io/debezium/operator/api/model/DebeziumServerSpec.java

@@ -140,6 +140,7 @@ public void setPredicates(Map<String, Predicate> predicates) {
     @Override
     public ConfigMapping asConfiguration() {
         var dbzConfig = ConfigMapping.prefixed("debezium")
+                .putAll(runtime)
                 .putAll("source", source)
                 .putAll("sink", sink)
                 .putAll("format", format)

debezium-operator-api/src/main/java/io/debezium/operator/api/model/runtime/Runtime.java

@@ -9,17 +9,23 @@
 import com.fasterxml.jackson.annotation.JsonPropertyDescription;
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 
+import io.debezium.operator.api.config.ConfigMappable;
+import io.debezium.operator.api.config.ConfigMapping;
 import io.debezium.operator.api.model.runtime.jmx.JmxConfig;
 import io.debezium.operator.api.model.runtime.metrics.Metrics;
 import io.debezium.operator.api.model.runtime.storage.RuntimeStorage;
 import io.debezium.operator.api.model.runtime.templates.Templates;
 import io.debezium.operator.docs.annotations.Documented;
 import io.sundr.builder.annotations.Buildable;
 
-@JsonPropertyOrder({ "storage", "environment", "jmx", "templates" })
+@JsonPropertyOrder({ "api", "storage", "environment", "jmx", "templates", "serviceAccount", "metrics" })
 @Documented
 @Buildable(editableEnabled = false, builderPackage = "io.fabric8.kubernetes.api.builder", lazyCollectionInitEnabled = false)
-public class Runtime {
+public class Runtime implements ConfigMappable {
+
+    @JsonPropertyDescription("API configuration")
+    private RuntimeApi api;
+
     @JsonPropertyDescription("Storage configuration")
     private RuntimeStorage storage;
 
@@ -40,13 +46,22 @@ public class Runtime {
     private Metrics metrics;
 
     public Runtime() {
+        this.api = new RuntimeApi();
         this.storage = new RuntimeStorage();
         this.environment = new RuntimeEnvironment();
         this.jmx = new JmxConfig();
         this.templates = new Templates();
         this.metrics = new Metrics();
     }
 
+    public RuntimeApi getApi() {
+        return api;
+    }
+
+    public void setApi(RuntimeApi api) {
+        this.api = api;
+    }
+
     public Templates getTemplates() {
         return templates;
     }
@@ -94,4 +109,10 @@ public Metrics getMetrics() {
     public void setMetrics(Metrics metrics) {
         this.metrics = metrics;
     }
+
+    @Override
+    public ConfigMapping asConfiguration() {
+        return ConfigMapping.empty()
+                .putAll("api", api);
+    }
 }

debezium-operator-api/src/main/java/io/debezium/operator/api/model/runtime/RuntimeApi.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.operator.api.model.runtime;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+
+import io.debezium.operator.api.config.ConfigMappable;
+import io.debezium.operator.api.config.ConfigMapping;
+import io.debezium.operator.docs.annotations.Documented;
+import io.sundr.builder.annotations.Buildable;
+
+@JsonPropertyOrder({ "enabled" })
+@Documented
+@Buildable(editableEnabled = false, builderPackage = "io.fabric8.kubernetes.api.builder", lazyCollectionInitEnabled = false)
+public class RuntimeApi implements ConfigMappable {
+
+    @JsonPropertyDescription("Whether the API should be enabled for this instance of Debezium Server")
+    private boolean enabled = false;
+
+    @JsonProperty(defaultValue = "8080")
+    @JsonPropertyDescription("Port number used by the k8s service exposing the API")
+    private int port = 8080;
+
+    public boolean isEnabled() {
+        return enabled;
+    }
+
+    public void setEnabled(boolean enabled) {
+        this.enabled = enabled;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public void setPort(int port) {
+        this.port = port;
+    }
+
+    @Override
+    public ConfigMapping asConfiguration() {
+        return ConfigMapping.empty()
+                .put("enabled", enabled);
+    }
+}

debezium-operator-api/src/test/java/io/debezium/operator/api/model/runtime/RuntimeConfigTest.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.operator.api.model.runtime;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.operator.api.model.DebeziumServerSpec;
+import io.debezium.operator.api.model.DebeziumServerSpecBuilder;
+
+public class RuntimeConfigTest {
+
+    static DebeziumServerSpec spec;
+
+    @BeforeAll
+    static void setup() {
+        spec = new DebeziumServerSpecBuilder()
+                .withRuntime(new RuntimeBuilder()
+                        .withApi(new RuntimeApiBuilder()
+                                .withEnabled(true)
+                                .build())
+                        .build())
+                .build();
+    }
+
+    @Test
+    void shouldHaveEnableApiProperty() {
+        assertThat(spec.asConfiguration().getAsMapSimple())
+                .containsEntry("debezium.api.enabled", "true");
+    }
+}

debezium-operator-core/src/main/java/io/debezium/operator/core/DebeziumServerReconciler.java

@@ -19,6 +19,7 @@
 import io.debezium.operator.api.model.status.ServerRunningCondition;
 import io.debezium.operator.api.model.status.ServerStoppedCondition;
 import io.debezium.operator.commons.OperatorConstants;
+import io.debezium.operator.core.dependent.ApiServiceDependent;
 import io.debezium.operator.core.dependent.ConfigMapDependent;
 import io.debezium.operator.core.dependent.DeploymentDependent;
 import io.debezium.operator.core.dependent.JmxExporterServiceDependent;
@@ -27,6 +28,7 @@
 import io.debezium.operator.core.dependent.RoleBindingDependent;
 import io.debezium.operator.core.dependent.RoleDependent;
 import io.debezium.operator.core.dependent.ServiceAccountDependent;
+import io.debezium.operator.core.dependent.conditions.ApiEnabled;
 import io.debezium.operator.core.dependent.conditions.CreatePvc;
 import io.debezium.operator.core.dependent.conditions.CreateServiceAccount;
 import io.debezium.operator.core.dependent.conditions.DeploymentReady;
@@ -58,7 +60,8 @@
                 "role-binding",
         }, reconcilePrecondition = PvcReady.class, readyPostcondition = DeploymentReady.class),
         @Dependent(name = "jmx-service", type = JmxServiceDependent.class, dependsOn = "deployment", reconcilePrecondition = JmxEnabled.class),
-        @Dependent(name = "jmx-exporter-service", type = JmxExporterServiceDependent.class, dependsOn = "deployment", reconcilePrecondition = JmxExporterEnabled.class)
+        @Dependent(name = "jmx-exporter-service", type = JmxExporterServiceDependent.class, dependsOn = "deployment", reconcilePrecondition = JmxExporterEnabled.class),
+        @Dependent(name = "api-service", type = ApiServiceDependent.class, dependsOn = "deployment", reconcilePrecondition = ApiEnabled.class)
 })
 @CSVMetadata(name = OperatorConstants.CSV_INTERNAL_BUNDLE_NAME)
 @RBACRule(verbs = { RBACVerbs.GET, RBACVerbs.LIST, RBACVerbs.WATCH }, apiGroups = "", resources = "secrets")

debezium-operator-core/src/main/java/io/debezium/operator/core/dependent/ApiServiceDependent.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.operator.core.dependent;
+
+import io.debezium.operator.api.model.CommonLabels;
+import io.debezium.operator.api.model.DebeziumServer;
+import io.debezium.operator.core.dependent.discriminators.ApiServiceDiscriminator;
+import io.fabric8.kubernetes.api.model.IntOrString;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.Service;
+import io.fabric8.kubernetes.api.model.ServiceBuilder;
+import io.fabric8.kubernetes.api.model.ServicePortBuilder;
+import io.fabric8.kubernetes.api.model.ServiceSpecBuilder;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.processing.dependent.kubernetes.CRUDKubernetesDependentResource;
+import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
+
+@KubernetesDependent(resourceDiscriminator = ApiServiceDiscriminator.class)
+public class ApiServiceDependent extends CRUDKubernetesDependentResource<Service, DebeziumServer> {
+
+    public static final String CLASSIFIER = "api";
+
+    public ApiServiceDependent() {
+        super(Service.class);
+    }
+
+    @Override
+    protected Service desired(DebeziumServer primary, Context<DebeziumServer> context) {
+        var name = primary.getMetadata().getName();
+        var apiConfig = primary.getSpec().getRuntime().getApi();
+
+        var labels = CommonLabels.serverComponent(name)
+                .withDbzClassifier(CLASSIFIER)
+                .getMap();
+        var selector = CommonLabels.serverComponent(name)
+                .getMap();
+
+        return new ServiceBuilder()
+                .withMetadata(new ObjectMetaBuilder()
+                        .withNamespace(primary.getMetadata().getNamespace())
+                        .withName(name + "-api")
+                        .withLabels(labels)
+                        .build())
+                .withSpec(new ServiceSpecBuilder()
+                        .withSelector(selector)
+                        .withPorts(new ServicePortBuilder()
+                                .withName("http")
+                                .withProtocol("TCP")
+                                .withTargetPort(new IntOrString(DeploymentDependent.DEFAULT_HTTP_PORT))
+                                .withPort(apiConfig.getPort())
+                                .build())
+                        .build())
+                .build();
+    }
+}

debezium-operator-core/src/main/java/io/debezium/operator/core/dependent/conditions/ApiEnabled.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.operator.core.dependent.conditions;
+
+import io.debezium.operator.api.model.DebeziumServer;
+import io.fabric8.kubernetes.api.model.Service;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.processing.dependent.workflow.Condition;
+
+public class ApiEnabled implements Condition<Service, DebeziumServer> {
+    @Override
+    public boolean isMet(
+                         DependentResource<Service, DebeziumServer> dependentResource,
+                         DebeziumServer primary,
+                         Context<DebeziumServer> context) {
+        var apiConfig = primary.getSpec().getRuntime().getApi();
+        return apiConfig.isEnabled();
+    }
+}

debezium-operator-core/src/main/java/io/debezium/operator/core/dependent/discriminators/ApiServiceDiscriminator.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.operator.core.dependent.discriminators;
+
+import static io.debezium.operator.api.model.CommonLabels.hasLabel;
+
+import java.util.Optional;
+
+import io.debezium.operator.api.model.CommonLabels;
+import io.debezium.operator.api.model.DebeziumServer;
+import io.debezium.operator.core.dependent.JmxServiceDependent;
+import io.fabric8.kubernetes.api.model.Service;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
+
+public class ApiServiceDiscriminator implements ResourceDiscriminator<Service, DebeziumServer> {
+
+    @Override
+    public Optional<Service> distinguish(Class<Service> resource, DebeziumServer primary, Context<DebeziumServer> context) {
+        return context.getSecondaryResourcesAsStream(Service.class)
+                .filter(s -> hasLabel(s, CommonLabels.KEY_DBZ_CLASSIFIER, JmxServiceDependent.JMX_CLASSIFIER))
+                .findFirst();
+    }
+}

debezium-operator-core/src/main/java/io/debezium/operator/core/dependent/discriminators/JmxExporterServiceDiscriminator.java

@@ -5,12 +5,13 @@
  */
 package io.debezium.operator.core.dependent.discriminators;
 
+import static io.debezium.operator.api.model.CommonLabels.hasLabel;
+
 import java.util.Optional;
 
 import io.debezium.operator.api.model.CommonLabels;
 import io.debezium.operator.api.model.DebeziumServer;
 import io.debezium.operator.core.dependent.JmxExporterServiceDependent;
-import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.Service;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
@@ -23,11 +24,4 @@ public Optional<Service> distinguish(Class<Service> resource, DebeziumServer pri
                 .filter(s -> hasLabel(s, CommonLabels.KEY_DBZ_CLASSIFIER, JmxExporterServiceDependent.JMX_EXPORTER_CLASSIFIER))
                 .findFirst();
     }
-
-    private boolean hasLabel(HasMetadata resource, String key, String expectedValue) {
-        var labels = resource.getMetadata().getLabels();
-        var actualValue = labels.get(key);
-
-        return actualValue != null && actualValue.equals(expectedValue);
-    }
 }

debezium-operator-core/src/main/java/io/debezium/operator/core/dependent/discriminators/JmxServiceDiscriminator.java

@@ -5,12 +5,13 @@
  */
 package io.debezium.operator.core.dependent.discriminators;
 
+import static io.debezium.operator.api.model.CommonLabels.hasLabel;
+
 import java.util.Optional;
 
 import io.debezium.operator.api.model.CommonLabels;
 import io.debezium.operator.api.model.DebeziumServer;
 import io.debezium.operator.core.dependent.JmxServiceDependent;
-import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.Service;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
@@ -23,11 +24,4 @@ public Optional<Service> distinguish(Class<Service> resource, DebeziumServer pri
                 .filter(s -> hasLabel(s, CommonLabels.KEY_DBZ_CLASSIFIER, JmxServiceDependent.JMX_CLASSIFIER))
                 .findFirst();
     }
-
-    private boolean hasLabel(HasMetadata resource, String key, String expectedValue) {
-        var labels = resource.getMetadata().getLabels();
-        var actualValue = labels.get(key);
-
-        return actualValue != null && actualValue.equals(expectedValue);
-    }
 }

debezium-operator-dist/src/main/examples/postgres/010_debezium-server-ephemeral.yml

@@ -9,6 +9,9 @@ spec:
       log.console.json: false
       kubernetes-config.enabled: true
       kubernetes-config.secrets: postgresql-credentials
+  runtime:
+    api:
+      enabled: true
   sink:
     type: kafka
     config: