1 # Copyright 2018 Whitestack, LLC
2 # *************************************************************
4 # This file is part of OSM Monitoring module
5 # All Rights Reserved to Whitestack, LLC
7 # Licensed under the Apache License, Version 2.0 (the "License"); you may
8 # not use this file except in compliance with the License. You may obtain
9 # a copy of the License at
11 # http://www.apache.org/licenses/LICENSE-2.0
13 # Unless required by applicable law or agreed to in writing, software
14 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16 # License for the specific language governing permissions and limitations
19 # For those usages not covered by the Apache License, Version 2.0 please
20 # contact: bdiaz@whitestack.com or glavado@whitestack.com
23 from typing
import List
26 from requests
.auth
import HTTPBasicAuth
28 from osm_mon
.collector
.infra_collectors
.base_sdnc
import BaseSdncInfraCollector
29 from osm_mon
.collector
.metric
import Metric
30 from osm_mon
.core
.common_db
import CommonDbClient
31 from osm_mon
.core
.config
import Config
33 log
= logging
.getLogger(__name__
)
36 class OnosInfraCollector(BaseSdncInfraCollector
):
37 def __init__(self
, config
: Config
, sdnc_id
: str):
38 super().__init
__(config
, sdnc_id
)
39 self
.common_db
= CommonDbClient(config
)
40 self
.sdnc
= self
.common_db
.get_sdnc(sdnc_id
)
42 def _obtain_url(self
, sdnc_dict
):
43 url
= sdnc_dict
.get("url")
47 if not sdnc_dict
.get("ip") or not sdnc_dict
.get("port"):
48 raise Exception("You must provide a URL to contact the SDN Controller")
50 return "http://{}:{}/onos/v1/devices".format(
51 sdnc_dict
["ip"], sdnc_dict
["port"]
54 def collect(self
) -> List
[Metric
]:
56 sdnc_status
= self
.is_sdnc_ok()
57 if self
.sdnc
["_admin"]["projects_read"]:
58 sdnc_project_id
= self
.sdnc
["_admin"]["projects_read"][0]
61 sdnc_tags
= {"sdnc_id": self
.sdnc
["_id"], "project_id": sdnc_project_id
}
62 sdnc_status_metric
= Metric(sdnc_tags
, "sdnc_status", sdnc_status
)
63 metrics
.append(sdnc_status_metric
)
67 def is_sdnc_ok(self
) -> bool:
69 url
= self
._obtain
_url
(self
.sdnc
)
70 user
= self
.sdnc
["user"]
71 password
= self
.common_db
.decrypt_sdnc_password(
72 self
.sdnc
["password"], self
.sdnc
["schema_version"], self
.sdnc
["_id"]
75 requests
.get(url
, auth
=HTTPBasicAuth(user
, password
))
78 log
.exception("SDNC status is not OK!")