Pawel plesniak/drunc op mon

python/kafkaopmon/OpMonFunction.py

@@ -0,0 +1,17 @@
+class OpMonFunction :
+    def __init__(self,
+                 function,
+                 opmon_id : re.Pattern,
+                 measurement : re.Pattern) :
+        self.function = function
+        self.opmon_id = opmon_id
+        self.measurement = measurement
+
+    def match(self, key : str) -> bool :
+        opmon_id,measure = key.split('/',1)
+        if not self.opmon_id.match(opmon_id) : return False
+        if not self.measurement.match(measure) : return False
+        return True
+
+    def execute(self, e : entry.OpMonEntry ) :
+        self.function(e)

python/kafkaopmon/OpMonPublisher.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+
+import os
+import socket
+import inspect
+from opmonlib.opmon_entry_pb2 import OpMonValue, OpMonId, OpMonEntry 
+import google.protobuf.message as msg
+from google.protobuf.descriptor import FieldDescriptor as fd
+from google.protobuf.timestamp_pb2 import Timestamp
+from datetime import datetime
+from kafka import KafkaProducer
+from typing import Optional
+import logging
+
+class OpMonPublisher:
+    def __init__(
+                    self, 
+                    topic:str,
+                    bootstrap:str = "monkafka.cern.ch:30092", # Removed for if we don't want to use OpMon (i.e. for ssh-standalone)
+                    application_name:str = "python",
+                    package_name:str = "unknown"
+    ) -> None:
+        ## Options from configurations
+        self.application_name = application_name
+        self.package_name = package_name
+        self.bootstrap = bootstrap
+        if not topic.startswith('monitoring.'):
+            topic = 'monitoring.' + topic
+        self.topic = topic
+
+        ## runtime options
+        self.log = logging.getLogger("OpMonPublisher")
+        self.log.info("Starting Kafka producer")
+        self.producer = KafkaProducer(
+                                        bootstrap_servers=self.bootstrap,
+                                        value_serializer=lambda v: v.SerializeToString(),
+                                        key_serializer=lambda k: str(k).encode('utf-8')
+        )
+        self.log.info("Initialized Kafka producer")        
+
+    def publish(
+        self,
+        session:str,
+        application:str,
+        message:msg,
+        custom_origin:Optional[dict[str,str]] = {"": ""},
+        substructure:Optional[str] = "",
+    ):
+        """Create an OpMonEntry and send it to Kafka."""
+        data_dict = self.map_message(message)
+        opmon_id = OpMonId(
+            session = session,
+            application = application,
+            substructure = substructure
+        )
+        t = Timestamp()
+        opmon_metric = OpMonEntry(
+            time = t.GetCurrentTime(),
+            origin = opmon_id,
+            custom_origin = custom_origin,
+            measurement = message.DESCRIPTOR.name,
+            data = data_dict,
+        )
+        return self.producer.send(opmon_metric)
+
+    def map_message(self, message:msg):
+        message_dict = {}
+        for name, descriptor in message.DESCRIPTOR.fields_by_name.items():
+            message_dict[name] = self.map_entry(getattr(message, name), descriptor.cpp_type)
+        return message_dict 
+
+    def map_entry(self, value, field_type:int):
+        formatted_OpMonValue = OpMonValue() 
+        match field_type:
+            case fd.CPPTYPE_INT32:
+                formatted_OpMonValue.int4_value = value
+            case fd.CPPTYPE_INT64:
+                formatted_OpMonValue.int8_value = value
+            case fd.CPPTYPE_UINT32:
+                formatted_OpMonValue.uint4_value = value
+            case fd.CPPTYPE_UINT64:
+                formatted_OpMonValue.uint8_value = value
+            case fd.CPPTYPE_DOUBLE:
+                formatted_OpMonValue.double_value = value
+            case fd.CPPTYPE_FLOAT:
+                formatted_OpMonValue.float_value = value
+            case fd.CPPTYPE_BOOL:
+                formatted_OpMonValue.boolean_value = value
+            case fd.CPPTYPE_STRING:
+                formatted_OpMonValue.string_value = value
+            case _:
+                raise ValueError("Value is of a non-supported type.")
+        return formatted_OpMonValue

python/kafkaopmon/OpMonSubscriber.py

@@ -11,25 +11,7 @@
 
 import opmonlib.opmon_entry_pb2 as entry
 import google.protobuf.message as msg
-
-class OpMonFunction :
-    def __init__(self,
-                 function,
-                 opmon_id : re.Pattern,
-                 measurement : re.Pattern) :
-        self.function = function
-        self.opmon_id = opmon_id
-        self.measurement = measurement
-
-    def match(self, key : str) -> bool :
-        opmon_id,measure = key.split('/',1);
-        if not self.opmon_id.match(opmon_id) : return False
-        if not self.measurement.match(measure) : return False
-        return True
-
-    def execute(self, e : entry.OpMonEntry ) :
-        self.function(e)
-
+from kafkaopmon import OpMonFunction
 
 class  OpMonSubscriber:
     def __init__(self, bootstrap, group_id=None, timeout_ms=500, topics=["opmon_stream"]) :

python/kafkaopmon/__init__ .py

@@ -0,0 +1 @@
+__version__ = '0.0.1'

setup.cfg

@@ -0,0 +1,16 @@
+[metadata]
+name = kafkaopmon
+version = attr: kafkaopmon.__version__
+url = https://github.com/DUNE-DAQ/kafkaopmon
+long_description = file: docs/README.md
+long_description_content_type = text/markdown
+
+[options]
+packages = find_namespace:
+package_dir = =python
+include_package_data = true
+python_requires = >= 3.6
+# Dependencies are in setup.py for GitHub's dependency graph.
+
+[options.packages.find]
+where = python

setup.py

@@ -0,0 +1,28 @@
+from setuptools import setup
+
+# Metadata goes in setup.cfg. These are here for GitHub's dependency graph.
+setup(
+    name="kafkaopmon",
+    install_requires=[
+        "os",
+		"re",
+		"socket",
+		"threading",
+		"logging",
+		"getpass",
+		"sys",
+		"inspect",
+		"datetime",
+		"time",
+		"typing",
+        "googleapis-common-protos",
+        "kafka-python",
+        "rich",
+        "sh"
+
+    ],
+    extras_require={"develop": [
+        "ipdb",
+        "ipython"
+    ]}
+)