Kafka deployment script in gitlab-ci.file

- In Kafka.variables files: get_kafka_address() and get_admin_client() is added.
- In KpiValueApiServerImpl Kafka Admin Client call is updated.
- Kafka deployment script is added.

scripts/run_tests_locally-kpi-value-API.sh

@@ -19,7 +19,8 @@ PROJECTDIR=`pwd`
 cd $PROJECTDIR/src
 
 RCFILE=$PROJECTDIR/coverage/.coveragerc
-
+KAFKA_IP=$(docker inspect kafka --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}")
+KFK_SERVER_ADDRESS=${KAFKA_IP}:9092
 # helpful pytest flags: --log-level=INFO -o log_cli=true --verbose --maxfail=1 --durations=0
 python3 -m pytest --log-level=DEBUG --log-cli-level=DEBUG -o log_cli=true --verbose \
     kpi_value_api/tests/test_kpi_value_api.py

src/common/tools/kafka/Variables.py

@@ -20,16 +20,27 @@ from common.Settings import get_setting
 
 
 LOGGER = logging.getLogger(__name__)
+KFK_SERVER_ADDRESS_TEMPLATE = 'kafka-service.{:s}.svc.cluster.local:{:s}'
 
 class KafkaConfig(Enum):
-    KFK_SERVER_ADDRESS_TEMPLATE = 'kafka-service.{:s}.svc.cluster.local:{:s}'
-    KFK_NAMESPACE               = 'kafka'
-    # KFK_NAMESPACE               = get_setting('KFK_NAMESPACE')
-    KFK_PORT                    = '9092'
-    # KFK_PORT                    = get_setting('KFK_SERVER_PORT')
-    # SERVER_ADDRESS              = "127.0.0.1:9092"
-    SERVER_ADDRESS =  KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT)
-    ADMIN_CLIENT                = AdminClient({'bootstrap.servers': SERVER_ADDRESS })
+
+    @staticmethod
+    def get_kafka_address() -> str:
+        kafka_server_address = get_setting('KFK_SERVER_ADDRESS', default=None)
+        if kafka_server_address is None:
+            KFK_NAMESPACE  = get_setting('KFK_NAMESPACE')
+            KFK_PORT       = get_setting('KFK_SERVER_PORT')
+            SERVER_ADDRESS = KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT)
+        return SERVER_ADDRESS
+        
+    @staticmethod
+    def get_admin_client():
+        SERVER_ADDRESS = KafkaConfig.get_kafka_address()
+        LOGGER.debug("KAFKA_SERVER_ADDRESS {:}".format(SERVER_ADDRESS))
+        # SERVER_ADDRESS = "127.0.0.1:9092"
+        ADMIN_CLIENT   = AdminClient({'bootstrap.servers': SERVER_ADDRESS })
+        return ADMIN_CLIENT
+
 
 class KafkaTopic(Enum):
     REQUEST  = 'topic_request' 

src/kpi_value_api/.gitlab-ci.yml

@@ -50,10 +50,30 @@ unit_test kpi-value-api:
     - docker login -u "$CI_REGISTRY_USER" -p "$CI_REGISTRY_PASSWORD" $CI_REGISTRY
     - if docker network list | grep teraflowbridge; then echo "teraflowbridge is already created"; else docker network create -d bridge teraflowbridge; fi
     - if docker container ls | grep $IMAGE_NAME; then docker rm -f $IMAGE_NAME; else echo "$IMAGE_NAME container is not in the system"; fi
+    - if docker container ls | grep kafka; then docker rm -f kafka; else echo "Kafka container is not in the system"; fi
+    - if docker container ls | grep zookeeper; then docker rm -f zookeeper; else echo "Zookeeper container is not in the system"; fi
     - docker container prune -f
   script:
     - docker pull "$CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG"
-    - docker run --name $IMAGE_NAME -d -p 30020:30020 -v "$PWD/src/$IMAGE_NAME/tests:/opt/results" --network=teraflowbridge $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG
+    - docker pull "bitnami/zookeeper:latest"
+    - docker pull "bitnami/kafka:latest"
+    - >
+      docker run --name zookeeper -d --network=teraflowbridge -p 2181:2181
+      bitnami/zookeeper:latest
+    - sleep 10 # Wait for Zookeeper to start
+    - docker run --name kafka -d --network=teraflowbridge -p 9092:9092
+      --env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+      --env ALLOW_PLAINTEXT_LISTENER=yes
+      bitnami/kafka:latest
+    - sleep 20 # Wait for Kafka to start
+    - KAFKA_IP=$(docker inspect kafka --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}")
+    - echo $KAFKA_IP
+    - > 
+      docker run --name $IMAGE_NAME -d -p 30020:30020
+      --env "KFK_SERVER_ADDRESS=${KAFKA_IP}:9092"
+      --volume "$PWD/src/$IMAGE_NAME/tests:/opt/results" 
+      --network=teraflowbridge 
+      $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG
     - sleep 5
     - docker ps -a
     - docker logs $IMAGE_NAME

src/kpi_value_api/service/KpiValueApiServiceServicerImpl.py

@@ -43,7 +43,7 @@ class KpiValueApiServiceServicerImpl(KpiValueAPIServiceServicer):
                        ) -> Empty:
         LOGGER.debug('StoreKpiValues: Received gRPC message object: {:}'.format(request))
         producer_obj = KafkaProducer({
-            'bootstrap.servers' : KafkaConfig.SERVER_ADDRESS.value    
+            'bootstrap.servers' : KafkaConfig.get_admin_client()
         })
         for kpi_value in request.kpi_value_list:
             kpi_value_to_produce : Tuple [str, Any, Any] = (