WIP - Check readiness

Author: kaggis

automation/argo_config.py

@@ -38,6 +38,7 @@ def __init__(self, path: str):
         self.web_api_token = automation.get("web_api_token")
         self.default_ops_profile_file = automation.get("default_ops_profile_file")
         self.hdfs_path = run.get("hdfs_path")
+        self.hdfs_check_path = run.get("hdfs_check_path")
         self.flink_path = run.get("flink_path")
         self.batch_jar_path = run.get("batch_jar_path")
         self.ingest_jar_path = run.get("ingest_jar_path")

automation/argo_web_api.py

@@ -9,6 +9,11 @@
 
 REQUEST_TIMEOUT = 30
 
+class TopoItem(Enum):
+    ENDPOINTS = "endpoints"
+    GROUPS = "groups"
+    SERVICE_TYPES = "service-types"
+
 
 class ArgoWebApi:
 
@@ -90,6 +95,29 @@ def update_tenant_db_info(
             f"tenant: {tenant_name} ({tenant_id}) - web-api updating db conf updated"
         )
 
+    def get_topology(
+        self,
+        tenant_id: str,
+        tenant_name: str,
+        tenant_access_token: str,
+        topology_item: TopoItem
+    ):  
+        """Retrieve topology items for specific tenant"""
+        logger.debug(
+            f"tenant: {tenant_name} ({tenant_id}) - retrieving report information from web-api..."
+        )
+        url = f"https://{self.config.web_api_endpoint}/api/v2/topology/{topology_item.value}"
+        headers = {
+            "x-api-key": tenant_access_token,
+            "Accept": "application/json",
+        }
+
+        response = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT)
+        response.raise_for_status()
+
+        return response.json().get("data")
+
+
     def get_reports(
         self,
         tenant_id: str,
@@ -110,7 +138,16 @@ def get_reports(
         response.raise_for_status()
 
         results = response.json().get("data")
-        return {item["info"]["name"]: item["id"] for item in results}
+    
+    def get_report_ids(
+        self,
+        tenant_id: str,
+        tenant_name: str,
+        tenant_access_token: str
+    ):
+        """Retrieve report names and ids for specific tenant"""
+        reports = get_reports(tenant_id,tenant_name,tenant_access_token)
+        return {item["info"]["name"]: item["id"] for item in reports}
 
     def create_ops_profile(
         self,

automation/check_readiness.py

@@ -0,0 +1,86 @@
+logger = logging.getLogger(__name__)
+
+from argo_config import ArgoConfig
+from argo_web_api import ArgoWebApi, TopoItem
+from argo_ams_library import (
+    AmsServiceException,
+    AmsUser,
+    AmsUserProject,
+    ArgoMessagingService,
+)
+import requests
+
+
+
+def check_hdfs(config: ArgoConfig, tenant_id: str, tenant_name: str) -> bool: 
+    """Checks if data for today exist in hdfs tenant folders"""
+
+    logger.debug(
+            f"tenant: {tenant_name} ({tenant_id}) - retrieving report information from web-api..."
+    )
+    today = date.today().strftime('%Y-%m-%d')
+    url = f"https://{config.hdfs_check_path}/{tenant_name}/mdata/{today}?op=LISTSTATUS"
+    headers = {
+        "x-api-key": tenant_access_token,
+        "Accept": "application/json",
+    }
+
+    response = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT)
+    response.raise_for_status()
+
+    result = response.json().get("FileStatuses").get("FileStatus")
+    if len(result) > 0: 
+        return True
+    return False
+     
+
+
+def check_readiness(config: ArgoConfig, tenant_id: str, tenant_name: str):
+    """Checks tenants readiness by doing web-api requests to see if topology and 
+       reports are defined and also by checking if data are present both in ams and hdfs"""
+
+    web_api = ArgoWebApi(config)
+    
+    # get access token from config file
+    tenant_token = config.tenants.get(tenant_name,{}).get("web_api_token")
+
+    # check if topology exists
+    topology_ready = True
+    topology_msg = []
+    topo_endpoints = web_api.get_topology(tenant_id, tenant_name, tenant_token, TopoItem.ENDPOINTS)
+    topo_groups = web_api.get_topology(tenant_id, tenant_name, tenant_token, TopoItem.GROUPS)
+    topo_service_types = web_api.get_topology(tenant_id, tenant_name, tenant_token, TopoItem.SERVICE_TYPES)
+
+    if len(topo_endpoints) > 0:
+        topology_msg.append("Topology endpoints are set.")
+    else:
+        topology_msg.append("Topology endpoints are missing!")
+        topology_ready = False
+    
+    if len(topo_groups) > 0:
+        topology_msg.append("Topology groups are set.")
+    else:
+        topology_msg.append("Topology groups are missing!")
+        topology_ready = False
+
+    if len(topo_groups) > 0:
+        topology_msg.append("Topology service-types are set.")
+    else:
+        topology_msg.append("Topology service-types are missing!")
+        topology_ready = False
+
+    # check reports
+    reports_ready = True
+    reports_msg = "Tenant has at least one report"
+
+    reports = web_api.get_reports(tenant_id, tenant_name, tenant_token)
+    if len(reports) < 0:
+        reports_msg = "Tenant has no reports!"
+    
+    # check ams
+    ams = ArgoMessagingService(
+        endpoint=config.ams_endpoint, token=config.ams_admin_token, project=tenant_name
+    )
+
+    topic_metric_data = ams.get_topic("metric_data")
+    topic_metric_data ams.get

automation/run_batch

@@ -141,7 +141,7 @@ def main():
     )
     if not report_id:
         # try to update the tenant report configuration
-        reports = web_api.get_reports(
+        reports = web_api.get_report_ids(
             tenant["id"], args.tenant, tenant["web_api_token"]
         )
         config.set_tenant_reports(tenant["id"], args.tenant, reports)