From d86b71f66e68f083a25279a2e7ed7f892332838d Mon Sep 17 00:00:00 2001
From: Taha Yassine Kraiem <tahayk2@gmail.com>
Date: Tue, 7 Mar 2023 18:36:48 +0100
Subject: [PATCH] feat(chalice): health-check test

---
 api/chalicelib/core/health.py    |  2 +-
 ee/api/chalicelib/core/health.py | 32 +++++++++++++++++++++++++++++---
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/api/chalicelib/core/health.py b/api/chalicelib/core/health.py
index 95b4abdb9..071cf7f9c 100644
--- a/api/chalicelib/core/health.py
+++ b/api/chalicelib/core/health.py
@@ -133,7 +133,7 @@ def __check_redis():
         r = redis.Redis(host=u.hostname, port=u.port, socket_timeout=2)
         r.ping()
     except Exception as e:
-        print("!! Issue getting assist-health response")
+        print("!! Issue getting redis-health response")
         print(str(e))
         fail_response["details"]["errors"].append(str(e))
         return fail_response
diff --git a/ee/api/chalicelib/core/health.py b/ee/api/chalicelib/core/health.py
index 4de9844f0..6139992a0 100644
--- a/ee/api/chalicelib/core/health.py
+++ b/ee/api/chalicelib/core/health.py
@@ -3,6 +3,7 @@ from urllib.parse import urlparse
 import redis
 import requests
 from decouple import config
+import kafka
 
 from chalicelib.utils import pg_client, ch_client
 
@@ -133,7 +134,7 @@ def __check_redis():
         r = redis.Redis(host=u.hostname, port=u.port, socket_timeout=2)
         r.ping()
     except Exception as e:
-        print("!! Issue getting assist-health response")
+        print("!! Issue getting redis-health response")
         print(str(e))
         fail_response["details"]["errors"].append(str(e))
         return fail_response
@@ -156,7 +157,7 @@ def get_health():
         },
         "ingestionPipeline": {
             "redis": __check_redis,
-            "kafka": __not_supported
+            "kafka": __check_kafka
         },
         "backendServices": {
             "alerts": __check_be_service("alerts"),
@@ -219,4 +220,29 @@ def __check_database_ch():
 
 
 def __check_kafka():
-    pass
+    fail_response = {
+        "health": False,
+        "details": {"errors": ["server health-check failed"]}
+    }
+    if config("KAFKA_SERVERS", default=None) is None:
+        fail_response["details"]["errors"].append("KAFKA_SERVERS not defined in env-vars")
+        return fail_response
+
+    try:
+        # consumer = kafka.KafkaConsumer(group_id='test', bootstrap_servers=[config("KAFKA_SERVERS")])
+        # topics = consumer.topics()
+        #
+        # if not topics:
+        #     raise RuntimeError()
+        client =kafka.KafkaClient(bootstrap_servers=[config("KAFKA_SERVERS")])
+    except Exception as e:
+        print("!! Issue getting kafka-health response")
+        print(str(e))
+        fail_response["details"]["errors"].append(str(e))
+        return fail_response
+
+    return {
+        "health": True,
+        "details": {"version": r.execute_command('INFO')['redis_version']}
+    }
+