A Kafka Support (GSI-933)

.devcontainer/.dev_config.yaml

@@ -20,3 +20,5 @@ object_storages:
       s3_endpoint_url: http://localstack:4566
       s3_access_key_id: test
       s3_secret_access_key: test
+
+kafka_servers: ["kafka:9092"]

.devcontainer/docker-compose.yml

@@ -13,7 +13,10 @@ services:
         USER_GID: 1000
 
     init: true
-    privileged: true
+
+    # Makes testcontainers work on linux based hosts
+    extra_hosts:
+      - host.docker.internal:host-gateway
 
     volumes:
       - ..:/workspace:cached

.pre-commit-config.yaml

@@ -45,7 +45,7 @@ repos:
       - id: no-commit-to-branch
         args: [--branch, dev, --branch, int, --branch, main]
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.5.6
+    rev: v0.6.2
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]

.pyproject_generation/pyproject_custom.toml

@@ -1,11 +1,11 @@
 [project]
 name = "sms"
-version = "1.2.0"
+version = "1.3.0"
 description = "State Management Service - Provides a REST API for basic infrastructure technology state management."
 dependencies = [
     "typer >= 0.12",
     "ghga-service-commons[api] >= 3.1",
-    "hexkit[mongodb,s3] >= 3.5",
+    "hexkit[mongodb,s3,akafka] >= 3.5",
 ]
 
 [project.urls]

README.md

@@ -24,21 +24,21 @@ We recommend using the provided Docker container.
 
 A pre-build version is available at [docker hub](https://hub.docker.com/repository/docker/ghga/state-management-service):
 ```bash
-docker pull ghga/state-management-service:1.2.0
+docker pull ghga/state-management-service:1.3.0
 ```
 
 Or you can build the container yourself from the [`./Dockerfile`](./Dockerfile):
 ```bash
 # Execute in the repo's root dir:
-docker build -t ghga/state-management-service:1.2.0 .
+docker build -t ghga/state-management-service:1.3.0 .
 ```
 
 For production-ready deployment, we recommend using Kubernetes, however,
 for simple use cases, you could execute the service using docker
 on a single server:
 ```bash
 # The entrypoint is preconfigured:
-docker run -p 8080:8080 ghga/state-management-service:1.2.0 --help
+docker run -p 8080:8080 ghga/state-management-service:1.3.0 --help
 ```
 
 If you prefer not to use containers, you may install the service from source:
@@ -55,6 +55,72 @@ sms --help
 ### Parameters
 
 The service requires the following configuration parameters:
+- **`service_name`** *(string)*: Short name of this service. Default: `"sms"`.
+
+- **`service_instance_id`** *(string, required)*: A string that uniquely identifies this instance across all instances of this service. This is included in log messages.
+
+
+  Examples:
+
+  ```json
+  "germany-bw-instance-001"
+  ```
+
+
+- **`kafka_servers`** *(array, required)*: A list of connection strings to connect to Kafka bootstrap servers.
+
+  - **Items** *(string)*
+
+
+  Examples:
+
+  ```json
+  [
+      "localhost:9092"
+  ]
+  ```
+
+
+- **`kafka_security_protocol`** *(string)*: Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL. Must be one of: `["PLAINTEXT", "SSL"]`. Default: `"PLAINTEXT"`.
+
+- **`kafka_ssl_cafile`** *(string)*: Certificate Authority file path containing certificates used to sign broker certificates. If a CA is not specified, the default system CA will be used if found by OpenSSL. Default: `""`.
+
+- **`kafka_ssl_certfile`** *(string)*: Optional filename of client certificate, as well as any CA certificates needed to establish the certificate's authenticity. Default: `""`.
+
+- **`kafka_ssl_keyfile`** *(string)*: Optional filename containing the client private key. Default: `""`.
+
+- **`kafka_ssl_password`** *(string, format: password)*: Optional password to be used for the client private key. Default: `""`.
+
+- **`generate_correlation_id`** *(boolean)*: A flag, which, if False, will result in an error when inbound requests don't possess a correlation ID. If True, requests without a correlation ID will be assigned a newly generated ID in the correlation ID middleware function. Default: `true`.
+
+
+  Examples:
+
+  ```json
+  true
+  ```
+
+
+  ```json
+  false
+  ```
+
+
+- **`kafka_max_message_size`** *(integer)*: The largest message size that can be transmitted, in bytes. Only services that have a need to send/receive larger messages should set this. Exclusive minimum: `0`. Default: `1048576`.
+
+
+  Examples:
+
+  ```json
+  1048576
+  ```
+
+
+  ```json
+  16777216
+  ```
+
+
 - **`object_storages`** *(object, required)*: Can contain additional properties.
 
   - **Additional properties**: Refer to *[#/$defs/S3ObjectStorageNodeConfig](#%24defs/S3ObjectStorageNodeConfig)*.
@@ -155,18 +221,6 @@ The service requires the following configuration parameters:
 
 - **`log_level`** *(string)*: The minimum log level to capture. Must be one of: `["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "TRACE"]`. Default: `"INFO"`.
 
-- **`service_name`** *(string)*: Short name of this service. Default: `"sms"`.
-
-- **`service_instance_id`** *(string, required)*: A string that uniquely identifies this instance across all instances of this service. This is included in log messages.
-
-
-  Examples:
-
-  ```json
-  "germany-bw-instance-001"
-  ```
-
-
 - **`log_format`**: If set, will replace JSON formatting with the specified string format. If not set, has no effect. In addition to the standard attributes, the following can also be specified: timestamp, service, instance, level, correlation_id, and details. Default: `null`.
 
   - **Any of**
@@ -282,21 +336,6 @@ The service requires the following configuration parameters:
   ```
 
 
-- **`generate_correlation_id`** *(boolean)*: A flag, which, if False, will result in an error when inbound requests don't possess a correlation ID. If True, requests without a correlation ID will be assigned a newly generated ID in the correlation ID middleware function. Default: `true`.
-
-
-  Examples:
-
-  ```json
-  true
-  ```
-
-
-  ```json
-  false
-  ```
-
-
 ## Definitions
 
 

config_schema.json

@@ -97,6 +97,90 @@
   "additionalProperties": false,
   "description": "Modifies the original Settings class provided by the user",
   "properties": {
+    "service_name": {
+      "default": "sms",
+      "description": "Short name of this service",
+      "title": "Service Name",
+      "type": "string"
+    },
+    "service_instance_id": {
+      "description": "A string that uniquely identifies this instance across all instances of this service. This is included in log messages.",
+      "examples": [
+        "germany-bw-instance-001"
+      ],
+      "title": "Service Instance Id",
+      "type": "string"
+    },
+    "kafka_servers": {
+      "description": "A list of connection strings to connect to Kafka bootstrap servers.",
+      "examples": [
+        [
+          "localhost:9092"
+        ]
+      ],
+      "items": {
+        "type": "string"
+      },
+      "title": "Kafka Servers",
+      "type": "array"
+    },
+    "kafka_security_protocol": {
+      "default": "PLAINTEXT",
+      "description": "Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL.",
+      "enum": [
+        "PLAINTEXT",
+        "SSL"
+      ],
+      "title": "Kafka Security Protocol",
+      "type": "string"
+    },
+    "kafka_ssl_cafile": {
+      "default": "",
+      "description": "Certificate Authority file path containing certificates used to sign broker certificates. If a CA is not specified, the default system CA will be used if found by OpenSSL.",
+      "title": "Kafka Ssl Cafile",
+      "type": "string"
+    },
+    "kafka_ssl_certfile": {
+      "default": "",
+      "description": "Optional filename of client certificate, as well as any CA certificates needed to establish the certificate's authenticity.",
+      "title": "Kafka Ssl Certfile",
+      "type": "string"
+    },
+    "kafka_ssl_keyfile": {
+      "default": "",
+      "description": "Optional filename containing the client private key.",
+      "title": "Kafka Ssl Keyfile",
+      "type": "string"
+    },
+    "kafka_ssl_password": {
+      "default": "",
+      "description": "Optional password to be used for the client private key.",
+      "format": "password",
+      "title": "Kafka Ssl Password",
+      "type": "string",
+      "writeOnly": true
+    },
+    "generate_correlation_id": {
+      "default": true,
+      "description": "A flag, which, if False, will result in an error when inbound requests don't possess a correlation ID. If True, requests without a correlation ID will be assigned a newly generated ID in the correlation ID middleware function.",
+      "examples": [
+        true,
+        false
+      ],
+      "title": "Generate Correlation Id",
+      "type": "boolean"
+    },
+    "kafka_max_message_size": {
+      "default": 1048576,
+      "description": "The largest message size that can be transmitted, in bytes. Only services that have a need to send/receive larger messages should set this.",
+      "examples": [
+        1048576,
+        16777216
+      ],
+      "exclusiveMinimum": 0,
+      "title": "Kafka Max Message Size",
+      "type": "integer"
+    },
     "object_storages": {
       "additionalProperties": {
         "$ref": "#/$defs/S3ObjectStorageNodeConfig"
@@ -176,20 +260,6 @@
       "title": "Log Level",
       "type": "string"
     },
-    "service_name": {
-      "default": "sms",
-      "description": "Short name of this service",
-      "title": "Service Name",
-      "type": "string"
-    },
-    "service_instance_id": {
-      "description": "A string that uniquely identifies this instance across all instances of this service. This is included in log messages.",
-      "examples": [
-        "germany-bw-instance-001"
-      ],
-      "title": "Service Instance Id",
-      "type": "string"
-    },
     "log_format": {
       "anyOf": [
         {
@@ -335,24 +405,15 @@
         []
       ],
       "title": "Cors Allowed Headers"
-    },
-    "generate_correlation_id": {
-      "default": true,
-      "description": "A flag, which, if False, will result in an error when inbound requests don't possess a correlation ID. If True, requests without a correlation ID will be assigned a newly generated ID in the correlation ID middleware function.",
-      "examples": [
-        true,
-        false
-      ],
-      "title": "Generate Correlation Id",
-      "type": "boolean"
     }
   },
   "required": [
+    "service_instance_id",
+    "kafka_servers",
     "object_storages",
     "token_hashes",
     "db_prefix",
-    "db_connection_str",
-    "service_instance_id"
+    "db_connection_str"
   ],
   "title": "ModSettings",
   "type": "object"

example_config.yaml

@@ -12,6 +12,14 @@ db_prefix: test_
 docs_url: /docs
 generate_correlation_id: true
 host: 127.0.0.1
+kafka_max_message_size: 1048576
+kafka_security_protocol: PLAINTEXT
+kafka_servers:
+- kafka:9092
+kafka_ssl_cafile: ''
+kafka_ssl_certfile: ''
+kafka_ssl_keyfile: ''
+kafka_ssl_password: ''
 log_format: null
 log_level: INFO
 log_traceback: true