Merge branch 'docker-compose-enterprise' into 'enterprise'

feat: Deploy using docker compose (PR 2/2)

See merge request dkinternal/observability/dataops-observability!57

Author: ci bot

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/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()

deploy/docker-compose/README.md

@@ -0,0 +1,79 @@
+# DataOps Observability Docker Compose
+
+## Overview
+
+This document describes how to use Docker Compose to deploy Observability. The provided compose
+file can be used without modifications to deploy to a local, trial docker environment, or can
+be modified to guide a cloud-based deployment.
+
+> [!IMPORTANT]
+> For your convenience, the provided compose file has the services (database and message broker)
+> connectivity configuration at their most simplistic form, which is not secure for a production deployment.
+
+## Local deployment
+
+The following command will start the platform, using its default configurations. All the commands
+mentioned in this document assumes Docker is installed and the `compose.yaml` and `config.env` files are within the
+current folder
+
+```shell
+docker compose up -d
+```
+
+When the command returns successfully, it means that the platform in running in the background, and you can proceed to
+seeding the minimal data it needs, which includes creating a user account. The following command will prompt you
+for the user account information.
+
+```shell
+docker compose run --rm -it observability_data_init init --data
+```
+
+Once that is finished, you can access the platform using the credentials you created at http://localhost:8082/
+
+The data initialization also provides some information that you must have available if you are planning to
+interact with the platform using its integration APIs. Store this information somewhere, if that is the case. This
+is an example of what this information will look like:
+
+```
+INFO:               company_id: 6a6e5aa0-76cb-45d9-b1f5-cad3ebaf1cb4
+INFO:          organization_id: 6cfa3063-5220-4061-8004-df5b32cd54a2
+INFO:               project_id: b85ebf07-fa4f-4cc8-9362-d78ec800e2c0
+INFO:                  user_id: f5c30e55-d2e2-4b1f-baec-9901f56060e0
+```
+
+Your DataOps Observability platform is deployed and ready.
+
+Once you are done, you can stop the platform without losing your data, by issuing the command below. You
+can start and stop it at any time.
+
+```shell
+docker compose down
+```
+
+## Configurations
+
+The Docker Compose file allows you to configure certain aspects of the deployment. The `config.env` file serves as a
+reference of which environment variables are considered, and can be edited as needed. In that case, most of the
+"docker compose" commands will need an argument pointing to this file to be included, as the example below.
+
+```shell
+docker compose --env-file=conig.env up -d
+```
+
+## Upgrading
+
+Upgrading Observability to its latest version only requires the new images to be pulled, and starting fresh containers
+running the updated images. The following command pulls the images. Note the `--policy=always` argument. It forces
+pulling even when the images are already present.
+
+```shell
+docker compose pull --policy=always
+```
+
+Next, you will need to restart the application containers, which can be achieved with the same installation command.
+The database migrations, if any, will automatically run during the restart. It can be executed whether the
+application is running or not.
+
+```
+docker compose up -d
+```

deploy/docker-compose/compose.yaml

@@ -1,7 +1,8 @@
+name: dataops-observability
 
 x-database-config: &database_config
   MYSQL_USER: ${MYSQL_USER:-observability}
-  MYSQL_PASSWORD: ${MYSQL_PASSWORD}
+  MYSQL_PASSWORD: ${MYSQL_PASSWORD:-193d036391fa7f14e3c48cabca7e47d0}
 
 x-database-client-config: &database_client_config
   MYSQL_SERVICE_HOST: ${MYSQL_HOST:-database}
@@ -28,10 +29,12 @@ services:
       # Transactions
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      # Topics
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       # Other
       KAFKA_LOG4J_LOGGERS: "kafka=WARN,org.apache.kafka=WARN"
     volumes:
-      - /var/lib/kafka/data
+      - kafka_data:/var/lib/kafka/data
     healthcheck:
       test: [ "CMD", "/opt/kafka/bin/kafka-topics.sh", "--bootstrap-server", "localhost:9092", "--list" ]
       interval: 10s
@@ -44,18 +47,20 @@ services:
     restart: always
     expose: ["3306"]
     environment:
-      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD}
+      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD:-1824f41cee3033a881e1888a8fea24a3}
       MYSQL_DATABASE: datakitchen
       <<: *database_config
+    volumes:
+      - mysql_data:/var/lib/mysql
+
     healthcheck:
       test: [ "CMD", "mysqladmin", "ping", "-h", "localhost" ]
       interval: 5s
       retries: 5
 
   observability_data_init:
     container_name: data-init
-    image: datakitchen/dataops-observability-be
-    pull_policy: never
+    image: ${DK_OBSERVABILITY_BE_IMAGE:-datakitchen/dataops-observability-be:v2}
     restart: on-failure
     depends_on:
       database:
@@ -65,12 +70,12 @@ services:
     environment:
       OBSERVABILITY_CONFIG: minikube
       <<: [*database_config, *database_client_config]
-    entrypoint: /bin/sh -c '/dk/bin/cli migrate'
+    entrypoint: /dk/bin/cli
+    command:  migrate
 
   observability_backend:
     container_name: back-end
-    image: datakitchen/dataops-observability-be
-    pull_policy: never
+    image: ${DK_OBSERVABILITY_BE_IMAGE:-datakitchen/dataops-observability-be:v2}
     restart: always
     depends_on:
       observability_data_init:
@@ -85,30 +90,31 @@ services:
       test: [ "CMD", "/bin/sh", "-c", "supervisorctl -c /dk/supervisord.conf status | grep -q RUNNING" ]
       interval: 5s
       retries: 10
-    volumes:
-      - type: bind
-        source: ../conf/supervisord.conf
-        target: /dk/supervisord.conf
 
   observability_ui:
     container_name: user-interface
-    image: datakitchen/dataops-observability-ui
-    pull_policy: never
+    image: ${DK_OBSERVABILITY_UI_IMAGE:-datakitchen/dataops-observability-ui:v2}
     restart: always
     depends_on:
       observability_backend:
         condition: service_healthy
     environment:
-      OBSERVABILITY_API_HOSTNAME: ${OBSERVABILITY_API_HOSTNAME}
-      OBSERVABILITY_CSP_EXTRA: ${observability_csp_extra}
+      OBSERVABILITY_API_HOSTNAME: ${DK_OBSERVABILITY_API_HOSTNAME:-}
+      OBSERVABILITY_CSP_EXTRA: ${DK_OBSERVABILITY_CSP_EXTRA:-}
+      OBSERVABILITY_API_BASE_URL: ${DK_OBSERVABILITY_API_BASE_URL:-}
+      OBSERVABILITY_AUTH_METHOD: ${DK_OBSERVABILITY_AUTH_METHOD:-basic}
     links:
       - "observability_backend:observability-api"
       - "observability_backend:event-api"
       - "observability_backend:agent-api"
     ports:
-      - "8080:8082"
+      - "${DK_OBSERVABILITY_HTTP_PORT:-8082}:8082"
 
 networks:
   datakitchen:
     name: datakitchen-network
-    external: true
+    external: true
+
+volumes:
+  mysql_data:
+  kafka_data: