[Metric] new cython interface for python worker metric (#9469)

python/ray/_raylet.pyx

@@ -112,6 +112,7 @@ include "includes/common.pxi"
 include "includes/serialization.pxi"
 include "includes/libcoreworker.pxi"
 include "includes/global_state_accessor.pxi"
+include "includes/metric.pxi"
 
 # Expose GCC & Clang macro to report
 # whether C++ optimizations were enabled during compilation.

python/ray/experimental/metrics.py

@@ -0,0 +1,38 @@
+from ray._raylet import (
+    Count,
+    Histogram,
+    Gauge,
+    Sum,
+)  # noqa: E402
+"""Metric/Stats module for worker.
+
+This module is responsible for providing four classes mapping from stats of
+cpp.
+
+How to use:
+  For Count, Gauge and Sum, we may define a metric like this following:
+    gauge = Gauge(
+      'ray.worker.metric',
+      'description',
+      'unit',
+      ['tagk1', 'tagk2']).
+  The last parameter is default tag map. You can use gauge.record(1.0) with
+  default tags or gauge.record(1.0, {'tagk1', 'tagv1'}) that means the tagk1
+  is updating in tagv1.
+
+  It's addtional boundaries to Histogram measurement,
+    histogram = Histogram(
+      'ray.worker.histogram1',
+      'a', 'b', [1.0, 2.0],
+      ['tagk1'])
+
+  Recommended metric name pattern : ray.{component_name}.{module_name}, and
+  name format must be in [0-9a-zA-Z].
+"""
+
+__all__ = [
+    "Count",
+    "Histogram",
+    "Gauge",
+    "Sum",
+]

python/ray/includes/metric.pxd

@@ -0,0 +1,45 @@
+from libcpp.string cimport string as c_string
+from libcpp.unordered_map cimport unordered_map
+from libcpp.vector cimport vector as c_vector
+
+cdef extern from "opencensus/tags/tag_key.h" nogil:
+    cdef cppclass CTagKey "opencensus::tags::TagKey":
+        @staticmethod
+        CTagKey Register(c_string &name)
+        const c_string &name() const
+
+cdef extern from "ray/stats/metric.h" nogil:
+    cdef cppclass CMetric "ray::stats::Metric":
+        CMetric(const c_string &name,
+               const c_string &description,
+               const c_string &unit,
+               const c_vector[CTagKey] &tag_keys)
+        c_string GetName() const
+        void Record(double value)
+        void Record(double value, 
+                unordered_map[c_string, c_string] &tags)
+
+    cdef cppclass CGauge "ray::stats::Gauge":
+        CGauge(const c_string &name,
+               const c_string &description,
+               const c_string &unit,
+               const c_vector[CTagKey] &tag_keys)
+
+    cdef cppclass CCount "ray::stats::Count":
+        CCount(const c_string &name,
+               const c_string &description,
+               const c_string &unit,
+               const c_vector[CTagKey] &tag_keys)
+
+    cdef cppclass CSum "ray::stats::Sum":
+        CSum(const c_string &name,
+             const c_string &description,
+             const c_string &unit,
+             const c_vector[CTagKey] &tag_keys)
+
+    cdef cppclass CHistogram "ray::stats::Histogram":
+        CHistogram(const c_string &name,
+                   const c_string &description,
+                   const c_string &unit,
+                   const c_vector[double] &boundaries,
+                   const c_vector[CTagKey] &tag_keys)

python/ray/includes/metric.pxi

@@ -0,0 +1,170 @@
+from ray.includes.metric cimport (
+    CCount,
+    CGauge,
+    CHistogram,
+    CTagKey,
+    CSum,
+    CMetric,
+)
+from libcpp.memory cimport unique_ptr
+from libcpp.string cimport string as c_string
+from libcpp.unordered_map cimport unordered_map
+from libcpp.vector cimport vector as c_vector
+
+cdef class TagKey:
+    """Cython wrapper class of C++ `opencensus::stats::TagKey`."""
+    cdef c_string name
+
+    def __init__(self, name):
+        self.name = name.encode("ascii")
+        CTagKey.Register(self.name)
+    
+    def name(self):
+        return self.name
+
+
+cdef class Metric:
+    """Cython wrapper class of C++ `ray::stats::Metric`.
+
+        It's an abstract class of all metric types.
+    """
+    cdef:
+        unique_ptr[CMetric] metric
+        c_vector[CTagKey] c_tag_keys
+
+    def __init__(self, tag_keys):
+        for tag_key in tag_keys:
+            self.c_tag_keys.push_back(CTagKey.Register(tag_key.encode("ascii")))
+    
+    def record(self, value, tags=None):
+        """Record a measurement of metric.
+
+           Flush a metric raw point to stats module with a key-value dict tags.
+        Args:
+            value (double): metric name.
+            tags (dict): default none.
+        """
+        cdef unordered_map[c_string, c_string] c_tags
+        cdef double c_value
+        # Default tags will be exported if it's empty map.
+        if tags:
+            for tag_k, tag_v in tags.items():
+                c_tags[tag_v.encode("ascii")] = tag_v.encode("ascii")
+        c_value = value
+        with nogil:
+            self.metric.get().Record(c_value, c_tags)
+
+    def get_name(self):
+        return self.metric.get().GetName()
+
+
+cdef class Gauge(Metric):
+    """Cython wrapper class of C++ `ray::stats::Gauge`.
+
+       Gauge: Keeps the last recorded value, drops everything before.
+    """
+    def __init__(self, name, description, unit, tag_keys):
+        """Create a gauge metric
+
+        Args:
+            name (string): metric name.
+            description (string): description of this metric.
+            unit (string): measure unit of this metric.
+            tag_keys (list): a list of tay keys in string format.
+        """
+        super().__init__(tag_keys)
+
+        self.metric.reset(
+            new CGauge(
+                name.encode("ascii"),
+                description.encode("ascii"),
+                unit.encode("ascii"),
+                self.c_tag_keys
+            )
+        )
+
+
+cdef class Count(Metric):
+    """Cython wrapper class of C++ `ray::stats::Count`.
+
+       Count: The count of the number of metric points.
+    """
+    def __init__(self, name, description, unit, tag_keys):
+        """Create a count metric
+
+        Args:
+            name (string): metric name.
+            description (string): description of this metric.
+            unit (string): measure unit of this metric.
+            tag_keys (list): a list of tay keys in string format.
+        """
+        super().__init__(tag_keys)
+
+        self.metric.reset(
+            new CCount(
+                name.encode("ascii"),
+                description.encode("ascii"),
+                unit.encode("ascii"),
+                self.c_tag_keys
+            )
+        )
+
+
+cdef class Sum(Metric):
+    """Cython wrapper class of C++ `ray::stats::Sum`.
+
+       Sum: A sum up of the metric points.
+    """
+    def __init__(self, name, description, unit, tag_keys):
+        """Create a sum metric
+
+        Args:
+            name (string): metric name.
+            description (string): description of this metric.
+            unit (string): measure unit of this metric.
+            tag_keys (list): a list of tay keys in string format.
+        """
+
+        super().__init__(tag_keys)
+
+        self.metric.reset(
+            new CSum(
+                name.encode("ascii"),
+                description.encode("ascii"),
+                unit.encode("ascii"),
+                self.c_tag_keys
+            )
+        )
+
+
+cdef class Histogram(Metric):
+    """Cython wrapper class of C++ `ray::stats::Histogram`.
+
+       Histogram: Histogram distribution of metric points.
+    """
+    def __init__(self, name, description, unit, boundaries, tag_keys):
+        """Create a sum metric
+
+        Args:
+            name (string): metric name.
+            description (string): description of this metric.
+            unit (string): measure unit of this metric.
+            boundaries (list): a double type list boundaries of histogram.
+            tag_keys (list): a list of tay key in string format.
+        """
+
+        super().__init__(tag_keys)
+
+        cdef c_vector[double] c_boundaries
+        for value in boundaries:
+            c_boundaries.push_back(value)
+
+        self.metric.reset(
+            new CHistogram(
+                name.encode("ascii"),
+                description.encode("ascii"),
+                unit.encode("ascii"),
+                c_boundaries,
+                self.c_tag_keys
+            )
+        )