Add image tag field to streams-bootstrap app values

docs/docs/schema/defaults.json

@@ -920,6 +920,12 @@
             "additionalProperties": true,
             "description": "Settings specific to producers.",
             "properties": {
+                "imageTag": {
+                    "default": "latest",
+                    "description": "Docker image tag of the Kafka Streams app.",
+                    "title": "Imagetag",
+                    "type": "string"
+                },
                 "nameOverride": {
                     "anyOf": [
                         {
@@ -1273,6 +1279,12 @@
                     "default": null,
                     "description": "Kubernetes event-driven autoscaling config"
                 },
+                "imageTag": {
+                    "default": "latest",
+                    "description": "Docker image tag of the Kafka Streams app.",
+                    "title": "Imagetag",
+                    "type": "string"
+                },
                 "nameOverride": {
                     "anyOf": [
                         {
@@ -1328,10 +1340,10 @@
                 "app": {
                     "allOf": [
                         {
-                            "$ref": "#/$defs/HelmAppValues"
+                            "$ref": "#/$defs/StreamsBootstrapValues"
                         }
                     ],
-                    "description": "Helm app values"
+                    "description": "Streams bootstrap app values"
                 },
                 "from": {
                     "anyOf": [
@@ -1409,12 +1421,39 @@
             },
             "required": [
                 "name",
-                "namespace",
-                "app"
+                "namespace"
             ],
             "title": "StreamsBootstrap",
             "type": "object"
         },
+        "StreamsBootstrapValues": {
+            "additionalProperties": true,
+            "description": "Base value class for all streams bootstrap related components.",
+            "properties": {
+                "imageTag": {
+                    "default": "latest",
+                    "description": "Docker image tag of the Kafka Streams app.",
+                    "title": "Imagetag",
+                    "type": "string"
+                },
+                "nameOverride": {
+                    "anyOf": [
+                        {
+                            "maxLength": 63,
+                            "type": "string"
+                        },
+                        {
+                            "type": "null"
+                        }
+                    ],
+                    "default": null,
+                    "description": "Helm chart name override, assigned automatically",
+                    "title": "Nameoverride"
+                }
+            },
+            "title": "StreamsBootstrapValues",
+            "type": "object"
+        },
         "StreamsConfig": {
             "additionalProperties": true,
             "description": "Streams Bootstrap streams section.",

docs/docs/schema/pipeline.json

@@ -588,6 +588,12 @@
             "additionalProperties": true,
             "description": "Settings specific to producers.",
             "properties": {
+                "imageTag": {
+                    "default": "latest",
+                    "description": "Docker image tag of the Kafka Streams app.",
+                    "title": "Imagetag",
+                    "type": "string"
+                },
                 "nameOverride": {
                     "anyOf": [
                         {
@@ -941,6 +947,12 @@
                     "default": null,
                     "description": "Kubernetes event-driven autoscaling config"
                 },
+                "imageTag": {
+                    "default": "latest",
+                    "description": "Docker image tag of the Kafka Streams app.",
+                    "title": "Imagetag",
+                    "type": "string"
+                },
                 "nameOverride": {
                     "anyOf": [
                         {

kpops/component_handlers/kafka_connect/model.py

@@ -5,13 +5,15 @@
 from pydantic import (
     BaseModel,
     ConfigDict,
+    Field,
     SerializationInfo,
     field_validator,
     model_serializer,
 )
 from pydantic.json_schema import SkipJsonSchema
 from typing_extensions import override
 
+from kpops.component_handlers.kubernetes.utils import validate_image_tag
 from kpops.components.base_components.helm_app import HelmAppValues
 from kpops.components.base_components.models.topic import KafkaTopic, KafkaTopicStr
 from kpops.utils.pydantic import (
@@ -123,3 +125,9 @@ class KafkaConnectorResetterConfig(CamelCaseConfigModel):
 class KafkaConnectorResetterValues(HelmAppValues):
     connector_type: Literal["source", "sink"]
     config: KafkaConnectorResetterConfig
+    image_tag: str = Field(default="latest")
+
+    @pydantic.field_validator("image_tag", mode="before")
+    @classmethod
+    def validate_image_tag_field(cls, image_tag: Any) -> str:
+        return validate_image_tag(image_tag)

kpops/component_handlers/kubernetes/utils.py

@@ -1,5 +1,8 @@
 import hashlib
 import logging
+import re
+
+from kpops.api.exception import ValidationError
 
 log = logging.getLogger("K8sUtils")
 
@@ -26,3 +29,27 @@ def trim(max_len: int, name: str, suffix: str) -> str:
         )
         return new_name
     return name
+
+
+def validate_image_tag(image_tag: str) -> str:
+    """Validate an image tag.
+
+    Image tags consist of lowercase and uppercase letters, digits, underscores (_), periods (.), and dashes (-).
+    It can be up to 128 characters long and must follow the regex pattern: [a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}
+
+    :param image_tag: Docker image tag to be validated.
+    :return: The validated image tag.
+    """
+    if isinstance(image_tag, str) and is_valid_image_tag(image_tag):
+        return image_tag
+    msg = (
+        "Image tag is not valid. "
+        "Image tags consist of lowercase and uppercase letters, digits, underscores (_), periods (.), and dashes (-). "
+        "It can be up to 128 characters long."
+    )
+    raise ValidationError(msg)
+
+
+def is_valid_image_tag(image_tag: str) -> bool:
+    pattern = r"^[a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}$"
+    return bool(re.match(pattern, image_tag))

kpops/components/streams_bootstrap/__init__.py

@@ -1,26 +1,60 @@
+import logging
 from abc import ABC
+from typing import Any
 
+import pydantic
 from pydantic import Field
 
 from kpops.component_handlers.helm_wrapper.model import HelmRepoConfig
-from kpops.components.base_components.helm_app import HelmApp
+from kpops.component_handlers.kubernetes.utils import validate_image_tag
+from kpops.components.base_components.helm_app import HelmApp, HelmAppValues
 from kpops.utils.docstring import describe_attr
 
+try:
+    from typing import Self  # pyright: ignore[reportAttributeAccessIssue]
+except ImportError:
+    from typing_extensions import Self
+
+
 STREAMS_BOOTSTRAP_HELM_REPO = HelmRepoConfig(
     repository_name="bakdata-streams-bootstrap",
     url="https://bakdata.github.io/streams-bootstrap/",
 )
 STREAMS_BOOTSTRAP_VERSION = "2.9.0"
 
+log = logging.getLogger("StreamsBootstrap")
+
+
+class StreamsBootstrapValues(HelmAppValues):
+    """Base value class for all streams bootstrap related components.
+
+    :param image_tag: Docker image tag of the Kafka Streams app.
+    """
+
+    image_tag: str = Field(
+        default="latest", description=describe_attr("image_tag", __doc__)
+    )
+
+    @pydantic.field_validator("image_tag", mode="before")
+    @classmethod
+    def validate_image_tag_field(cls, image_tag: Any) -> str:
+        return validate_image_tag(image_tag)
+
 
 class StreamsBootstrap(HelmApp, ABC):
     """Base for components with a streams-bootstrap Helm chart.
 
+    :param app: Streams bootstrap app values
     :param repo_config: Configuration of the Helm chart repo to be used for
         deploying the component, defaults to streams-bootstrap Helm repo
     :param version: Helm chart version, defaults to "2.9.0"
     """
 
+    app: StreamsBootstrapValues = Field(
+        default_factory=StreamsBootstrapValues,
+        description=describe_attr("app", __doc__),
+    )
+
     repo_config: HelmRepoConfig = Field(
         default=STREAMS_BOOTSTRAP_HELM_REPO,
         description=describe_attr("repo_config", __doc__),
@@ -29,3 +63,11 @@ class StreamsBootstrap(HelmApp, ABC):
         default=STREAMS_BOOTSTRAP_VERSION,
         description=describe_attr("version", __doc__),
     )
+
+    @pydantic.model_validator(mode="after")
+    def warning_for_latest_image_tag(self) -> Self:
+        if self.validate_ and self.app.image_tag == "latest":
+            log.warning(
+                f"The image tag for component '{self.name}' is set or defaulted to 'latest'. Please, consider providing a stable image tag."
+            )
+        return self

kpops/components/streams_bootstrap/producer/model.py

@@ -4,14 +4,15 @@
     KafkaAppValues,
     KafkaStreamsConfig,
 )
+from kpops.components.streams_bootstrap import StreamsBootstrapValues
 from kpops.utils.docstring import describe_attr
 
 
 class ProducerStreamsConfig(KafkaStreamsConfig):
     """Kafka Streams settings specific to Producer."""
 
 
-class ProducerAppValues(KafkaAppValues):
+class ProducerAppValues(StreamsBootstrapValues, KafkaAppValues):
     """Settings specific to producers.
 
     :param streams: Kafka Streams settings

kpops/components/streams_bootstrap/streams/model.py

@@ -11,6 +11,7 @@
     KafkaStreamsConfig,
 )
 from kpops.components.base_components.models.topic import KafkaTopic, KafkaTopicStr
+from kpops.components.streams_bootstrap import StreamsBootstrapValues
 from kpops.utils.docstring import describe_attr
 from kpops.utils.pydantic import (
     CamelCaseConfigModel,
@@ -237,7 +238,7 @@ def validate_mandatory_fields_are_set(
         return self
 
 
-class StreamsAppValues(KafkaAppValues):
+class StreamsAppValues(StreamsBootstrapValues, KafkaAppValues):
     """streams-bootstrap app configurations.
 
     The attributes correspond to keys and values that are used as values for the streams bootstrap helm chart.

tests/component_handlers/kubernetes/test_utils.py

@@ -0,0 +1,20 @@
+import pytest
+
+from kpops.api.exception import ValidationError
+from kpops.component_handlers.kubernetes.utils import validate_image_tag
+
+
+def test_is_valid_image_tag():
+    assert validate_image_tag("1.2.3") == "1.2.3"
+    assert validate_image_tag("123") == "123"
+    assert validate_image_tag("1_2_3") == "1_2_3"
+    assert validate_image_tag("1-2-3") == "1-2-3"
+    assert validate_image_tag("latest") == "latest"
+    assert (
+        validate_image_tag(
+            "1ff6c18fbef2045af6b9c16bf034cc421a29027b800e4f9b68ae9b1cb3e9ae07"
+        )
+        == "1ff6c18fbef2045af6b9c16bf034cc421a29027b800e4f9b68ae9b1cb3e9ae07"
+    )
+    with pytest.raises(ValidationError):
+        assert validate_image_tag("la!est") is False

tests/components/test_kafka_sink_connector.py

@@ -107,14 +107,6 @@ def test_connector_config_parsing(
         handlers: ComponentHandlers,
         connector_config: KafkaConnectorConfig,
     ):
-        connector = KafkaSinkConnector(
-            name=CONNECTOR_NAME,
-            config=config,
-            handlers=handlers,
-            app=connector_config,
-            resetter_namespace=RESETTER_NAMESPACE,
-        )
-
         topic_pattern = ".*"
         connector = KafkaSinkConnector(
             name=CONNECTOR_NAME,

tests/components/test_streams_bootstrap.py

@@ -48,6 +48,7 @@ def test_default_configs(self, config: KpopsConfig, handlers: ComponentHandlers)
         )
         assert streams_bootstrap.version == "2.9.0"
         assert streams_bootstrap.namespace == "test-namespace"
+        assert streams_bootstrap.app.image_tag == "latest"
 
     @pytest.mark.asyncio()
     async def test_should_deploy_streams_bootstrap_app(
@@ -63,6 +64,7 @@ async def test_should_deploy_streams_bootstrap_app(
             **{
                 "namespace": "test-namespace",
                 "app": {
+                    "imageTag": "1.0.0",
                     "streams": {
                         "outputTopic": "test",
                         "brokers": "fake-broker:9092",
@@ -94,6 +96,7 @@ async def test_should_deploy_streams_bootstrap_app(
             "test-namespace",
             {
                 "nameOverride": "${pipeline.name}-example-name",
+                "imageTag": "1.0.0",
                 "streams": {
                     "brokers": "fake-broker:9092",
                     "outputTopic": "test",

tests/pipeline/resources/resetter_values/defaults.yaml

@@ -9,3 +9,5 @@ helm-app:
 kafka-sink-connector:
   app:
     "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector"
+  resetter_values:
+    imageTag: override-default-image-tag

tests/pipeline/snapshots/test_example/test_generate/atm-fraud/pipeline.yaml

@@ -393,6 +393,7 @@
         brokers: http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092
         connector: atm-fraud-postgresql-connector
       connectorType: sink
+      imageTag: latest
     name: postgresql-connector
     namespace: ${NAMESPACE}
     prefix: atm-fraud-

tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml

@@ -142,6 +142,7 @@
         brokers: http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092
         connector: word-count-redis-sink-connector
       connectorType: sink
+      imageTag: latest
     name: redis-sink-connector
     namespace: ${NAMESPACE}
     prefix: word-count-

tests/pipeline/snapshots/test_generate/test_default_config/pipeline.yaml

@@ -1,5 +1,6 @@
 - _cleaner:
     app:
+      imageTag: latest
       resources:
         limits:
           memory: 2G
@@ -21,6 +22,7 @@
     type: producer-app-cleaner
     version: 2.9.0
   app:
+    imageTag: latest
     resources:
       limits:
         memory: 2G
@@ -50,6 +52,7 @@
 - _cleaner:
     app:
       image: some-image
+      imageTag: latest
       labels:
         pipeline: resources-custom-config
       persistence:
@@ -77,6 +80,7 @@
     version: 2.9.0
   app:
     image: some-image
+    imageTag: latest
     labels:
       pipeline: resources-custom-config
     persistence: