Add Python logging to DAGs

Change-Id: I2f8dd3b351ceb9a7da8e8b28d392e6fdef73f663
Signed-off-by: aguilard <e.dah.tid@telefonica.com>

diff --git a/src/osm_ngsa/dags/multivim_vim_status.py b/src/osm_ngsa/dags/multivim_vim_status.py
index 93894b1..f63ab4f 100644 (file)
--- a/src/osm_ngsa/dags/multivim_vim_status.py
+++ b/src/osm_ngsa/dags/multivim_vim_status.py
@@ -15,6 +15,7 @@
 # limitations under the License.
 #######################################################################################
 from datetime import datetime, timedelta
 # limitations under the License.
 #######################################################################################
 from datetime import datetime, timedelta

+import logging

 
 from airflow import DAG
 from airflow.decorators import task
 
 from airflow import DAG
 from airflow.decorators import task


@@ -33,24 +34,27 @@ PROMETHEUS_METRIC = "vim_status"
 PROMETHEUS_METRIC_DESCRIPTION = "VIM status"
 SCHEDULE_INTERVAL = 1
 
 PROMETHEUS_METRIC_DESCRIPTION = "VIM status"
 SCHEDULE_INTERVAL = 1
 

+# Logging
+logger = logging.getLogger("airflow.task")
+

 
 def get_all_vim():
     """Get VIMs from MongoDB"""
 
 def get_all_vim():
     """Get VIMs from MongoDB"""

-    print("Getting VIM list")
+    logger.info("Getting VIM list")

 
     cfg = Config()
 
     cfg = Config()

-    print(cfg.conf)
+    logger.info(cfg.conf)

     common_db = CommonDbClient(cfg)
     vim_accounts = common_db.get_vim_accounts()
     vim_list = []
     for vim in vim_accounts:
     common_db = CommonDbClient(cfg)
     vim_accounts = common_db.get_vim_accounts()
     vim_list = []
     for vim in vim_accounts:

-        print(f'Read VIM {vim["_id"]} ({vim["name"]})')
+        logger.info(f'Read VIM {vim["_id"]} ({vim["name"]})')

         vim_list.append(
             {"_id": vim["_id"], "name": vim["name"], "vim_type": vim["vim_type"]}
         )
 
         vim_list.append(
             {"_id": vim["_id"], "name": vim["name"], "vim_type": vim["vim_type"]}
         )
 

-    print(vim_list)
-    print("Getting VIM list OK")
+    logger.info(vim_list)
+    logger.info("Getting VIM list OK")

     return vim_list
 
 
     return vim_list
 
 


@@ -87,7 +91,7 @@ def create_dag(dag_id, dag_number, dag_description, vim_id):
                 return GcpCollector(vim_account)
             if vim_type == "azure":
                 return AzureCollector(vim_account)
                 return GcpCollector(vim_account)
             if vim_type == "azure":
                 return AzureCollector(vim_account)

-            print(f"VIM type '{vim_type}' not supported")
+            logger.info(f"VIM type '{vim_type}' not supported")

             return None
 
         @task(task_id="get_vim_status_and_send_to_prometheus")
             return None
 
         @task(task_id="get_vim_status_and_send_to_prometheus")


@@ -95,11 +99,11 @@ def create_dag(dag_id, dag_number, dag_description, vim_id):
             """Authenticate against VIM and check status"""
 
             # Get VIM account info from MongoDB
             """Authenticate against VIM and check status"""
 
             # Get VIM account info from MongoDB

-            print(f"Reading VIM info, id: {vim_id}")
+            logger.info(f"Reading VIM info, id: {vim_id}")

             cfg = Config()
             common_db = CommonDbClient(cfg)
             vim_account = common_db.get_vim_account(vim_account_id=vim_id)
             cfg = Config()
             common_db = CommonDbClient(cfg)
             vim_account = common_db.get_vim_account(vim_account_id=vim_id)

-            print(vim_account)
+            logger.info(vim_account)

 
             # Define Prometheus Metric for NS topology
             registry = CollectorRegistry()
 
             # Define Prometheus Metric for NS topology
             registry = CollectorRegistry()


@@ -117,10 +121,10 @@ def create_dag(dag_id, dag_number, dag_description, vim_id):
             collector = get_vim_collector(vim_account)
             if collector:
                 status = collector.is_vim_ok()
             collector = get_vim_collector(vim_account)
             if collector:
                 status = collector.is_vim_ok()

-                print(f"VIM status: {status}")
+                logger.info(f"VIM status: {status}")

                 metric.labels(vim_id).set(1)
             else:
                 metric.labels(vim_id).set(1)
             else:

-                print("Error creating VIM collector")
+                logger.info("Error creating VIM collector")

             # Push to Prometheus
             push_to_gateway(
                 gateway=PROMETHEUS_PUSHGW,
             # Push to Prometheus
             push_to_gateway(
                 gateway=PROMETHEUS_PUSHGW,


@@ -142,7 +146,7 @@ for index, vim in enumerate(vim_list):
         vim_name = vim["name"]
         dag_description = f"Dag for VIM {vim_name} status"
         dag_id = f"vim_status_{vim_id}"
         vim_name = vim["name"]
         dag_description = f"Dag for VIM {vim_name} status"
         dag_id = f"vim_status_{vim_id}"

-        print(f"Creating DAG {dag_id}")
+        logger.info(f"Creating DAG {dag_id}")

         globals()[dag_id] = create_dag(
             dag_id=dag_id,
             dag_number=index,
         globals()[dag_id] = create_dag(
             dag_id=dag_id,
             dag_number=index,


@@ -150,4 +154,4 @@ for index, vim in enumerate(vim_list):
             vim_id=vim_id,
         )
     else:
             vim_id=vim_id,
         )
     else:

-        print(f"VIM type '{vim_type}' not supported for monitoring VIM status")
+        logger.info(f"VIM type '{vim_type}' not supported for monitoring VIM status")

diff --git a/src/osm_ngsa/dags/multivim_vm_status.py b/src/osm_ngsa/dags/multivim_vm_status.py
index dbdbbc0..3cedabc 100644 (file)
--- a/src/osm_ngsa/dags/multivim_vm_status.py
+++ b/src/osm_ngsa/dags/multivim_vm_status.py
@@ -15,6 +15,7 @@
 # limitations under the License.
 #######################################################################################
 from datetime import datetime, timedelta
 # limitations under the License.
 #######################################################################################
 from datetime import datetime, timedelta

+import logging

 
 from airflow import DAG
 from airflow.decorators import task
 
 from airflow import DAG
 from airflow.decorators import task


@@ -33,24 +34,27 @@ PROMETHEUS_METRIC = "vm_status"
 PROMETHEUS_METRIC_DESCRIPTION = "VM Status from VIM"
 SCHEDULE_INTERVAL = 1
 
 PROMETHEUS_METRIC_DESCRIPTION = "VM Status from VIM"
 SCHEDULE_INTERVAL = 1
 

+# Logging
+logger = logging.getLogger("airflow.task")
+

 
 def get_all_vim():
     """Get VIMs from MongoDB"""
 
 def get_all_vim():
     """Get VIMs from MongoDB"""

-    print("Getting VIM list")
+    logger.info("Getting VIM list")

 
     cfg = Config()
 
     cfg = Config()

-    print(cfg.conf)
+    logger.info(cfg.conf)

     common_db = CommonDbClient(cfg)
     vim_accounts = common_db.get_vim_accounts()
     vim_list = []
     for vim in vim_accounts:
     common_db = CommonDbClient(cfg)
     vim_accounts = common_db.get_vim_accounts()
     vim_list = []
     for vim in vim_accounts:

-        print(f'Read VIM {vim["_id"]} ({vim["name"]})')
+        logger.info(f'Read VIM {vim["_id"]} ({vim["name"]})')

         vim_list.append(
             {"_id": vim["_id"], "name": vim["name"], "vim_type": vim["vim_type"]}
         )
 
         vim_list.append(
             {"_id": vim["_id"], "name": vim["name"], "vim_type": vim["vim_type"]}
         )
 

-    print(vim_list)
-    print("Getting VIM list OK")
+    logger.info(vim_list)
+    logger.info("Getting VIM list OK")

     return vim_list
 
 
     return vim_list
 
 


@@ -87,7 +91,7 @@ def create_dag(dag_id, dag_number, dag_description, vim_id):
                 return GcpCollector(vim_account)
             if vim_type == "azure":
                 return AzureCollector(vim_account)
                 return GcpCollector(vim_account)
             if vim_type == "azure":
                 return AzureCollector(vim_account)

-            print(f"VIM type '{vim_type}' not supported")
+            logger.info(f"VIM type '{vim_type}' not supported")

             return None
 
         def get_all_vm_status(vim_account):
             return None
 
         def get_all_vm_status(vim_account):


@@ -95,7 +99,7 @@ def create_dag(dag_id, dag_number, dag_description, vim_id):
             collector = get_vim_collector(vim_account)
             if collector:
                 status = collector.is_vim_ok()
             collector = get_vim_collector(vim_account)
             if collector:
                 status = collector.is_vim_ok()

-                print(f"VIM status: {status}")
+                logger.info(f"VIM status: {status}")

                 vm_status_list = collector.collect_servers_status()
                 return vm_status_list
             else:
                 vm_status_list = collector.collect_servers_status()
                 return vm_status_list
             else:


@@ -106,11 +110,11 @@ def create_dag(dag_id, dag_number, dag_description, vim_id):
             """Authenticate against VIM, collect servers status and send to prometheus"""
 
             # Get VIM account info from MongoDB
             """Authenticate against VIM, collect servers status and send to prometheus"""
 
             # Get VIM account info from MongoDB

-            print(f"Reading VIM info, id: {vim_id}")
+            logger.info(f"Reading VIM info, id: {vim_id}")

             cfg = Config()
             common_db = CommonDbClient(cfg)
             vim_account = common_db.get_vim_account(vim_account_id=vim_id)
             cfg = Config()
             common_db = CommonDbClient(cfg)
             vim_account = common_db.get_vim_account(vim_account_id=vim_id)

-            print(vim_account)
+            logger.info(vim_account)

 
             # Define Prometheus Metric for NS topology
             registry = CollectorRegistry()
 
             # Define Prometheus Metric for NS topology
             registry = CollectorRegistry()


@@ -126,13 +130,13 @@ def create_dag(dag_id, dag_number, dag_description, vim_id):
 
             # Get status of all VM from VIM
             all_vm_status = get_all_vm_status(vim_account)
 
             # Get status of all VM from VIM
             all_vm_status = get_all_vm_status(vim_account)

-            print(f"Got {len(all_vm_status)} VMs with their status:")
+            logger.info(f"Got {len(all_vm_status)} VMs with their status:")

             if all_vm_status:
                 for vm in all_vm_status:
                     vm_id = vm["id"]
                     vm_status = vm["status"]
                     vm_name = vm.get("name", "")
             if all_vm_status:
                 for vm in all_vm_status:
                     vm_id = vm["id"]
                     vm_status = vm["status"]
                     vm_name = vm.get("name", "")

-                    print(f"    {vm_name} ({vm_id}) {vm_status}")
+                    logger.info(f"    {vm_name} ({vm_id}) {vm_status}")

                     metric.labels(vm_id, vim_id).set(vm_status)
                 # Push to Prometheus only if there are VM
                 push_to_gateway(
                     metric.labels(vm_id, vim_id).set(vm_status)
                 # Push to Prometheus only if there are VM
                 push_to_gateway(


@@ -155,7 +159,7 @@ for index, vim in enumerate(vim_list):
         vim_name = vim["name"]
         dag_description = f"Dag for vim {vim_name}"
         dag_id = f"vm_status_vim_{vim_id}"
         vim_name = vim["name"]
         dag_description = f"Dag for vim {vim_name}"
         dag_id = f"vm_status_vim_{vim_id}"

-        print(f"Creating DAG {dag_id}")
+        logger.info(f"Creating DAG {dag_id}")

         globals()[dag_id] = create_dag(
             dag_id=dag_id,
             dag_number=index,
         globals()[dag_id] = create_dag(
             dag_id=dag_id,
             dag_number=index,


@@ -163,4 +167,4 @@ for index, vim in enumerate(vim_list):
             vim_id=vim_id,
         )
     else:
             vim_id=vim_id,
         )
     else:

-        print(f"VIM type '{vim_type}' not supported for collecting VM status")
+        logger.info(f"VIM type '{vim_type}' not supported for collecting VM status")

diff --git a/src/osm_ngsa/dags/ns_topology.py b/src/osm_ngsa/dags/ns_topology.py
index d3fb504..0fc20c3 100644 (file)
--- a/src/osm_ngsa/dags/ns_topology.py
+++ b/src/osm_ngsa/dags/ns_topology.py
@@ -15,6 +15,7 @@
 # limitations under the License.
 #######################################################################################
 from datetime import datetime, timedelta
 # limitations under the License.
 #######################################################################################
 from datetime import datetime, timedelta

+import logging

 
 from airflow.decorators import dag, task
 from osm_mon.core.common_db import CommonDbClient
 
 from airflow.decorators import dag, task
 from osm_mon.core.common_db import CommonDbClient


@@ -28,6 +29,9 @@ PROMETHEUS_METRIC = "ns_topology"
 PROMETHEUS_METRIC_DESCRIPTION = "Network services topology"
 SCHEDULE_INTERVAL = 2
 
 PROMETHEUS_METRIC_DESCRIPTION = "Network services topology"
 SCHEDULE_INTERVAL = 2
 

+# Logging
+logger = logging.getLogger("airflow.task")
+

 
 @dag(
     catchup=False,
 
 @dag(
     catchup=False,


@@ -70,9 +74,9 @@ def ns_topology():
         )
 
         # Getting VNFR list from MongoDB
         )
 
         # Getting VNFR list from MongoDB

-        print("Getting VNFR list from MongoDB")
+        logger.info("Getting VNFR list from MongoDB")

         cfg = Config()
         cfg = Config()

-        print(cfg.conf)
+        # logger.debug(cfg.conf)

         common_db = CommonDbClient(cfg)
         vnfr_list = common_db.get_vnfrs()
 
         common_db = CommonDbClient(cfg)
         vnfr_list = common_db.get_vnfrs()
 


@@ -93,45 +97,48 @@ def ns_topology():
             project_id = "None"
             if project_list:
                 project_id = project_list[0]
             project_id = "None"
             if project_list:
                 project_id = project_list[0]

-            # TODO: use logger with loglevels instead of print

             # Other info
             ns_state = vnfr["_admin"]["nsState"]
             vnf_membex_index = vnfr["member-vnf-index-ref"]
             # Other info
             ns_state = vnfr["_admin"]["nsState"]
             vnf_membex_index = vnfr["member-vnf-index-ref"]

-            print(
-                f"Read VNFR: id: {vnf_id}, ns_id: {ns_id}, ",
-                f"state: {ns_state}, vnfd_id: {vnfd_id}, ",
-                f"vnf_membex_index: {vnf_membex_index}, ",
-                f"project_id: {project_id}",
+            logger.info(
+                f"Read VNFR: id: {vnf_id}, ns_id: {ns_id}, "
+                f"state: {ns_state}, vnfd_id: {vnfd_id}, "
+                f"vnf_membex_index: {vnf_membex_index}, "
+                f"project_id: {project_id}"

             )
             # Only send topology if ns State is one of the nsAllowedStatesSet
             if ns_state not in nsAllowedStatesSet:
                 continue
 
             )
             # Only send topology if ns State is one of the nsAllowedStatesSet
             if ns_state not in nsAllowedStatesSet:
                 continue
 

-            print("VDU list:")
+            logger.debug("VDU list:")

             for vdu in vnfr.get("vdur", []):
                 # Label vdu_id
                 vdu_id = vdu["_id"]
                 # Label vim_id
                 vim_info = vdu.get("vim_info")
                 if not vim_info:
             for vdu in vnfr.get("vdur", []):
                 # Label vdu_id
                 vdu_id = vdu["_id"]
                 # Label vim_id
                 vim_info = vdu.get("vim_info")
                 if not vim_info:

-                    print("Error: vim_info not available in vdur")
+                    logger.info("Error: vim_info not available in vdur")

                     continue
                 if len(vim_info) != 1:
                     continue
                 if len(vim_info) != 1:

-                    print("Error: more than one vim_info in vdur")
+                    logger.info("Error: more than one vim_info in vdur")

                     continue
                 vim_id = next(iter(vim_info))[4:]
                     continue
                 vim_id = next(iter(vim_info))[4:]

+                # TODO: check if it makes sense to use vnfr.vim-account-id as vim_id instead of the vim_info key

                 # Label vm_id
                 # Label vm_id

-                vm_id = vdu["vim-id"]
+                vm_id = vdu.get("vim-id")
+                if not vm_id:
+                    logger.info("Error: vim-id not available in vdur")
+                    continue

                 # Other VDU info
                 vdu_name = vdu.get("name", "UNKNOWN")
                 # Other VDU info
                 vdu_name = vdu.get("name", "UNKNOWN")

-                print(
-                    f"    id: {vdu_id}, name: {vdu_name}, "
+                logger.debug(
+                    f"  VDU id: {vdu_id}, name: {vdu_name}, "

                     f"vim_id: {vim_id}, vm_id: {vm_id}"
                 )
                     f"vim_id: {vim_id}, vm_id: {vm_id}"
                 )

-                print(
-                    f"METRIC SAMPLE: ns_id: {ns_id}, ",
-                    f"project_id: {project_id}, vnf_id: {vnf_id}, ",
-                    f"vdu_id: {vdu_id}, vm_id: {vm_id}, vim_id: {vim_id}",
+                logger.info(
+                    f"METRIC SAMPLE: ns_id: {ns_id}, "
+                    f"project_id: {project_id}, vnf_id: {vnf_id}, "
+                    f"vdu_id: {vdu_id}, vm_id: {vm_id}, vim_id: {vim_id}"

                 )
                 metric.labels(
                     ns_id,
                 )
                 metric.labels(
                     ns_id,


@@ -144,7 +151,6 @@ def ns_topology():
                     vnf_membex_index,
                 ).set(1)
 
                     vnf_membex_index,
                 ).set(1)
 

-        # print("Push to gateway")

         push_to_gateway(
             gateway=PROMETHEUS_PUSHGW, job=PROMETHEUS_JOB, registry=registry
         )
         push_to_gateway(
             gateway=PROMETHEUS_PUSHGW, job=PROMETHEUS_JOB, registry=registry
         )