Source code for osprofiler.drivers.base

# Copyright 2016 Mirantis Inc.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import datetime
import logging
from urllib import parse as urlparse

from osprofiler import _utils

LOG = logging.getLogger(__name__)


[docs]def get_driver(connection_string, *args, **kwargs): """Create driver's instance according to specified connection string""" # NOTE(ayelistratov) Backward compatibility with old Messaging notation # Remove after patching all OS services # NOTE(ishakhat) Raise exception when ParsedResult.scheme is empty if "://" not in connection_string: connection_string += "://" parsed_connection = urlparse.urlparse(connection_string) LOG.debug("String %s looks like a connection string, trying it.", connection_string) backend = parsed_connection.scheme # NOTE(toabctl): To be able to use the connection_string for as sqlalchemy # connection string, transform the backend to the correct driver # See https://docs.sqlalchemy.org/en/latest/core/engines.html#database-urls if backend in ["mysql", "mysql+pymysql", "mysql+mysqldb", "postgresql", "postgresql+psycopg2"]: backend = "sqlalchemy" for driver in _utils.itersubclasses(Driver): if backend == driver.get_name(): return driver(connection_string, *args, **kwargs) raise ValueError("Driver not found for connection string: " "%s" % connection_string)
[docs]class Driver(object): """Base Driver class. This class provides protected common methods that do not rely on a specific storage backend. Public methods notify() and/or get_report(), which require using storage backend API, must be overridden and implemented by any class derived from this class. """ default_trace_fields = {"base_id", "timestamp"} def __init__(self, connection_str, project=None, service=None, host=None, **kwargs): self.connection_str = connection_str self.project = project self.service = service self.host = host self.result = {} self.started_at = None self.finished_at = None # Last trace started time self.last_started_at = None profiler_config = kwargs.get("conf", {}).get("profiler", {}) if hasattr(profiler_config, "filter_error_trace"): self.filter_error_trace = profiler_config.filter_error_trace else: self.filter_error_trace = False
[docs] def notify(self, info, **kwargs): """This method will be called on each notifier.notify() call. To add new drivers you should, create new subclass of this class and implement notify method. :param info: Contains information about trace element. In payload dict there are always 3 ids: "base_id" - uuid that is common for all notifications related to one trace. Used to simplify retrieving of all trace elements from the backend. "parent_id" - uuid of parent element in trace "trace_id" - uuid of current element in trace With parent_id and trace_id it's quite simple to build tree of trace elements, which simplify analyze of trace. """ raise NotImplementedError("{0}: This method is either not supported " "or has to be overridden".format( self.get_name()))
[docs] def get_report(self, base_id): """Forms and returns report composed from the stored notifications. :param base_id: Base id of trace elements. """ raise NotImplementedError("{0}: This method is either not supported " "or has to be overridden".format( self.get_name()))
[docs] @classmethod def get_name(cls): """Returns backend specific name for the driver.""" return cls.__name__
[docs] def list_traces(self, fields=None): """Query all traces from the storage. :param fields: Set of trace fields to return. Defaults to 'base_id' and 'timestamp' :returns: List of traces, where each trace is a dictionary containing at least `base_id` and `timestamp`. """ raise NotImplementedError("{0}: This method is either not supported " "or has to be overridden".format( self.get_name()))
[docs] def list_error_traces(self): """Query all error traces from the storage. :return List of traces, where each trace is a dictionary containing `base_id` and `timestamp`. """ raise NotImplementedError("{0}: This method is either not supported " "or has to be overridden".format( self.get_name()))
@staticmethod def _build_tree(nodes): """Builds the tree (forest) data structure based on the list of nodes. Tree building works in O(n*log(n)). :param nodes: dict of nodes, where each node is a dictionary with fields "parent_id", "trace_id", "info" :returns: list of top level ("root") nodes in form of dictionaries, each containing the "info" and "children" fields, where "children" is the list of child nodes ("children" will be empty for leafs) """ tree = [] for trace_id in nodes: node = nodes[trace_id] node.setdefault("children", []) parent_id = node["parent_id"] if parent_id in nodes: nodes[parent_id].setdefault("children", []) nodes[parent_id]["children"].append(node) else: tree.append(node) # no parent => top-level node for trace_id in nodes: nodes[trace_id]["children"].sort( key=lambda x: x["info"]["started"]) return sorted(tree, key=lambda x: x["info"]["started"]) def _append_results(self, trace_id, parent_id, name, project, service, host, timestamp, raw_payload=None): """Appends the notification to the dictionary of notifications. :param trace_id: UUID of current trace point :param parent_id: UUID of parent trace point :param name: name of operation :param project: project name :param service: service name :param host: host name or FQDN :param timestamp: Unicode-style timestamp matching the pattern "%Y-%m-%dT%H:%M:%S.%f" , e.g. 2016-04-18T17:42:10.77 :param raw_payload: raw notification without any filtering, with all fields included """ timestamp = datetime.datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%f") if trace_id not in self.result: self.result[trace_id] = { "info": { "name": name.split("-")[0], "project": project, "service": service, "host": host, }, "trace_id": trace_id, "parent_id": parent_id, } self.result[trace_id]["info"]["meta.raw_payload.%s" % name] = raw_payload if name.endswith("stop"): self.result[trace_id]["info"]["finished"] = timestamp self.result[trace_id]["info"]["exception"] = "None" if raw_payload and "info" in raw_payload: exc = raw_payload["info"].get("etype", "None") self.result[trace_id]["info"]["exception"] = exc else: self.result[trace_id]["info"]["started"] = timestamp if not self.last_started_at or self.last_started_at < timestamp: self.last_started_at = timestamp if not self.started_at or self.started_at > timestamp: self.started_at = timestamp if not self.finished_at or self.finished_at < timestamp: self.finished_at = timestamp def _parse_results(self): """Parses Driver's notifications placed by _append_results() . :returns: full profiling report """ def msec(dt): # NOTE(boris-42): Unfortunately this is the simplest way that works # in py26 and py27 microsec = (dt.microseconds + (dt.seconds + dt.days * 24 * 3600) * 1e6) return int(microsec / 1000.0) stats = {} for r in self.result.values(): # NOTE(boris-42): We are not able to guarantee that the backend # consumed all messages => so we should at make duration 0ms. if "started" not in r["info"]: r["info"]["started"] = r["info"]["finished"] if "finished" not in r["info"]: r["info"]["finished"] = r["info"]["started"] op_type = r["info"]["name"] op_started = msec(r["info"]["started"] - self.started_at) op_finished = msec(r["info"]["finished"] - self.started_at) duration = op_finished - op_started r["info"]["started"] = op_started r["info"]["finished"] = op_finished if op_type not in stats: stats[op_type] = { "count": 1, "duration": duration } else: stats[op_type]["count"] += 1 stats[op_type]["duration"] += duration return { "info": { "name": "total", "started": 0, "finished": msec( self.finished_at - self.started_at ) if self.started_at else None, "last_trace_started": msec( self.last_started_at - self.started_at ) if self.started_at else None }, "children": self._build_tree(self.result), "stats": stats }