Merge pull request #35 from DataKitchen/release/2.8.0

Release/2.8.0

Author: datakitchen-devops

cli/entry_points/init.py

@@ -9,6 +9,13 @@
 from common.auth.keys.lib import hash_value
 from common.auth.keys.service_key import generate_key
 from common.entities import DB, Action, AuthProvider, Company, Organization, Project, User
+from common.kafka import (
+    TOPIC_IDENTIFIED_EVENTS,
+    TOPIC_UNIDENTIFIED_EVENTS,
+    TOPIC_SCHEDULED_EVENTS,
+    TOPIC_DEAD_LETTER_OFFICE,
+)
+from common.kafka.admin import create_topics
 from common.model import create_all_tables
 
 USER_FIELDS = ["name", "email", "username", "password"]
@@ -89,10 +96,19 @@ def args(parser: ArgumentParser) -> None:
             action="store_true",
             help="Outputs the generated IDs in JSON format when successful",
         )
+        parser.add_argument(
+            "-k",
+            "--topics",
+            action="store_true",
+            help="Create the Kafka Topics",
+        )
 
     def subcmd_entry_point(self) -> None:
         try:
+            if not any([self.kwargs.get(arg) for arg in ("data", "demo", "tables", "topics")]):
+                raise OperationAborted("Either --data or --demo or --tables or --topics has to be set.")
             self.initialize_database()
+            self.create_kafka_topics()
         except OperationAborted as e:
             LOG.info("Operation #y<ABORTED>: %s", e)
             sys.exit(1)
@@ -102,10 +118,14 @@ def subcmd_entry_point(self) -> None:
         else:
             LOG.info("Operation #g<SUCCEEDED>")
 
-    def initialize_database(self) -> None:
-        if not (self.kwargs.get("tables") or self.kwargs.get("data") or self.kwargs.get("demo")):
-            raise OperationAborted("Either --data or --demo or --tables has to be set.")
+    def create_kafka_topics(self) -> None:
+        if self.kwargs.get("topics"):
+            LOG.info("#c<Creating Kafka topics...>")
+            create_topics(
+                [TOPIC_IDENTIFIED_EVENTS, TOPIC_UNIDENTIFIED_EVENTS, TOPIC_SCHEDULED_EVENTS, TOPIC_DEAD_LETTER_OFFICE]
+            )
 
+    def initialize_database(self) -> None:
         if self.kwargs.get("tables"):
             if self.kwargs.get("force"):
                 raise OperationAborted("The --force option can not be used when creating the tables.")

cli/tests/entry_points/test_init.py

@@ -15,6 +15,12 @@ def create_tables_mock():
         yield create_tables_mock
 
 
+@pytest.fixture
+def create_topics_mock():
+    with patch("cli.entry_points.init.create_topics") as mock:
+        yield mock
+
+
 @pytest.fixture
 def init_entry_point(test_db, create_tables_mock):
     yield Initialize(default=True)
@@ -124,6 +130,16 @@ def test_init_not_empty(arg, init_entry_point, create_tables_mock, mock_user_inp
     assert ServiceAccountKey.select().count() == demo_data_count
 
 
+@pytest.mark.integration
+def test_init_topics(init_entry_point, create_topics_mock):
+    with (
+        patch.dict(init_entry_point.kwargs, {"topics": True}),
+    ):
+        init_entry_point.subcmd_entry_point()
+
+    create_topics_mock.assert_called_once()
+
+
 @pytest.mark.integration
 def test_init_error(init_entry_point, mock_user_input):
     DB.create_tables(ALL_MODELS)

common/kafka/admin.py

@@ -0,0 +1,29 @@
+import logging
+
+from confluent_kafka.admin import AdminClient, NewTopic
+from confluent_kafka import KafkaException, KafkaError
+
+from common.kafka.topic import Topic
+from conf import settings
+
+LOG = logging.getLogger(__name__)
+
+
+def create_topics(topics: list[Topic], num_partitions: int = 1, replication_factor: int = 1) -> None:
+    client = AdminClient(settings.KAFKA_CONNECTION_PARAMS)
+    new_topics = [
+        NewTopic(topic.name, num_partitions=num_partitions, replication_factor=replication_factor) for topic in topics
+    ]
+    failed_topics = []
+    for topic_name, future in client.create_topics(new_topics).items():
+        try:
+            future.result()
+        except KafkaException as e:
+            if e.args[0] != KafkaError.TOPIC_ALREADY_EXISTS:
+                failed_topics.append(topic_name)
+        except Exception:
+            LOG.exception("Error creating %s Kafka topic", topic_name)
+            failed_topics.append(topic_name)
+
+    if failed_topics:
+        raise Exception(f"Creating the topics {", ".join(failed_topics)} failed.")

common/kubernetes/readiness_probe.py

@@ -14,13 +14,14 @@
 import time
 from argparse import ArgumentParser, Namespace
 from contextlib import contextmanager
+from pathlib import Path
 from typing import NoReturn
 from collections.abc import Generator
 from collections.abc import Callable
 
 LOG = logging.getLogger(__name__)
 
-FILE_NAME = "observability_readyz"
+FILE_NAME = "{service}_readyz"
 """Name of the file that will be created into the temp dir to mark the service as ready."""
 
 POLLING_SLEEP_SECONDS = 0.1
@@ -34,19 +35,26 @@ class NotReadyException(Exception):
     """Raised when the service shouldn't be marked as ready."""
 
 
-def _get_mark_file_path() -> str:
-    return os.path.join(tempfile.gettempdir(), FILE_NAME)
+def _get_mark_file_path() -> Path:
+    file_name = FILE_NAME.format(service=os.getenv("SUPERVISOR_PROCESS_NAME", "observability"))
+    return Path(tempfile.gettempdir()) / file_name
 
 
 def set_ready() -> None:
     """Mark the service as ready by adding a PID file to the temp dir."""
-    with open(_get_mark_file_path(), "w") as mark_file:
-        mark_file.write(str(os.getpid()))
+    mark_file_path = _get_mark_file_path()
+    mark_file_path.parent.mkdir(parents=True, exist_ok=True)
+    mark_file_path.write_text(str(os.getpid()))
 
 
 def is_ready() -> bool:
     """Checks the existence of the readiness mark file."""
-    return os.path.exists(_get_mark_file_path())
+    try:
+        os.kill(int(_get_mark_file_path().read_text().strip()), 0)
+    except (FileNotFoundError, ProcessLookupError, ValueError):
+        return False
+    else:
+        return True
 
 
 @contextmanager

common/tests/unit/kafka/test_admin.py

@@ -0,0 +1,62 @@
+from unittest.mock import patch, Mock
+
+import pytest
+from confluent_kafka.cimpl import NewTopic, KafkaException, KafkaError
+
+from common.kafka.admin import create_topics
+from common.kafka.topic import Topic
+
+
+@pytest.fixture
+def client_class_mock():
+    with patch("common.kafka.admin.AdminClient") as mock:
+        yield mock
+
+
+@pytest.fixture
+def client_mock(client_class_mock):
+    mock = Mock()
+    client_class_mock.return_value = mock
+    yield mock
+
+
+@pytest.fixture
+def create_topics_mock(client_mock):
+    yield client_mock.create_topics
+
+
+@pytest.fixture
+def result_mock(create_topics_mock):
+    future_mock = Mock()
+    result_mock = future_mock.result
+    create_topics_mock.return_value = {"TEST_TOPIC": future_mock}
+    yield result_mock
+
+
+@pytest.fixture
+def topic():
+    return Topic(name="TEST_TOPIC")
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("result_side_effect", (lambda: None, KafkaException(KafkaError.TOPIC_ALREADY_EXISTS)))
+def test_create_topics(result_side_effect, client_class_mock, create_topics_mock, result_mock, topic):
+    result_mock.side_effect = result_side_effect
+
+    create_topics([topic], num_partitions=5, replication_factor=5)
+
+    client_class_mock.assert_called_once()
+    create_topics_mock.assert_called_once_with([NewTopic("TEST_TOPIC", num_partitions=5, replication_factor=5)])
+    result_mock.assert_called_once()
+
+
+@pytest.mark.unit
+def test_create_topics_fail(client_class_mock, create_topics_mock, result_mock, topic):
+    result_mock.side_effect = KafkaException(KafkaError.MEMBER_ID_REQUIRED)
+
+    with pytest.raises(Exception):
+        create_topics([topic], num_partitions=5, replication_factor=5)
+
+    client_class_mock.assert_called_once()
+    create_topics_mock.assert_called_once_with([NewTopic("TEST_TOPIC", num_partitions=5, replication_factor=5)])
+    result_mock.assert_called_once()

common/tests/unit/kubernetes/test_readiness_probe.py

@@ -1,4 +1,6 @@
+import os
 from itertools import count
+from pathlib import Path
 from unittest.mock import patch
 
 import pytest
@@ -18,8 +20,16 @@ def set_ready_mock():
 
 @pytest.mark.unit
 @patch.object(readiness_probe.tempfile, "gettempdir", return_value="/xpto")
-def test_get_filename(temp_dir_mock):
-    assert readiness_probe._get_mark_file_path() == "/xpto/observability_readyz"
+@pytest.mark.parametrize(
+    "env, expected",
+    (
+        ({}, Path("/xpto/observability_readyz")),
+        ({"SUPERVISOR_PROCESS_NAME": "a_service"}, Path("/xpto/a_service_readyz")),
+    ),
+)
+def test_get_filename(temp_dir_mock, env, expected):
+    with patch.dict(os.environ, env):
+        assert readiness_probe._get_mark_file_path() == expected
     temp_dir_mock.assert_called_once()
 
 

conf/defaults.py

@@ -1,7 +1,6 @@
 RULE_REFRESH_SECONDS: int = 30
 """Number of seconds to cache rules in rules engine"""
 
-
 MIGRATIONS_SRC_PATH = "/dk/lib/migrations"
 """Yoyo migrations source folder."""
 

deploy/charts/observability-app/Chart.yaml

@@ -2,7 +2,7 @@ apiVersion: v2
 name: dataops-observability-app
 type: application
 appVersion: "2.x.x"
-version: "2.3.1"
+version: "2.4.0"
 
 description: DataOps Observability
 home: https://datakitchen.io

deploy/charts/observability-app/templates/observability-ui.yaml

@@ -30,21 +30,6 @@ spec:
       serviceAccountName: {{ include "observability.serviceAccountName" . }}
       securityContext:
         {{- toYaml .Values.observability_ui.podSecurityContext | nindent 8 }}
-      {{- if or .Values.observability_ui.environmentJson .Values.observability_ui.manifestJson }}
-      volumes:
-        - name: {{ include "observability.observability_ui.name" . }}-configmap-volume
-          configMap:
-            name: {{ include "observability.observability_ui.name" . }}-configmap
-            items:
-            {{- if .Values.observability_ui.environmentJson }}
-              - key: "environment.json"
-                path: "environment.json"
-            {{- end }}
-            {{- if .Values.observability_ui.manifestJson }}
-              - key: "module-federation.manifest.json"
-                path: "module-federation.manifest.json"
-            {{- end }}
-      {{- end }}
       containers:
         - name: {{ include "observability.observability_ui.name" . }}
           securityContext:
@@ -58,31 +43,21 @@ spec:
           resources:
             {{- toYaml .Values.observability_ui.resources | nindent 12 }}
           env:
-            {{- if .Values.observability_api.hostname }}
+            - name: OBSERVABILITY_API_BASE_URL
+              value: {{ .Values.observability_ui.api_base_url | quote }}
+            - name: OBSERVABILITY_AUTH_METHOD
+              value: {{ .Values.observability_ui.auth_method | quote }}
+            {{- if .Values.observability_ui.hostname }}
             - name: OBSERVABILITY_API_HOSTNAME
               value: {{ tpl .Values.observability_api.hostname . | quote }}
             {{- end }}
             {{- if not (quote .Values.observability_ui.csp_extra | empty) }}
             - name: OBSERVABILITY_CSP_EXTRA
               value: {{ tpl .Values.observability_ui.csp_extra . | quote }}
             {{- end }}
-          {{- if or .Values.observability_ui.environmentJson .Values.observability_ui.manifestJson .Values.extraVolumeMounts }}
+          {{- with .Values.extraVolumeMounts }}
           volumeMounts:
-            {{- if .Values.observability_ui.environmentJson }}
-            - mountPath: /observability_ui/shell/environments/environment.json
-              name: {{ include "observability.observability_ui.name" . }}-configmap-volume
-              readOnly: true
-              subPath: environment.json
-            {{- end }}
-            {{- if .Values.observability_ui.manifestJson }}
-            - mountPath: /observability_ui/shell/assets/module-federation.manifest.json
-              name: {{ include "observability.observability_ui.name" . }}-configmap-volume
-              readOnly: true
-              subPath: module-federation.manifest.json
-            {{- end }}
-            {{- with .Values.extraVolumeMounts }}
             {{ toYaml . | nindent 12 }}
-            {{- end }}
           {{- end }}
       {{- with .Values.extraVolumes }}
       volumes:
@@ -122,21 +97,4 @@ spec:
       {{- end }}
   selector:
     {{- include "observability.observability_ui.selectorLabels" . | nindent 4 }}
-{{- if or .Values.observability_ui.environmentJson .Values.observability_ui.manifestJson }}
----
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: {{ include "observability.observability_ui.name" . }}-configmap
-  labels:
-    {{- include "observability.labels" . | nindent 4 }}
-    {{- include "observability.observability_ui.selectorLabels" . | nindent 4 }}
-data:
-  {{- if .Values.observability_ui.environmentJson }}
-  environment.json: {{ .Values.observability_ui.environmentJson | toPrettyJson | quote }}
-  {{- end }}
-  {{- if .Values.observability_ui.manifestJson }}
-  module-federation.manifest.json: {{ .Values.observability_ui.manifestJson | toPrettyJson | quote }}
-  {{- end }}
-{{- end }}
 {{- end }}

deploy/charts/observability-app/values.yaml

@@ -24,6 +24,8 @@ observability_ui:
     type: NodePort
     port: 8082
     nodePort: 8082
+  api_base_url: "/api"
+  auth_method: "sso"
   podSecurityContext:
     sysctls:
       - name: net.core.somaxconn
@@ -32,7 +34,6 @@ observability_ui:
   nodeSelector: { }
   tolerations: [ ]
   affinity: { }
-  environmentJson: {"apiBaseUrl": "/api"}
 
 observability_api:
   enable: true