Ray serve consumer

README.md

@@ -67,7 +67,7 @@ OSPP 2022 and GSOC 2022 student research outcomes.
 
 **Metric Anomaly Detection and Visualizations**
 
-TBD - Soon to be added
+![img.png](docs/static/metric-detector.png)
 
 ### Roadmap
 

engine/models/metric/detectors/spot_detector.py

@@ -96,7 +96,7 @@ def jac_w(y, t):
             ) * (1 - vs)
             jac_vs = np.divide(
                 1, t, out=np.array(1 / epsilon), where=t != 0
-            ) * (-vs + np.mean(1 / s**2))
+            ) * (-vs + np.mean(1 / s ** 2))
             return us * jac_vs + vs * jac_us
 
         self.peaks[side][self.peaks[side] == 0] = epsilon
@@ -117,7 +117,7 @@ def jac_w(y, t):
         )
         c = 2 * np.divide(
             y_mean - y_min,
-            y_min**2,
+            y_min ** 2,
             out=np.array((y_mean - y_min) / epsilon + epsilon),
             where=y_min != 0,
         )
@@ -198,7 +198,7 @@ def obj_fun(x_sampled, f, jac):
             i = 0
             for x in x_sampled:
                 fx = f(x)
-                g = g + fx**2
+                g = g + fx ** 2
                 j[i] = 2 * fx * jac(x)
                 i = i + 1
             return g, j
@@ -312,7 +312,6 @@ def _cal_back_mean(self, data):
 
     def fit(self, data: Union[float, int]):
 
-
         self.back_mean_window.append(data)
 
         if self.index >= self.back_mean_len:
@@ -333,7 +332,10 @@ def fit(self, data: Union[float, int]):
             if (
                 abs(
                     np.divide(
-                        data - last_data, last_data, np.array(data), where=last_data != 0
+                        data - last_data,
+                        last_data,
+                        np.array(data),
+                        where=last_data != 0,
                     )
                 )
                 < self.deviance_ratio
@@ -351,15 +353,21 @@ def fit(self, data: Union[float, int]):
 
     def score(self, data: Union[float, int]) -> float:
 
-
         last_data = (
             self.window[-2]
             if self.back_mean_len == 0
             else (data - self.window[-1])
         )
 
         if (
-            abs(np.divide(data - last_data, last_data, np.array(data), where=last_data != 0))
+            abs(
+                np.divide(
+                    data - last_data,
+                    last_data,
+                    np.array(data),
+                    where=last_data != 0,
+                )
+            )
             < self.deviance_ratio
         ):
             score = 0.0

engine/models/metric/serve/deployment.py

@@ -0,0 +1,58 @@
+# Copyright 2022 SkyAPM org
+#
+# 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
+#
+#      https://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 ray
+
+
+@ray.remote(num_cpus=1)
+class RayMetricConsumer(object):
+    def __init__(self, detector) -> None:
+        self.detector = detector
+
+    def run(self, timestamp: str, data: float) -> float:
+        score = self.detector.fit_score(timestamp=timestamp, data=float(data))
+        return score
+
+
+if __name__ == '__main__':
+
+    from engine.models.metric.detectors import SpotDetector
+    import pandas as pd
+
+    df = pd.read_csv(
+        'experiments/metric/data/univarate_dataset.csv', index_col='timestamp'
+    )
+
+    # This for loop can be replaced by a MQ
+    for index, row in df.iterrows():
+        timestamp = index
+        data = row.value
+
+        # Here we setup multi-consumer according to unique name
+        detector = RayMetricConsumer.options(
+            name='cpu.load', lifetime='detached', get_if_exists=True
+        ).remote(SpotDetector())
+
+        score1 = ray.get(
+            detector.run.remote(timestamp=timestamp, data=float(data))
+        )
+
+        detector = RayMetricConsumer.options(
+            name='mem.load', lifetime='detached', get_if_exists=True
+        ).remote(SpotDetector())
+        score2 = ray.get(
+            detector.run.remote(timestamp=timestamp, data=float(data))
+        )
+
+        assert score1 == score2

engine/models/tests/test_detector.py

@@ -13,12 +13,13 @@
 #  limitations under the License.
 
 
-
 import sys
 
 import pandas as pd
 
 from engine.models.metric.detectors import SpotDetector
+from engine.models.metric.serve.deployment import RayMetricConsumer
+import ray
 
 sys.path.append('../../')
 
@@ -27,10 +28,40 @@ def test_detector_uni_dataset():
     df = pd.read_csv(
         'experiments/metric/data/univarate_dataset.csv', index_col='timestamp'
     )
-    detector = SpotDetector()
+    detector = SpotDetector.remote()
 
     for index, row in df.iterrows():
         timestamp = index
         data = row.value
-        score = detector.fit_score(timestamp=timestamp, data=float(data))
+
+        score = ray.get(
+            detector.fit_score.remote(timestamp=timestamp, data=float(data))
+        )
         assert score is None or 0 <= score <= 1
+
+
+def test_multi_detector_uni_dataset():
+    df = pd.read_csv(
+        'experiments/metric/data/univarate_dataset.csv', index_col='timestamp'
+    )
+
+    for index, row in df.iterrows():
+        timestamp = index
+        data = row.value
+
+        detector = RayMetricConsumer.options(
+            name='cpu.load', lifetime='detached', get_if_exists=True
+        ).remote(SpotDetector())
+
+        score1 = ray.get(
+            detector.run.remote(timestamp=timestamp, data=float(data))
+        )
+
+        detector = RayMetricConsumer.options(
+            name='mem.load', lifetime='detached', get_if_exists=True
+        ).remote(SpotDetector())
+        score2 = ray.get(
+            detector.run.remote(timestamp=timestamp, data=float(data))
+        )
+
+        assert score1 == score2