Sample oslo.metrics codes

This patch is a sample oslo.metrics code to gather oslo.messaging's
metrics and export the metrics to prometheus. Some parts of this code
doesn't follow OpenStack standards style, and this lacks test codes.
Please use this patch as PoC for oslo.metrics.

Co-Authored-By: Yuki Nishiwaki <yuki.nishiwaki@linecorp.com>
Change-Id: I9434d11466e7626fdbebd1340a8bb3d664518bd1

.gitignore

@@ -0,0 +1,5 @@
+build
+dev
+*.egg-info
+test.sock
+*.pyc

README.rst

@@ -1,10 +1,6 @@
-============
+====================
-oslo.metrics
+Oslo Metrics Library
-============
+====================
 
-This library will allow instrumentation at Oslo library level for
+This Oslo metrics API supports collecting metrics data from other Oslo
-oslo.messaging and other common abstraction libraries, to get
+libraries and exposing the metrics data to monitoring system.
-operational metrics.
-
-This library is currently under development, initial code drop is
-expected in June, 2020.

main.py

@@ -0,0 +1,89 @@
+# Copyright 2020 LINE Corp.
+#
+#    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 os
+import select
+import socket
+import sys
+import threading
+from wsgiref.simple_server import make_server
+
+from oslo_config import cfg
+from oslo_log import log as logging
+from prometheus_client import make_wsgi_app
+
+from oslo_metrics import message_router
+
+
+oslo_metrics_configs = [
+    cfg.StrOpt('metrics_socket_file',
+               default='/var/tmp/metrics_collector.sock',
+               help='Unix domain socket file to be used'
+                    'to send rpc related metrics'),
+]
+cfg.CONF.register_opts(oslo_metrics_configs, group='oslo_metrics')
+
+
+LOG = logging.getLogger(__name__)
+CONF = cfg.CONF
+logging.register_options(CONF)
+logging.setup(CONF, 'oslo-metrics')
+LOG.logger.setLevel(logging.DEBUG)
+
+
+class MetricsListener():
+
+    def __init__(self, socket_path):
+        self.socket_path = socket_path
+        self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+        self.socket.bind(self.socket_path)
+        self.start = True
+        self.router = message_router.MessageRouter()
+
+    def serve(self):
+        while self.start:
+            readable, writable, exceptional = select.select([self.socket], [], [], 1)
+            if len(readable) == 0:
+                continue
+            try:
+                LOG.debug("wait for socket.recv")
+                # 1 message size should be smaller than 65565
+                msg = self.socket.recv(65565)
+                LOG.debug("got message")
+                self.router.process(msg)
+            except socket.timeout:
+                pass
+
+    def stop(self):
+        self.socket.close()
+        self.start = False
+
+if __name__ == "__main__":
+    cfg.CONF(sys.argv[1:])
+    m = MetricsListener(cfg.CONF.oslo_metrics.metrics_socket_file)
+    mt = threading.Thread(target=m.serve)
+    LOG.info("Start oslo.metrics")
+    mt.start()
+
+    app = make_wsgi_app()
+    try:
+        httpd = make_server('', 3000, app)
+        httpd.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        LOG.info("Try to stop...")
+        os.remove(cfg.CONF.oslo_metrics.metrics_socket_file)
+        m.stop()
+        httpd.server_close()

oslo_metrics/message_router.py

@@ -0,0 +1,80 @@
+# Copyright 2020 LINE Corp.
+#
+#    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.
+
+from oslo_log import log as logging
+from oslo_utils import importutils
+
+from oslo_metrics import message_type
+
+LOG = logging.getLogger(__name__)
+
+
+MODULE_LISTS = [
+    "oslo_metrics.metrics.oslo_messaging",
+]
+
+
+class MessageRouter():
+
+    def __init__(self):
+        self.modules = {}
+        for m_str in MODULE_LISTS:
+            mod = importutils.try_import(m_str, False)
+            if not mod:
+                LOG.error("Failed to load module %s" % m_str)
+            self.modules[m_str.split('.')[-1]] = mod
+
+    def process(self, raw_string):
+        try:
+            metric = message_type.Metric.from_json(raw_string.decode())
+            self.dispatch(metric)
+        except Exception as e:
+            LOG.error("Failed to parse: %s", e)
+
+    def dispatch(self, metric):
+        if metric.module not in self.modules:
+            LOG.error("Failed to lookup modules by %s" % metric.module)
+            return
+        mod = self.modules.get(metric.module)
+
+        # Get metric
+        try:
+            metric_definition = getattr(mod, metric.name)
+        except AttributeError as e:
+            LOG.error("Failed to load metrics %s: %s" % (metric.name, e))
+            return
+
+        # Get labels
+        try:
+            metric_with_label = getattr(metric_definition, "labels")
+            metric_with_label = metric_with_label(**metric.labels)
+        except AttributeError as e:
+            LOG.error("Failed to load labales func from metrics %s: %s" %
+                      (metric.name, e))
+            return
+        LOG.info("Get labels with %s: %s" % (metric.name, metric.labels))
+
+        # perform action
+        try:
+            embed_action = getattr(metric_with_label, metric.action.action)
+            if metric.action.value is not None:
+                embed_action(metric.action.value)
+            else:
+                embed_action()
+        except AttributeError as e:
+            LOG.error("Failed to perform metric actionv %s, %s: %s" %
+                      (metric.action.action, metric.action.value, e))
+            return
+        LOG.info("Perform action %s for %s metrics" %
+                 (metric.action.action, metric.name))

oslo_metrics/message_type.py

@@ -0,0 +1,83 @@
+import json
+
+
+class UnSupportedMetricActionError(Exception):
+    pass
+
+
+class MetricValidationError(Exception):
+    pass
+
+
+class MetricAction():
+    actions = ['inc', 'observe']
+
+    def __init__(self, action, value):
+        if action not in self.actions:
+            raise UnSupportedMetricActionError(
+                "%s action is not supported" % action)
+        self.action = action
+        self.value = value
+
+    @classmethod
+    def validate(cls, metric_action_dict):
+        if "value" not in metric_action_dict:
+            raise MetricValidationError("action need 'value' field")
+        if "action" not in metric_action_dict:
+            raise MetricValidationError("action need 'action' field")
+        if metric_action_dict["action"] not in cls.actions:
+            raise MetricValidationError(
+                    "action should be choosen from %s" % cls.actions)
+
+    @classmethod
+    def from_dict(cls, metric_action_dict):
+        return cls(
+            metric_action_dict["action"],
+            metric_action_dict["value"]
+        )
+
+
+class Metric():
+    def __init__(self, module, name, action, **labels):
+        self.module = module
+        self.name = name
+        self.action = action
+        self.labels = labels
+
+    def to_json(self):
+        raw = {
+            "module": self.module,
+            "name": self.name,
+            "action": {
+                "value": self.action.value,
+                "action": self.action.action
+            },
+            "labels": self.labels
+        }
+        return json.dumps(raw)
+
+    @classmethod
+    def from_json(cls, encoded):
+        metric_dict = json.loads(encoded)
+        cls._validate(metric_dict)
+        return Metric(
+            metric_dict["module"],
+            metric_dict["name"],
+            MetricAction.from_dict(metric_dict["action"]),
+            **metric_dict["labels"])
+
+    @classmethod
+    def _validate(cls, metric_dict):
+        if "module" not in metric_dict:
+            raise MetricValidationError("module should be specified")
+
+        if "name" not in metric_dict:
+            raise MetricValidationError("name should be specified")
+
+        if "action" not in metric_dict:
+            raise MetricValidationError("action should be specified")
+
+        if "labels" not in metric_dict:
+            raise MetricValidationError("labels should be specified")
+
+        MetricAction.validate(metric_dict["action"])

oslo_metrics/metrics/oslo_messaging.py

@@ -0,0 +1,68 @@
+# Copyright 2019 LINE Corp.
+#
+#    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 prometheus_client
+
+standard_labels_for_server = [
+    'exchange', 'topic', 'server', 'endpoint', 'namespace',
+    'version', 'method', 'process'
+]
+standard_labels_for_client = [
+    'call_type', 'exchange', 'topic', 'namespace', 'version',
+    'server', 'fanout', 'process'
+]
+
+
+# RPC Server Metrics
+rpc_server_count_for_exception = prometheus_client.Counter(
+    'oslo_messaging_rpc_server_exception',
+    'The number of times to hit Exception',
+    standard_labels_for_server + ['exception', ])
+
+rpc_server_count_for_invocation_start = prometheus_client.Counter(
+    'oslo_messaging_rpc_server_invocation_start',
+    'The number of times to attempt to invoke method. It doesn\'t count'
+    'if rpc server failed to find method from endpoints',
+    standard_labels_for_server)
+
+rpc_server_count_for_invocation_end = prometheus_client.Counter(
+    'oslo_messaging_rpc_server_invocation_end',
+    'The number of times to finish to invoke method.',
+    standard_labels_for_server)
+
+rpc_server_processing_time = prometheus_client.Histogram(
+    'oslo_messaging_rpc_server_processing_second',
+    'rpc server processing time[second]',
+    standard_labels_for_server)
+
+
+# RPC Client Metrics
+rpc_client_count_for_exception = prometheus_client.Counter(
+    'oslo_messaging_rpc_client_exception',
+    'The number of times to hit Exception',
+    standard_labels_for_client + ['exception', ])
+
+rpc_client_count_for_invocation_start = prometheus_client.Counter(
+    'oslo_messaging_rpc_client_invocation_start',
+    'The number of times to invoke method',
+    standard_labels_for_client)
+
+rpc_client_count_for_invocation_end = prometheus_client.Counter(
+    'oslo_messaging_rpc_client_invocation_end',
+    'The number of times to invoke method',
+    standard_labels_for_client)
+
+rpc_client_processing_time = prometheus_client.Histogram(
+    'oslo_messaging_rpc_client_processing_second',
+    'rpc client processing time[second]',
+    standard_labels_for_client)

requirements.txt

@@ -0,0 +1,13 @@
+# The order of packages is significant, because pip processes them in the order
+# of appearance. Changing the order has an impact on the overall integration
+# process, which may cause wedges in the gate later.
+
+pbr!=2.1.0,>=2.0.0 # Apache-2.0
+
+# General
+oslo.utils==3.41.0
+oslo.log==3.44.0
+oslo.config==6.9.0
+
+# Metrics Exporter
+prometheus-client==0.6.0

setup.cfg

@@ -0,0 +1,26 @@
+[metadata]
+name = oslo.metrics
+author = OpenStack
+author-email = openstack-discuss@lists.openstack.org
+summary = Oslo Metrics API
+description-file =
+    README.rst
+home-page = https://opendev.org/openstack/oslo.metrics
+python-requires = >=3.6
+classifier =
+    Environment :: OpenStack
+    Intended Audience :: Developers
+    Intended Audience :: Information Technology
+    License :: OSI Approved :: Apache Software License
+    Operating System :: OS Independent
+    Programming Language :: Python
+    Programming Language :: Python :: 3
+    Programming Language :: Python :: 3.6
+    Programming Language :: Python :: 3.7
+    Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3 :: Only
+    Programming Language :: Python :: Implementation :: CPython
+
+[files]
+packages =
+    oslo_metrics

setup.py

@@ -0,0 +1,20 @@
+# Copyright (c) 2020 LINE Corp.
+#
+# 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 setuptools
+
+setuptools.setup(
+    setup_requires=['pbr>=2.0.0'],
+    pbr=True)