From 98c3ac7c859e34fe60d061b9ca591aba429e4118 Mon Sep 17 00:00:00 2001 From: Koren Lev Date: Mon, 18 Dec 2017 19:16:16 +0200 Subject: release 1.2 + new tagging Change-Id: I1e876451ec4a330f458dd57adadb15e39969b225 Signed-off-by: Koren Lev --- app/discover/clique_finder.py | 138 ++++++++++++--------- app/discover/event_manager.py | 4 +- app/discover/fetchers/api/api_access.py | 61 +++------ .../fetchers/api/api_fetch_host_instances.py | 2 +- .../fetchers/api/api_fetch_project_hosts.py | 44 ++++++- app/discover/fetchers/api/api_fetch_regions.py | 2 +- app/discover/fetchers/db/db_access.py | 29 +++-- app/discover/fetchers/kube/__init__.py | 9 ++ app/discover/fetchers/kube/kube_access.py | 28 +++++ .../fetchers/kube/kube_fetch_namespaces.py | 32 +++++ app/discover/link_finders/find_implicit_links.py | 128 +++++++++++++++++++ app/discover/link_finders/find_links.py | 3 + .../link_finders/find_links_for_instance_vnics.py | 2 + app/discover/scan_manager.py | 125 ++++++++++--------- app/discover/scan_metadata_parser.py | 28 +++-- app/discover/scanner.py | 21 +++- 16 files changed, 451 insertions(+), 205 deletions(-) create mode 100644 app/discover/fetchers/kube/__init__.py create mode 100644 app/discover/fetchers/kube/kube_access.py create mode 100644 app/discover/fetchers/kube/kube_fetch_namespaces.py create mode 100644 app/discover/link_finders/find_implicit_links.py (limited to 'app/discover') diff --git a/app/discover/clique_finder.py b/app/discover/clique_finder.py index 57b2e3b..4d68eb4 100644 --- a/app/discover/clique_finder.py +++ b/app/discover/clique_finder.py @@ -48,61 +48,53 @@ class CliqueFinder(Fetcher): self.find_cliques_for_type(clique_type) self.log.info("finished scanning for cliques") - # Calculate priority score + # Calculate priority score for clique type per environment and configuration def _get_priority_score(self, clique_type): + # environment-specific clique type takes precedence if self.env == clique_type['environment']: + return 16 + if (self.env_config['distribution'] == clique_type.get('distribution') + and + self.env_config['distribution_version'] == + clique_type.get('distribution_version')): + return 8 + if clique_type.get('mechanism_drivers') \ + in self.env_config['mechanism_drivers']: return 4 - if (self.env_config['distribution'] == clique_type.get('distribution') and - self.env_config['distribution_version'] == clique_type.get('distribution_version')): - return 3 - if clique_type.get('mechanism_drivers') in self.env_config['mechanism_drivers']: - return 2 if self.env_config['type_drivers'] == clique_type.get('type_drivers'): + return 2 + if clique_type.get('environment', '') == 'ANY': + # environment=ANY serves as fallback option, but it's not mandatory return 1 else: return 0 # Get clique type with max priority - # for given environment configuration and focal point type - def _get_clique_type(self, focal_point, clique_types): - # If there's no configuration match for the specified environment, - # we use the default clique type definition with environment='ANY' - fallback_type = next( - filter(lambda t: t['environment'] == 'ANY', clique_types), - None - ) - if not fallback_type: - raise ValueError("No fallback clique type (ANY) " - "defined for focal point type '{}'" - .format(focal_point)) - - clique_types.remove(fallback_type) - - priority_scores = [self._get_priority_score(clique_type) - for clique_type - in clique_types] - max_score = max(priority_scores) if priority_scores else 0 - - return (fallback_type - if max_score == 0 - else clique_types[priority_scores.index(max_score)]) + # for given focal point type + def _get_clique_type(self, clique_types): + scored_clique_types = [{'score': self._get_priority_score(clique_type), + 'clique_type': clique_type} + for clique_type in clique_types] + max_score = max(scored_clique_types, key=lambda t: t['score']) + if max_score['score'] == 0: + self.log.warn('No matching clique types for focal point type: {}' + .format(clique_types[0].get('focal_point_type'))) + return None + return max_score.get('clique_type') def get_clique_types(self): if not self.clique_types_by_type: - clique_types_by_focal_point = self.clique_types.aggregate([{ - "$group": { - "_id": "$focal_point_type", - "types": {"$push": "$$ROOT"} - } - }]) - - self.clique_types_by_type = { - cliques['_id']: self._get_clique_type(cliques['_id'], - cliques['types']) - for cliques in - clique_types_by_focal_point - } - + clique_types_candidates = {} + for clique in self.clique_types.find({}): + fp_type = clique.get('focal_point_type', '') + if not clique_types_candidates.get(fp_type): + clique_types_candidates[fp_type] = [] + clique_types_candidates[fp_type].append(clique) + for t in clique_types_candidates.keys(): + selected = self._get_clique_type(clique_types_candidates[t]) + if not selected: + continue + self.clique_types_by_type[t] = selected return self.clique_types_by_type def find_cliques_for_type(self, clique_type): @@ -125,11 +117,14 @@ class CliqueFinder(Fetcher): .find_one({"focal_point_type": o['type']}) constraints = [] if not constraint else constraint["constraints"] clique_types = self.get_clique_types() - clique_type = clique_types[o['type']] - new_clique = self.construct_clique_for_focal_point(o, clique_type, - constraints) - if not new_clique: + clique_type = clique_types.get(o['type']) + if not clique_type: self.cliques.delete({'_id': clique['_id']}) + else: + new_clique = self.construct_clique_for_focal_point(o, clique_type, + constraints) + if not new_clique: + self.cliques.delete({'_id': clique['_id']}) def construct_clique_for_focal_point(self, o, clique_type, constraints): # keep a hash of nodes in clique that were visited for each type @@ -146,12 +141,15 @@ class CliqueFinder(Fetcher): for c in constraints: val = o[c] if c in o else None clique["constraints"][c] = val + allow_implicit = clique_type.get('use_implicit_links', False) for link_type in clique_type["link_types"]: - self.check_link_type(clique, link_type, nodes_of_type) + self.check_link_type(clique, link_type, nodes_of_type, + allow_implicit=allow_implicit) # after adding the links to the clique, create/update the clique if not clique["links"]: return None + clique["clique_type"] = clique_type["_id"] focal_point_obj = self.inventory.find({"_id": clique["focal_point"]}) if not focal_point_obj: return None @@ -198,25 +196,32 @@ class CliqueFinder(Fetcher): '-'.join(link_type_parts) return CliqueFinder.link_type_reversed.get(link_type) - def check_link_type(self, clique, link_type, nodes_of_type): + def check_link_type(self, clique, link_type, nodes_of_type, + allow_implicit=False): # check if it's backwards link_type_reversed = self.get_link_type_reversed(link_type) # handle case of links like T<-->T self_linked = link_type == link_type_reversed use_reversed = False if not self_linked: - matches = self.links.find_one({ + link_search_condition = { "environment": self.env, "link_type": link_type_reversed - }) + } + if not allow_implicit: + link_search_condition['implicit'] = False + matches = self.links.find_one(link_search_condition) use_reversed = True if matches else False if self_linked or not use_reversed: - self.check_link_type_forward(clique, link_type, nodes_of_type) + self.check_link_type_forward(clique, link_type, nodes_of_type, + allow_implicit=allow_implicit) if self_linked or use_reversed: - self.check_link_type_back(clique, link_type, nodes_of_type) + self.check_link_type_back(clique, link_type, nodes_of_type, + allow_implicit=allow_implicit) def check_link_type_for_direction(self, clique, link_type, nodes_of_type, - is_reversed=False): + is_reversed=False, + allow_implicit=False): if is_reversed: link_type = self.get_link_type_reversed(link_type) from_type = link_type[:link_type.index("-")] @@ -233,7 +238,8 @@ class CliqueFinder(Fetcher): clique, link_type, side_to_match, - other_side) + other_side, + allow_implicit=allow_implicit) nodes_to_add = nodes_to_add | matches if other_side_type not in nodes_of_type: nodes_of_type[other_side_type] = set() @@ -241,13 +247,17 @@ class CliqueFinder(Fetcher): nodes_of_type[other_side_type] | nodes_to_add def find_matches_for_point(self, match_point, clique, link_type, - side_to_match, other_side) -> set: + side_to_match, other_side, + allow_implicit=False) -> set: nodes_to_add = set() - matches = self.links.find({ + link_search_condition = { "environment": self.env, "link_type": link_type, side_to_match: ObjectId(match_point) - }) + } + if not allow_implicit: + link_search_condition['implicit'] = False + matches = self.links.find(link_search_condition) for link in matches: link_id = link["_id"] if link_id in clique["links"]: @@ -260,10 +270,14 @@ class CliqueFinder(Fetcher): nodes_to_add.add(other_side_point) return nodes_to_add - def check_link_type_forward(self, clique, link_type, nodes_of_type): + def check_link_type_forward(self, clique, link_type, nodes_of_type, + allow_implicit=False): self.check_link_type_for_direction(clique, link_type, nodes_of_type, - is_reversed=False) + is_reversed=False, + allow_implicit=allow_implicit) - def check_link_type_back(self, clique, link_type, nodes_of_type): + def check_link_type_back(self, clique, link_type, nodes_of_type, + allow_implicit=False): self.check_link_type_for_direction(clique, link_type, nodes_of_type, - is_reversed=True) + is_reversed=True, + allow_implicit=allow_implicit) diff --git a/app/discover/event_manager.py b/app/discover/event_manager.py index 4855acc..c01916c 100644 --- a/app/discover/event_manager.py +++ b/app/discover/event_manager.py @@ -113,8 +113,8 @@ class EventManager(Manager): def get_listener(self, env: str): env_config = self.inv.get_env_config(env) return (self.LISTENERS.get(env_config.get('distribution'), {}) - .get(env_config.get('distribution_version', - DefaultListener))) + .get(env_config.get('distribution_version'), + DefaultListener)) def listen_to_events(self, listener: ListenerBase, env_name: str, process_vars: dict): listener.listen({ diff --git a/app/discover/fetchers/api/api_access.py b/app/discover/fetchers/api/api_access.py index f685faf..1fca202 100644 --- a/app/discover/fetchers/api/api_access.py +++ b/app/discover/fetchers/api/api_access.py @@ -12,21 +12,18 @@ import re import requests import time -from discover.configuration import Configuration -from discover.fetcher import Fetcher +from utils.api_access_base import ApiAccessBase from utils.string_utils import jsonify -class ApiAccess(Fetcher): +class ApiAccess(ApiAccessBase): + + ADMIN_PORT = "35357" + subject_token = None initialized = False regions = {} - config = None - api_config = None - host = "" - base_url = "" - admin_token = "" tokens = {} admin_endpoint = "" admin_project = None @@ -38,28 +35,19 @@ class ApiAccess(Fetcher): # identity API v2 version with admin token def __init__(self, config=None): - super(ApiAccess, self).__init__() - if ApiAccess.initialized: + super().__init__('OpenStack', config) + self.base_url = "http://" + self.host + ":" + self.port + if self.initialized: return - ApiAccess.config = {'OpenStack': config} if config else Configuration() - ApiAccess.api_config = ApiAccess.config.get("OpenStack") - host = ApiAccess.api_config.get("host", "") - ApiAccess.host = host - port = ApiAccess.api_config.get("port", "") - if not (host and port): - raise ValueError('Missing definition of host or port ' + - 'for OpenStack API access') - ApiAccess.base_url = "http://" + host + ":" + port - ApiAccess.admin_token = ApiAccess.api_config.get("admin_token", "") - ApiAccess.admin_project = ApiAccess.api_config.get("admin_project", - "admin") - ApiAccess.admin_endpoint = "http://" + host + ":" + "35357" + ApiAccess.admin_project = self.api_config.get("admin_project", "admin") + ApiAccess.admin_endpoint = "http://" + self.host + ":" + self.ADMIN_PORT token = self.v2_auth_pwd(ApiAccess.admin_project) if not token: raise ValueError("Authentication failed. Failed to obtain token") else: self.subject_token = token + self.initialized = True @staticmethod def parse_time(time_str): @@ -95,9 +83,9 @@ class ApiAccess(Fetcher): subject_token = self.get_existing_token(project_id) if subject_token: return subject_token - req_url = ApiAccess.base_url + "/v2.0/tokens" + req_url = self.base_url + "/v2.0/tokens" response = requests.post(req_url, json=post_body, headers=headers, - timeout=5) + timeout=self.CONNECT_TIMEOUT) response = response.json() ApiAccess.auth_response[project_id] = response if 'error' in response: @@ -120,8 +108,8 @@ class ApiAccess(Fetcher): return token_details def v2_auth_pwd(self, project): - user = ApiAccess.api_config["user"] - pwd = ApiAccess.api_config["pwd"] + user = self.api_config["user"] + pwd = self.api_config["pwd"] post_body = { "auth": { "passwordCredentials": { @@ -148,23 +136,6 @@ class ApiAccess(Fetcher): auth_response = ApiAccess.auth_response.get('admin', {}) return auth_response - def get_rel_url(self, relative_url, headers): - req_url = ApiAccess.base_url + relative_url - return self.get_url(req_url, headers) - - def get_url(self, req_url, headers): - response = requests.get(req_url, headers=headers) - if response.status_code != requests.codes.ok: - # some error happened - if "reason" in response: - msg = ", reason: {}".format(response.reason) - else: - msg = ", response: {}".format(response.text) - self.log.error("req_url: {} {}".format(req_url, msg)) - return None - ret = response.json() - return ret - def get_region_url(self, region_name, service): if region_name not in self.regions: return None @@ -174,7 +145,7 @@ class ApiAccess(Fetcher): return None orig_url = s["adminURL"] # replace host name with the host found in config - url = re.sub(r"^([^/]+)//[^:]+", r"\1//" + ApiAccess.host, orig_url) + url = re.sub(r"^([^/]+)//[^:]+", r"\1//" + self.host, orig_url) return url # like get_region_url(), but remove everything starting from the "/v2" diff --git a/app/discover/fetchers/api/api_fetch_host_instances.py b/app/discover/fetchers/api/api_fetch_host_instances.py index 56cffda..bf8513a 100644 --- a/app/discover/fetchers/api/api_fetch_host_instances.py +++ b/app/discover/fetchers/api/api_fetch_host_instances.py @@ -18,7 +18,7 @@ class ApiFetchHostInstances(ApiAccess, DbAccess, metaclass=Singleton): def __init__(self): super(ApiFetchHostInstances, self).__init__() self.inv = InventoryMgr() - self.endpoint = ApiAccess.base_url.replace(":5000", ":8774") + self.endpoint = self.base_url.replace(":5000", ":8774") self.projects = None self.db_fetcher = DbFetchInstances() diff --git a/app/discover/fetchers/api/api_fetch_project_hosts.py b/app/discover/fetchers/api/api_fetch_project_hosts.py index 5b911f5..2aeb24f 100644 --- a/app/discover/fetchers/api/api_fetch_project_hosts.py +++ b/app/discover/fetchers/api/api_fetch_project_hosts.py @@ -11,9 +11,11 @@ import json from discover.fetchers.api.api_access import ApiAccess from discover.fetchers.db.db_access import DbAccess +from discover.fetchers.cli.cli_access import CliAccess +from utils.ssh_connection import SshError -class ApiFetchProjectHosts(ApiAccess, DbAccess): +class ApiFetchProjectHosts(ApiAccess, DbAccess, CliAccess): def __init__(self): super(ApiFetchProjectHosts, self).__init__() @@ -107,6 +109,7 @@ class ApiFetchProjectHosts(ApiAccess, DbAccess): s = services["nova-compute"] if s["available"] and s["active"]: self.add_host_type(doc, "Compute", az['zoneName']) + self.fetch_host_os_details(doc) return doc # fetch more details of network nodes from neutron DB agents table @@ -121,7 +124,12 @@ class ApiFetchProjectHosts(ApiAccess, DbAccess): """.format(self.neutron_db) results = self.get_objects_list(query, "") for r in results: - host = hosts[r["host"]] + host = r["host"] + if host not in hosts: + self.log.error("host from agents table not in hosts list: {}" + .format(host)) + continue + host = hosts[host] host["config"] = json.loads(r["configurations"]) self.add_host_type(host, "Network", '') @@ -136,9 +144,33 @@ class ApiFetchProjectHosts(ApiAccess, DbAccess): for db_row in results: doc.update(db_row) - def add_host_type(self, doc, type, zone): - if not type in doc["host_type"]: - doc["host_type"].append(type) - if type == 'Compute': + @staticmethod + def add_host_type(doc, host_type, zone): + if host_type not in doc["host_type"]: + doc["host_type"].append(host_type) + if host_type == 'Compute': doc['zone'] = zone doc['parent_id'] = zone + + def fetch_host_os_details(self, doc): + cmd = 'cat /etc/os-release && echo "ARCHITECURE=`arch`"' + try: + lines = self.run_fetch_lines(cmd, ssh_to_host=doc['host']) + except SshError as e: + self.log.error('{}: {}', cmd, str(e)) + os_attributes = {} + attributes_to_fetch = { + 'NAME': 'name', + 'VERSION': 'version', + 'ID': 'ID', + 'ID_LIKE': 'ID_LIKE', + 'ARCHITECURE': 'architecure' + } + for attr in attributes_to_fetch: + matches = [l for l in lines if l.startswith(attr + '=')] + if matches: + line = matches[0] + attr_name = attributes_to_fetch[attr] + os_attributes[attr_name] = line[line.index('=')+1:].strip('"') + if os_attributes: + doc['OS'] = os_attributes diff --git a/app/discover/fetchers/api/api_fetch_regions.py b/app/discover/fetchers/api/api_fetch_regions.py index 23a3736..4e83b01 100644 --- a/app/discover/fetchers/api/api_fetch_regions.py +++ b/app/discover/fetchers/api/api_fetch_regions.py @@ -13,7 +13,7 @@ from discover.fetchers.api.api_access import ApiAccess class ApiFetchRegions(ApiAccess): def __init__(self): super(ApiFetchRegions, self).__init__() - self.endpoint = ApiAccess.base_url + self.endpoint = self.base_url def get(self, regions_folder_id): token = self.v2_auth_pwd(self.admin_project) diff --git a/app/discover/fetchers/db/db_access.py b/app/discover/fetchers/db/db_access.py index 090ab84..5ff49d5 100644 --- a/app/discover/fetchers/db/db_access.py +++ b/app/discover/fetchers/db/db_access.py @@ -38,8 +38,7 @@ class DbAccess(Fetcher): conn = None query_count_per_con = 0 - # connection timeout set to 30 seconds, - # due to problems over long connections + # connection timeout set to 5 seconds TIMEOUT = 5 def __init__(self, mysql_config=None): @@ -47,6 +46,9 @@ class DbAccess(Fetcher): self.config = {'mysql': mysql_config} if mysql_config \ else Configuration() self.conf = self.config.get("mysql") + self.connect_timeout = int(self.conf['connect_timeout']) \ + if 'connect_timeout' in self.conf \ + else self.TIMEOUT self.connect_to_db() self.neutron_db = self.get_neutron_db_name() @@ -55,16 +57,18 @@ class DbAccess(Fetcher): return try: connector = mysql.connector - DbAccess.conn = connector.connect(host=_host, port=_port, - connection_timeout=self.TIMEOUT, - user=_user, - password=_pwd, - database=_database, - raise_on_warnings=True) + conn = connector.connect(host=_host, port=_port, + connection_timeout=self.connect_timeout, + user=_user, + password=_pwd, + database=_database, + raise_on_warnings=True) + DbAccess.conn = conn DbAccess.conn.ping(True) # auto-reconnect if necessary except Exception as e: - self.log.critical("failed to connect to MySQL DB: {}" - .format(str(e))) + msg = "failed to connect to MySQL DB: {}".format(str(e)) + self.log.critical(msg) + raise ScanError(msg) return DbAccess.query_count_per_con = 0 @@ -93,8 +97,11 @@ class DbAccess(Fetcher): DbAccess.conn = None self.conf = self.config.get("mysql") cnf = self.conf + pwd = cnf.get('pwd', '') + if not pwd: + raise ScanError('db_access: attribute pwd is missing') self.db_connect(cnf.get('host', ''), cnf.get('port', ''), - cnf.get('user', ''), cnf.get('pwd', ''), + cnf.get('user', ''), pwd, cnf.get('schema', 'nova')) @with_cursor diff --git a/app/discover/fetchers/kube/__init__.py b/app/discover/fetchers/kube/__init__.py new file mode 100644 index 0000000..b0637e9 --- /dev/null +++ b/app/discover/fetchers/kube/__init__.py @@ -0,0 +1,9 @@ +############################################################################### +# Copyright (c) 2017 Koren Lev (Cisco Systems), Yaron Yogev (Cisco Systems) # +# and others # +# # +# All rights reserved. This program and the accompanying materials # +# are made available under the terms of the Apache License, Version 2.0 # +# which accompanies this distribution, and is available at # +# http://www.apache.org/licenses/LICENSE-2.0 # +############################################################################### diff --git a/app/discover/fetchers/kube/kube_access.py b/app/discover/fetchers/kube/kube_access.py new file mode 100644 index 0000000..38bb978 --- /dev/null +++ b/app/discover/fetchers/kube/kube_access.py @@ -0,0 +1,28 @@ +############################################################################### +# Copyright (c) 2017 Koren Lev (Cisco Systems), Yaron Yogev (Cisco Systems) # +# and others # +# # +# All rights reserved. This program and the accompanying materials # +# are made available under the terms of the Apache License, Version 2.0 # +# which accompanies this distribution, and is available at # +# http://www.apache.org/licenses/LICENSE-2.0 # +############################################################################### +from kubernetes.client import Configuration as KubConf, CoreV1Api + +from utils.api_access_base import ApiAccessBase + + +class KubeAccess(ApiAccessBase): + + def __init__(self, config=None): + super().__init__('Kubernetes', config) + self.base_url = 'https://{}:{}'.format(self.host, self.port) + self.bearer_token = self.api_config.get('token', '') + conf = KubConf() + conf.host = self.base_url + conf.user = self.api_config.get('user') + conf.api_key_prefix['authorization'] = 'Bearer' + conf.api_key['authorization'] = self.bearer_token + conf.verify_ssl = False + self.api = CoreV1Api() + diff --git a/app/discover/fetchers/kube/kube_fetch_namespaces.py b/app/discover/fetchers/kube/kube_fetch_namespaces.py new file mode 100644 index 0000000..951ddb8 --- /dev/null +++ b/app/discover/fetchers/kube/kube_fetch_namespaces.py @@ -0,0 +1,32 @@ +############################################################################### +# Copyright (c) 2017 Koren Lev (Cisco Systems), Yaron Yogev (Cisco Systems) # +# and others # +# # +# All rights reserved. This program and the accompanying materials # +# are made available under the terms of the Apache License, Version 2.0 # +# which accompanies this distribution, and is available at # +# http://www.apache.org/licenses/LICENSE-2.0 # +############################################################################### +from discover.fetchers.kube.kube_access import KubeAccess + + +class KubeFetchNamespaces(KubeAccess): + + def __init__(self, config=None): + super().__init__(config) + + def get(self, object_id): + namespaces = self.api.list_namespace() + return [self.get_namespace(i) for i in namespaces.items] + + @staticmethod + def get_namespace(namespace): + attrs = ['creation_timestamp', 'self_link', 'uid'] + namespace_details = { + 'name': namespace.metadata.name, + 'status': namespace.status.phase + } + namespace_details.update({x: getattr(namespace.metadata, x, '') + for x in attrs}) + namespace_details['id'] = namespace_details['uid'] + return namespace_details diff --git a/app/discover/link_finders/find_implicit_links.py b/app/discover/link_finders/find_implicit_links.py new file mode 100644 index 0000000..01eaa7b --- /dev/null +++ b/app/discover/link_finders/find_implicit_links.py @@ -0,0 +1,128 @@ +############################################################################### +# Copyright (c) 2017 Koren Lev (Cisco Systems), Yaron Yogev (Cisco Systems) # +# and others # +# # +# All rights reserved. This program and the accompanying materials # +# are made available under the terms of the Apache License, Version 2.0 # +# which accompanies this distribution, and is available at # +# http://www.apache.org/licenses/LICENSE-2.0 # +############################################################################### +from discover.link_finders.find_links import FindLinks + + +class FindImplicitLinks(FindLinks): + + def __init__(self): + super().__init__() + self.links = [] + self.constraint_attributes = self.get_constraint_attributes() + + def add_links(self): + self.log.info('adding implicit links') + self.get_existing_links() + self.get_transitive_closure() + + def get_constraint_attributes(self) -> list: + attributes = set() + for c in self.inv.find({'environment': self.get_env()}, + collection='clique_constraints'): + for a in c['constraints']: + attributes.add(a) + return list(attributes) + + def get_existing_links(self): + self.log.info('fetching existing links') + existing_links = self.inv.find({'environment': self.get_env()}, + collection='links') + for l in existing_links: + self.links.append({'pass': 0, 'link': l}) + + def constraints_match(self, link1, link2): + if 'attributes' not in link1 or 'attributes' not in link2: + return True + attr1 = link1['attributes'] + attr2 = link2['attributes'] + for a in self.constraint_attributes: + if a in attr1 and a in attr2 and attr1[a] != attr2[a]: + return False + return True + + def links_match(self, start, dest): + if start['link_type'] == dest['link_type']: + return False # obviously we cannot make an implicit link of this + if start['source_id'] == dest['target_id']: + return False # avoid cyclic links + if not self.constraints_match(start, dest): + return False + return start['target_id'] == dest['source_id'] + + def add_matching_links(self, link, pass_no): + self.log.debug('looking for matches for link: {};{}' + .format(link['source_id'], link['target_id'])) + matches = [l for l in self.links + if l['pass'] == 0 # take only original links + and self.links_match(link, l['link'])] + for l in matches: + implicit = self.add_implicit_link(link, l['link']) + self.links.append({'pass': pass_no, 'link': implicit}) + return len(matches) + + def get_link_constraint_attributes(self, link1, link2) -> dict: + attributes = {} + for a in self.constraint_attributes: + # constraints_match() verified the attribute values don't conflict + if a in link1.get('attributes', {}): + attributes[a] = link1['attributes'][a] + elif a in link2.get('attributes', {}): + attributes[a] = link2['attributes'][a] + return attributes + + @staticmethod + def get_attr(attr, link1, link2): + if attr not in link1 and attr not in link2: + return None + if attr not in link1: + return link2[attr] + if attr not in link2 or link1[attr] == link2[attr]: + return link1[attr] + return None + + def add_implicit_link(self, link1, link2): + link_type_from = link1['link_type'].split('-')[0] + link_type_to = link2['link_type'].split('-')[1] + link_type = '{}-{}'.format(link_type_from, link_type_to) + link_name = '' + state = 'down' \ + if link1['state'] == 'down' or link2['state'] == 'down' \ + else 'up' + link_weight = 0 # TBD + host = self.get_attr('host', link1, link2) + switch = self.get_attr('switch', link1, link2) + extra_attributes = self.get_link_constraint_attributes(link1, link2) + self.log.debug('adding implicit link: link type: {}, from: {}, to: {}' + .format(link_type, + link1['source_id'], + link2['target_id'])) + implicit = self.create_link(self.get_env(), + link1['source'], link1['source_id'], + link2['target'], link2['target_id'], + link_type, link_name, state, link_weight, + host=host, switch=switch, + implicit=True, + extra_attributes=extra_attributes) + return implicit + + def get_transitive_closure(self): + pass_no = 1 + while True: + match_count = 0 + last_pass_links = [l for l in self.links if l['pass'] == pass_no-1] + for l in last_pass_links: + match_count += self.add_matching_links(l['link'], pass_no) + self.log.info('Transitive closure pass #{}: ' + 'found {} implicit links' + .format(pass_no, match_count)) + if match_count == 0: + break + pass_no += 1 + self.log.info('done adding implicit links') diff --git a/app/discover/link_finders/find_links.py b/app/discover/link_finders/find_links.py index d234479..31d39e5 100644 --- a/app/discover/link_finders/find_links.py +++ b/app/discover/link_finders/find_links.py @@ -19,6 +19,7 @@ class FindLinks(Fetcher): def create_link(self, env, source, source_id, target, target_id, link_type, link_name, state, link_weight, host=None, switch=None, + implicit=False, extra_attributes=None): if extra_attributes is None: extra_attributes = {} @@ -27,9 +28,11 @@ class FindLinks(Fetcher): link = self.inv.create_link(env, source, source_id, target, target_id, link_type, link_name, state, link_weight, + implicit=implicit, source_label=source_label, target_label=target_label, host=host, switch=switch, extra_attributes=extra_attributes) if self.inv.monitoring_setup_manager: self.inv.monitoring_setup_manager.create_setup(link) + return link diff --git a/app/discover/link_finders/find_links_for_instance_vnics.py b/app/discover/link_finders/find_links_for_instance_vnics.py index 975ab1a..1dfb818 100644 --- a/app/discover/link_finders/find_links_for_instance_vnics.py +++ b/app/discover/link_finders/find_links_for_instance_vnics.py @@ -49,6 +49,8 @@ class FindLinksForInstanceVnics(FindLinks): network_id = net['network']['id'] v['network'] = network_id self.inv.set(v) + if self.inv.monitoring_setup_manager: + self.inv.monitoring_setup_manager.create_setup(instance) break state = "up" # TBD link_weight = 0 # TBD diff --git a/app/discover/scan_manager.py b/app/discover/scan_manager.py index 6c46d47..91dd06c 100644 --- a/app/discover/scan_manager.py +++ b/app/discover/scan_manager.py @@ -219,71 +219,74 @@ class ScanManager(Manager): for interval in self.INTERVALS.keys(): self._prepare_scheduled_requests_for_interval(interval) + def handle_scans(self): + self._prepare_scheduled_requests() + + # Find a pending request that is waiting the longest time + results = self.scans_collection \ + .find({'status': ScanStatus.PENDING.value, + 'submit_timestamp': {'$ne': None}}) \ + .sort("submit_timestamp", pymongo.ASCENDING) \ + .limit(1) + + # If no scans are pending, sleep for some time + if results.count() == 0: + time.sleep(self.interval) + else: + scan_request = results[0] + env = scan_request.get('environment') + scan_feature = EnvironmentFeatures.SCANNING + if not self.inv.is_feature_supported(env, scan_feature): + self.log.error("Scanning is not supported for env '{}'" + .format(scan_request.get('environment'))) + self._fail_scan(scan_request) + return + + scan_request['start_timestamp'] = datetime.datetime.utcnow() + scan_request['status'] = ScanStatus.RUNNING.value + self._update_document(scan_request) + + # Prepare scan arguments and run the scan with them + try: + scan_args = self._build_scan_args(scan_request) + + self.log.info("Starting scan for '{}' environment" + .format(scan_args.get('env'))) + self.log.debug("Scan arguments: {}".format(scan_args)) + result, message = ScanController().run(scan_args) + except ScanArgumentsError as e: + self.log.error("Scan request '{id}' " + "has invalid arguments. " + "Errors:\n{errors}" + .format(id=scan_request['_id'], + errors=e)) + self._fail_scan(scan_request) + except Exception as e: + self.log.exception(e) + self.log.error("Scan request '{}' has failed." + .format(scan_request['_id'])) + self._fail_scan(scan_request) + else: + # Check is scan returned success + if not result: + self.log.error(message) + self.log.error("Scan request '{}' has failed." + .format(scan_request['_id'])) + self._fail_scan(scan_request) + return + + # update the status and timestamps. + self.log.info("Request '{}' has been scanned. ({})" + .format(scan_request['_id'], message)) + end_time = datetime.datetime.utcnow() + scan_request['end_timestamp'] = end_time + self._complete_scan(scan_request, message) + def do_action(self): self._clean_up() try: while True: - self._prepare_scheduled_requests() - - # Find a pending request that is waiting the longest time - results = self.scans_collection \ - .find({'status': ScanStatus.PENDING.value, - 'submit_timestamp': {'$ne': None}}) \ - .sort("submit_timestamp", pymongo.ASCENDING) \ - .limit(1) - - # If no scans are pending, sleep for some time - if results.count() == 0: - time.sleep(self.interval) - else: - scan_request = results[0] - env = scan_request.get('environment') - scan_feature = EnvironmentFeatures.SCANNING - if not self.inv.is_feature_supported(env, scan_feature): - self.log.error("Scanning is not supported for env '{}'" - .format(scan_request.get('environment'))) - self._fail_scan(scan_request) - continue - - scan_request['start_timestamp'] = datetime.datetime.utcnow() - scan_request['status'] = ScanStatus.RUNNING.value - self._update_document(scan_request) - - # Prepare scan arguments and run the scan with them - try: - scan_args = self._build_scan_args(scan_request) - - self.log.info("Starting scan for '{}' environment" - .format(scan_args.get('env'))) - self.log.debug("Scan arguments: {}".format(scan_args)) - result, message = ScanController().run(scan_args) - except ScanArgumentsError as e: - self.log.error("Scan request '{id}' " - "has invalid arguments. " - "Errors:\n{errors}" - .format(id=scan_request['_id'], - errors=e)) - self._fail_scan(scan_request) - except Exception as e: - self.log.exception(e) - self.log.error("Scan request '{}' has failed." - .format(scan_request['_id'])) - self._fail_scan(scan_request) - else: - # Check is scan returned success - if not result: - self.log.error(message) - self.log.error("Scan request '{}' has failed." - .format(scan_request['_id'])) - self._fail_scan(scan_request) - continue - - # update the status and timestamps. - self.log.info("Request '{}' has been scanned. ({})" - .format(scan_request['_id'], message)) - end_time = datetime.datetime.utcnow() - scan_request['end_timestamp'] = end_time - self._complete_scan(scan_request, message) + self.handle_scans() finally: self._clean_up() diff --git a/app/discover/scan_metadata_parser.py b/app/discover/scan_metadata_parser.py index df27e18..8757f79 100644 --- a/app/discover/scan_metadata_parser.py +++ b/app/discover/scan_metadata_parser.py @@ -49,21 +49,28 @@ class ScanMetadataParser(MetadataParser): self.add_error('missing or empty fetcher in scanner {} type #{}' .format(scanner_name, str(type_index))) elif isinstance(fetcher, str): + error_str = None try: - module_name = ClassResolver.get_module_file_by_class_name(fetcher) + get_module = ClassResolver.get_module_file_by_class_name + module_name = get_module(fetcher) fetcher_package = module_name.split("_")[0] if package: fetcher_package = ".".join((package, fetcher_package)) - instance = ClassResolver.get_instance_of_class(package_name=fetcher_package, - module_name=module_name, - class_name=fetcher) - except ValueError: - instance = None - if not instance: + # get the fetcher qualified class but not a class instance + # instances will be created just-in-time (before fetching): + # this avoids init of access classes not needed in some envs + get_class = ClassResolver.get_fully_qualified_class + class_qualified = get_class(fetcher, fetcher_package, + module_name) + except ValueError as e: + class_qualified = None + error_str = str(e) + if not class_qualified: self.add_error('failed to find fetcher class {} in scanner {}' - ' type #{}' - .format(fetcher, scanner_name, type_index)) - scan_type[self.FETCHER] = instance + ' type #{} ({})' + .format(fetcher, scanner_name, type_index, + error_str)) + scan_type[self.FETCHER] = class_qualified elif isinstance(fetcher, dict): is_folder = fetcher.get('folder', False) if not is_folder: @@ -81,7 +88,6 @@ class ScanMetadataParser(MetadataParser): def validate_children_scanner(self, scanner_name: str, type_index: int, scanners: dict, scan_type: dict): - scanner = scanners[scanner_name] if 'children_scanner' in scan_type: children_scanner = scan_type.get('children_scanner') if not isinstance(children_scanner, str): diff --git a/app/discover/scanner.py b/app/discover/scanner.py index 1fbcc68..8aac40b 100644 --- a/app/discover/scanner.py +++ b/app/discover/scanner.py @@ -26,6 +26,10 @@ from utils.ssh_connection import SshError class Scanner(Fetcher): + + ENV_TYPE_OPENSTACK = 'OpenStack' + ENV_TYPE_KUBERNETES = 'Kubernetes' + config = None environment = None env = None @@ -82,16 +86,21 @@ class Scanner(Fetcher): def check_type_env(self, type_to_fetch): # check if type is to be run in this environment - if "environment_condition" not in type_to_fetch: - return True - env_cond = type_to_fetch.get("environment_condition", {}) + basic_cond = {'environment_type': self.ENV_TYPE_OPENSTACK} + env_cond = type_to_fetch.get("environment_condition", {}) \ + if "environment_condition" in type_to_fetch \ + else basic_cond if not env_cond: - return True + env_cond = basic_cond + if 'environment_type' not in env_cond: + env_cond.update(basic_cond) if not isinstance(env_cond, dict): self.log.warn('illegal environment_condition given ' 'for type {}'.format(type_to_fetch['type'])) return True conf = self.config.get_env_config() + if 'environment_type' not in conf: + conf.update(basic_cond) for attr, required_val in env_cond.items(): if attr == "mechanism_drivers": if "mechanism_drivers" not in conf: @@ -120,6 +129,9 @@ class Scanner(Fetcher): # get Fetcher instance fetcher = type_to_fetch["fetcher"] + if not isinstance(fetcher, Fetcher): + type_to_fetch['fetcher'] = fetcher() # make it an instance + fetcher = type_to_fetch["fetcher"] fetcher.set_env(self.get_env()) # get children_scanner instance @@ -254,7 +266,6 @@ class Scanner(Fetcher): def load_link_finders_metadata(self): parser = FindLinksMetadataParser() - conf = self.config.get_env_config() finders_file = os.path.join(self.get_run_app_path(), 'config', FindLinksMetadataParser.FINDERS_FILE) -- cgit 1.2.3-korg