1 #######################################################################################
2 # Copyright ETSI Contributors and Others.
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
8 # http://www.apache.org/licenses/LICENSE-2.0
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
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #######################################################################################
17 from datetime
import datetime
, timedelta
20 from airflow
import DAG
21 from airflow
.decorators
import task
22 from osm_mon
.core
.common_db
import CommonDbClient
23 from osm_mon
.core
.config
import Config
24 from osm_mon
.vim_connectors
.azure
import AzureCollector
25 from osm_mon
.vim_connectors
.gcp
import GcpCollector
26 from osm_mon
.vim_connectors
.openstack
import OpenStackCollector
27 from prometheus_client
import CollectorRegistry
, Gauge
, push_to_gateway
30 SUPPORTED_VIM_TYPES
= ["openstack", "vio", "gcp", "azure"]
31 PROMETHEUS_PUSHGW
= "pushgateway-prometheus-pushgateway:9091"
32 PROMETHEUS_JOB_PREFIX
= "airflow_osm_vim_status_"
33 PROMETHEUS_METRIC
= "vim_status"
34 PROMETHEUS_METRIC_DESCRIPTION
= "VIM status"
38 logger
= logging
.getLogger("airflow.task")
42 """Get VIMs from MongoDB"""
43 logger
.info("Getting VIM list")
47 common_db
= CommonDbClient(cfg
)
48 vim_accounts
= common_db
.get_vim_accounts()
50 for vim
in vim_accounts
:
51 logger
.info(f
'Read VIM {vim["_id"]} ({vim["name"]})')
53 {"_id": vim
["_id"], "name": vim
["name"], "vim_type": vim
["vim_type"]}
57 logger
.info("Getting VIM list OK")
61 def create_dag(dag_id
, dag_number
, dag_description
, vim_id
):
66 "depends_on_past": False,
68 # "retry_delay": timedelta(minutes=1),
69 "retry_delay": timedelta(seconds
=10),
71 description
=dag_description
,
72 is_paused_upon_creation
=False,
73 # schedule_interval=timedelta(minutes=SCHEDULE_INTERVAL),
74 schedule_interval
=f
"*/{SCHEDULE_INTERVAL} * * * *",
75 start_date
=datetime(2022, 1, 1),
81 def get_vim_collector(vim_account
):
82 """Return a VIM collector for the vim_account"""
83 vim_type
= vim_account
["vim_type"]
84 if "config" in vim_account
and "vim_type" in vim_account
["config"]:
85 vim_type
= vim_account
["config"]["vim_type"].lower()
86 if vim_type
== "vio" and "vrops_site" not in vim_account
["config"]:
87 vim_type
= "openstack"
88 if vim_type
== "openstack":
89 return OpenStackCollector(vim_account
)
91 return GcpCollector(vim_account
)
92 if vim_type
== "azure":
93 return AzureCollector(vim_account
)
94 logger
.info(f
"VIM type '{vim_type}' not supported")
97 @task(task_id
="get_vim_status_and_send_to_prometheus")
98 def get_vim_status_and_send_to_prometheus(vim_id
: str):
99 """Authenticate against VIM and check status"""
101 # Get VIM account info from MongoDB
102 logger
.info(f
"Reading VIM info, id: {vim_id}")
104 common_db
= CommonDbClient(cfg
)
105 vim_account
= common_db
.get_vim_account(vim_account_id
=vim_id
)
106 logger
.info(vim_account
)
108 # Define Prometheus Metric for NS topology
109 registry
= CollectorRegistry()
112 PROMETHEUS_METRIC_DESCRIPTION
,
118 metric
.labels(vim_id
).set(0)
121 collector
= get_vim_collector(vim_account
)
123 status
= collector
.is_vim_ok()
124 logger
.info(f
"VIM status: {status}")
125 metric
.labels(vim_id
).set(1)
127 logger
.info("Error creating VIM collector")
130 gateway
=PROMETHEUS_PUSHGW
,
131 job
=f
"{PROMETHEUS_JOB_PREFIX}{vim_id}",
136 get_vim_status_and_send_to_prometheus(vim_id
)
141 vim_list
= get_all_vim()
142 for index
, vim
in enumerate(vim_list
):
143 vim_type
= vim
["vim_type"]
144 if vim_type
in SUPPORTED_VIM_TYPES
:
146 vim_name
= vim
["name"]
147 dag_description
= f
"Dag for VIM {vim_name} status"
148 dag_id
= f
"vim_status_{vim_id}"
149 logger
.info(f
"Creating DAG {dag_id}")
150 globals()[dag_id
] = create_dag(
153 dag_description
=dag_description
,
157 logger
.info(f
"VIM type '{vim_type}' not supported for monitoring VIM status")