src/dags/multivim_vm_status.py

   1 #######################################################################################
   2 # Copyright ETSI Contributors and Others.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #######################################################################################
  17 from datetime import datetime, timedelta
  18
  19 from airflow import DAG
  20 from airflow.decorators import task
  21 from osm_mon.core.common_db import CommonDbClient
  22 from osm_mon.core.config import Config
  23 from osm_mon.vim_connectors.azure import AzureCollector
  24 from osm_mon.vim_connectors.gcp import GcpCollector
  25 from osm_mon.vim_connectors.openstack import OpenStackCollector
  26 from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
  27
  28
  29 SUPPORTED_VIM_TYPES = ["openstack", "vio", "gcp", "azure"]
  30 PROMETHEUS_PUSHGW = "pushgateway-prometheus-pushgateway:9091"
  31 PROMETHEUS_JOB_PREFIX = "airflow_osm_vm_status_"
  32 PROMETHEUS_METRIC = "vm_status"
  33 PROMETHEUS_METRIC_DESCRIPTION = "VM Status from VIM"
  34 SCHEDULE_INTERVAL = 1
  35
  36
  37 def get_all_vim():
  38     """Get VIMs from MongoDB"""
  39     print("Getting VIM list")
  40
  41     cfg = Config()
  42     print(cfg.conf)
  43     common_db = CommonDbClient(cfg)
  44     vim_accounts = common_db.get_vim_accounts()
  45     vim_list = []
  46     for vim in vim_accounts:
  47         print(f'Read VIM {vim["_id"]} ({vim["name"]})')
  48         vim_list.append(
  49             {"_id": vim["_id"], "name": vim["name"], "vim_type": vim["vim_type"]}
  50         )
  51
  52     print(vim_list)
  53     print("Getting VIM list OK")
  54     return vim_list
  55
  56
  57 def create_dag(dag_id, dag_number, dag_description, vim_id):
  58     dag = DAG(
  59         dag_id,
  60         catchup=False,
  61         default_args={
  62             "depends_on_past": False,
  63             "retries": 1,
  64             # "retry_delay": timedelta(minutes=1),
  65             "retry_delay": timedelta(seconds=10),
  66         },
  67         description=dag_description,
  68         is_paused_upon_creation=False,
  69         # schedule_interval=timedelta(minutes=SCHEDULE_INTERVAL),
  70         schedule_interval=f"*/{SCHEDULE_INTERVAL} * * * *",
  71         start_date=datetime(2022, 1, 1),
  72         tags=["osm", "vim"],
  73     )
  74
  75     with dag:
  76
  77         def get_vim_collector(vim_account):
  78             """Return a VIM collector for the vim_account"""
  79             vim_type = vim_account["vim_type"]
  80             if "config" in vim_account and "vim_type" in vim_account["config"]:
  81                 vim_type = vim_account["config"]["vim_type"].lower()
  82                 if vim_type == "vio" and "vrops_site" not in vim_account["config"]:
  83                     vim_type = "openstack"
  84             if vim_type == "openstack":
  85                 return OpenStackCollector(vim_account)
  86             if vim_type == "gcp":
  87                 return GcpCollector(vim_account)
  88             if vim_type == "azure":
  89                 return AzureCollector(vim_account)
  90             print(f"VIM type '{vim_type}' not supported")
  91             return None
  92
  93         def get_all_vm_status(vim_account):
  94             """Get VM status from the VIM"""
  95             collector = get_vim_collector(vim_account)
  96             if collector:
  97                 # status = collector.is_vim_ok()
  98                 # print(f"VIM status: {status}")
  99                 vm_status_list = collector.collect_servers_status()
 100                 return vm_status_list
 101             else:
 102                 return None
 103
 104         @task(task_id="get_all_vm_status_and_send_to_prometheus")
 105         def get_all_vm_status_and_send_to_prometheus(vim_id: str):
 106             """Authenticate against VIM, collect servers status and send to prometheus"""
 107
 108             # Get VIM account info from MongoDB
 109             print(f"Reading VIM info, id: {vim_id}")
 110             cfg = Config()
 111             common_db = CommonDbClient(cfg)
 112             vim_account = common_db.get_vim_account(vim_account_id=vim_id)
 113             print(vim_account)
 114
 115             # Define Prometheus Metric for NS topology
 116             registry = CollectorRegistry()
 117             metric = Gauge(
 118                 PROMETHEUS_METRIC,
 119                 PROMETHEUS_METRIC_DESCRIPTION,
 120                 labelnames=[
 121                     "vm_id",
 122                     "vim_id",
 123                 ],
 124                 registry=registry,
 125             )
 126
 127             # Get status of all VM from VIM
 128             all_vm_status = get_all_vm_status(vim_account)
 129             print(f"Got {len(all_vm_status)} VMs with their status:")
 130             if all_vm_status:
 131                 for vm in all_vm_status:
 132                     vm_id = vm["id"]
 133                     vm_status = vm["status"]
 134                     vm_name = vm.get("name", "")
 135                     print(f"    {vm_name} ({vm_id}) {vm_status}")
 136                     metric.labels(vm_id, vim_id).set(vm_status)
 137                 # Push to Prometheus only if there are VM
 138                 push_to_gateway(
 139                     gateway=PROMETHEUS_PUSHGW,
 140                     job=f"{PROMETHEUS_JOB_PREFIX}{vim_id}",
 141                     registry=registry,
 142                 )
 143             return
 144
 145         get_all_vm_status_and_send_to_prometheus(vim_id)
 146
 147     return dag
 148
 149
 150 vim_list = get_all_vim()
 151 for index, vim in enumerate(vim_list):
 152     vim_type = vim["vim_type"]
 153     if vim_type in SUPPORTED_VIM_TYPES:
 154         vim_id = vim["_id"]
 155         vim_name = vim["name"]
 156         dag_description = f"Dag for vim {vim_name}"
 157         dag_id = f"vm_status_vim_{vim_id}"
 158         print(f"Creating DAG {dag_id}")
 159         globals()[dag_id] = create_dag(
 160             dag_id=dag_id,
 161             dag_number=index,
 162             dag_description=dag_description,
 163             vim_id=vim_id,
 164         )
 165     else:
 166         print(f"VIM type '{vim_type}' not supported for collecting VM status")