Add support for Image Volumes (strimzi#11467)

Signed-off-by: Jakub Scholz <www@scholzj.com>

Author: scholzj

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 * Add support for Kafka 3.9.1
 * Fixed MirrorMaker 2 client rack init container override being ignored.
+* Support for Kubernetes Image Volumes to mount custom plugins
 
 ### Major changes, deprecations and removals
 

api/src/main/java/io/strimzi/api/kafka/model/common/template/AdditionalVolume.java

@@ -9,6 +9,7 @@
 import io.fabric8.kubernetes.api.model.CSIVolumeSource;
 import io.fabric8.kubernetes.api.model.ConfigMapVolumeSource;
 import io.fabric8.kubernetes.api.model.EmptyDirVolumeSource;
+import io.fabric8.kubernetes.api.model.ImageVolumeSource;
 import io.fabric8.kubernetes.api.model.PersistentVolumeClaimVolumeSource;
 import io.fabric8.kubernetes.api.model.SecretVolumeSource;
 import io.strimzi.api.kafka.model.common.Constants;
@@ -28,14 +29,15 @@
  */
 @Buildable(editableEnabled = false, builderPackage = Constants.FABRIC8_KUBERNETES_API)
 @JsonInclude(JsonInclude.Include.NON_NULL)
-@JsonPropertyOrder({ "name", "secret", "configMap", "emptyDir", "persistentVolumeClaim", "csi" })
+@JsonPropertyOrder({ "name", "secret", "configMap", "emptyDir", "persistentVolumeClaim", "csi", "image" })
 @OneOf({
     @OneOf.Alternative({
         @OneOf.Alternative.Property(value = "secret", required = false),
         @OneOf.Alternative.Property(value = "configMap", required = false),
         @OneOf.Alternative.Property(value = "emptyDir", required = false),
         @OneOf.Alternative.Property(value = "persistentVolumeClaim", required = false),
-        @OneOf.Alternative.Property(value = "csi", required = false)
+        @OneOf.Alternative.Property(value = "csi", required = false),
+        @OneOf.Alternative.Property(value = "image", required = false)
         })
 })
 @EqualsAndHashCode
@@ -47,6 +49,7 @@ public class AdditionalVolume implements UnknownPropertyPreserving {
     private EmptyDirVolumeSource emptyDir;
     private PersistentVolumeClaimVolumeSource persistentVolumeClaim;
     private CSIVolumeSource csi;
+    private ImageVolumeSource image;
     private Map<String, Object> additionalProperties = new HashMap<>(0);
 
     @Description("Name to use for the volume. Required.")
@@ -59,7 +62,7 @@ public void setName(String name) {
         this.name = name;
     }
 
-    @Description("Secret to use populate the volume.")
+    @Description("`Secret` to use to populate the volume.")
     @KubeLink(group = "core", version = "v1", kind = "secretvolumesource")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     public SecretVolumeSource getSecret() {
@@ -70,7 +73,7 @@ public void setSecret(SecretVolumeSource secret) {
         this.secret = secret;
     }
 
-    @Description("ConfigMap to use to populate the volume.")
+    @Description("`ConfigMap` to use to populate the volume.")
     @KubeLink(group = "core", version = "v1", kind = "configmapvolumesource")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     public ConfigMapVolumeSource getConfigMap() {
@@ -81,7 +84,7 @@ public void setConfigMap(ConfigMapVolumeSource configMap) {
         this.configMap = configMap;
     }
 
-    @Description("EmptyDir to use to populate the volume.")
+    @Description("`EmptyDir` to use to populate the volume.")
     @KubeLink(group = "core", version = "v1", kind = "emptydirvolumesource")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     public EmptyDirVolumeSource getEmptyDir() {
@@ -92,7 +95,7 @@ public void setEmptyDir(EmptyDirVolumeSource emptyDir) {
         this.emptyDir = emptyDir;
     }
 
-    @Description("PersistentVolumeClaim object to use to populate the volume.")
+    @Description("`PersistentVolumeClaim` object to use to populate the volume.")
     @KubeLink(group = "core", version = "v1", kind = "persistentvolumeclaimvolumesource")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     public PersistentVolumeClaimVolumeSource getPersistentVolumeClaim() {
@@ -103,7 +106,7 @@ public void setPersistentVolumeClaim(PersistentVolumeClaimVolumeSource persisten
         this.persistentVolumeClaim = persistentVolumeClaim;
     }
 
-    @Description("CSIVolumeSource object to use to populate the volume.")
+    @Description("`CSIVolumeSource` object to use to populate the volume.")
     @KubeLink(group = "core", version = "v1", kind = "csivolumesource")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     public CSIVolumeSource getCsi() {
@@ -114,6 +117,17 @@ public void setCsi(CSIVolumeSource csi) {
         this.csi = csi;
     }
 
+    @Description("`ImageVolumeSource` object to use to populate the volume.")
+    @KubeLink(group = "core", version = "v1", kind = "imagevolumesource")
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    public ImageVolumeSource getImage() {
+        return image;
+    }
+
+    public void setImage(ImageVolumeSource image) {
+        this.image = image;
+    }
+
     @Override
     public Map<String, Object> getAdditionalProperties() {
         return this.additionalProperties;

api/src/main/java/io/strimzi/api/kafka/model/connect/ImageArtifact.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model.connect;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.strimzi.api.kafka.model.common.Constants;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.sundr.builder.annotations.Buildable;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+/**
+ * Image Volume artifact represents an artifact which is mounted as a Kubernetes Image Volume
+ */
+@Buildable(
+        editableEnabled = false,
+        builderPackage = Constants.FABRIC8_KUBERNETES_API
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonPropertyOrder({ "type", "reference", "pullPolicy" })
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true)
+public class ImageArtifact extends MountedArtifact {
+    private String reference;
+    private String pullPolicy;
+
+    @Description("Must be `" + TYPE_IMAGE + "`")
+    @Override
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public String getType() {
+        return TYPE_IMAGE;
+    }
+
+    @Description("Reference to the container image (OCI artifact) containing the Kafka Connect plugin. " +
+            "The image is mounted as a volume and provides the plugin binary. " +
+            "Required.")
+    @JsonProperty(required = true)
+    public String getReference() {
+        return reference;
+    }
+
+    public void setReference(String reference) {
+        this.reference = reference;
+    }
+
+    @Description("Policy that determines when the container image (OCI artifact) is pulled.\n\n" +
+            "Possible values are:\n\n" +
+            "* `Always`: Always pull the image. If the pull fails, container creation fails.\n" +
+            "* `Never`: Never pull the image. Use only a locally available image. Container creation fails if the image isn’t present.\n" +
+            "* `IfNotPresent`: Pull the image only if it’s not already available locally. Container creation fails if the image isn’t present and the pull fails.\n\n" +
+            "Defaults to `Always` if `:latest` tag is specified, or `IfNotPresent` otherwise.")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public String getPullPolicy() {
+        return pullPolicy;
+    }
+
+    public void setPullPolicy(String pullPolicy) {
+        this.pullPolicy = pullPolicy;
+    }
+}

api/src/main/java/io/strimzi/api/kafka/model/connect/KafkaConnectSpec.java

@@ -18,6 +18,7 @@
 import lombok.ToString;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 @DescriptionFile
@@ -28,7 +29,7 @@
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @JsonPropertyOrder({ "version", "replicas", "image", "bootstrapServers", "tls", "authentication", "config", "resources",
     "livenessProbe", "readinessProbe", "jvmOptions", "jmxOptions", "logging", "clientRackInitImage", "rack",
-    "metricsConfig", "tracing", "template", "externalConfiguration", "build" })
+    "metricsConfig", "tracing", "template", "externalConfiguration", "build", "plugins" })
 @EqualsAndHashCode(callSuper = true, doNotUseGetters = true)
 @ToString(callSuper = true)
 public class KafkaConnectSpec extends AbstractKafkaConnectSpec {
@@ -40,6 +41,7 @@ public class KafkaConnectSpec extends AbstractKafkaConnectSpec {
     private ClientTls tls;
     private KafkaClientAuthentication authentication;
     private Build build;
+    private List<MountedPlugin> plugins;
 
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     @Description("The Kafka Connect configuration. Properties with the following prefixes cannot be set: " + FORBIDDEN_PREFIXES + " (with the exception of: " + FORBIDDEN_PREFIX_EXCEPTIONS + ").")
@@ -91,4 +93,13 @@ public Build getBuild() {
     public void setBuild(Build build) {
         this.build = build;
     }
+
+    @Description("List of connector plugins to mount into the `KafkaConnect` pod.")
+    public List<MountedPlugin> getPlugins() {
+        return plugins;
+    }
+
+    public void setPlugins(List<MountedPlugin> plugins) {
+        this.plugins = plugins;
+    }
 }

api/src/main/java/io/strimzi/api/kafka/model/connect/MountedArtifact.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model.connect;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.strimzi.api.kafka.model.common.UnknownPropertyPreserving;
+import io.strimzi.crdgenerator.annotations.Description;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Abstract baseclass for different representations of mounted connector artifacts, discriminated by {@link #getType() type}.
+ */
+@JsonTypeInfo(
+        use = JsonTypeInfo.Id.NAME,
+        include = JsonTypeInfo.As.EXISTING_PROPERTY,
+        property = "type"
+)
+@JsonSubTypes(
+    {
+        @JsonSubTypes.Type(value = ImageArtifact.class, name = MountedArtifact.TYPE_IMAGE)
+    }
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@EqualsAndHashCode
+@ToString
+public abstract class MountedArtifact implements UnknownPropertyPreserving {
+    public static final String TYPE_IMAGE = "image";
+
+    private Map<String, Object> additionalProperties;
+
+    @Description("Artifact type. " +
+            "Currently, the only supported artifact type is `image`.")
+    public abstract String getType();
+
+    @Override
+    public Map<String, Object> getAdditionalProperties() {
+        return this.additionalProperties != null ? this.additionalProperties : Map.of();
+    }
+
+    @Override
+    public void setAdditionalProperty(String name, Object value) {
+        if (this.additionalProperties == null) {
+            this.additionalProperties = new HashMap<>(2);
+        }
+        this.additionalProperties.put(name, value);
+    }
+}

api/src/main/java/io/strimzi/api/kafka/model/connect/MountedPlugin.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model.connect;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.strimzi.api.kafka.model.common.Constants;
+import io.strimzi.api.kafka.model.common.UnknownPropertyPreserving;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.strimzi.crdgenerator.annotations.DescriptionFile;
+import io.strimzi.crdgenerator.annotations.Pattern;
+import io.sundr.builder.annotations.Buildable;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Represents a connector plugin in Kafka Connect (not in Kafka Connect Build)
+ */
+@Buildable(
+        editableEnabled = false,
+        builderPackage = Constants.FABRIC8_KUBERNETES_API
+)
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+@JsonPropertyOrder({ "name", "artifacts" })
+@DescriptionFile
+@EqualsAndHashCode
+@ToString
+public class MountedPlugin implements UnknownPropertyPreserving {
+    private String name;
+    private List<MountedArtifact> artifacts;
+    private Map<String, Object> additionalProperties;
+
+    @Description("A unique name for the connector plugin. " +
+            "This name is used to generate the mount path for the connector artifacts. " +
+            "The name has to be unique within the KafkaConnect resource. " +
+            "The name must be unique within the `KafkaConnect` resource and match the pattern: `^[a-z][-_a-z0-9]*[a-z]$`. " +
+            "Required")
+    @JsonProperty(required = true)
+    @Pattern("^[a-z0-9][-_a-z0-9]*[a-z0-9]$")
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    @Description("List of artifacts associated with this connector plugin. " +
+            "Required.")
+    @JsonProperty(required = true)
+    public List<MountedArtifact> getArtifacts() {
+        return artifacts;
+    }
+
+    public void setArtifacts(List<MountedArtifact> artifacts) {
+        this.artifacts = artifacts;
+    }
+
+    @Override
+    public Map<String, Object> getAdditionalProperties() {
+        return this.additionalProperties != null ? this.additionalProperties : Map.of();
+    }
+
+    @Override
+    public void setAdditionalProperty(String name, Object value) {
+        if (this.additionalProperties == null) {
+            this.additionalProperties = new HashMap<>(2);
+        }
+        this.additionalProperties.put(name, value);
+    }
+}
+