Adds interface for dd integration

processors/datadog_webclient_processor.py

@@ -0,0 +1,46 @@
+import logging
+from typing import Dict
+
+from datadog import api, initialize as dd_api_initialize
+
+logger = logging.getLogger(__name__)
+
+
+class DatadogRestApiProcessor:
+
+    def __init__(self, dd_api_key, dd_app_key):
+        options = {
+            'api_key': dd_api_key,
+            'app_key': dd_app_key,
+        }
+        dd_api_initialize(**options)
+
+    def fetch_monitors(self, monitor_id: int = None, params: Dict = dict({})):
+        try:
+            if monitor_id is not None:
+                monitors = [api.Monitor.get(id=monitor_id, **params)]
+            else:
+                monitors = api.Monitor.get_all(**params)
+            if not monitors:
+                logger.error(f"No monitors found")
+            print("Found total monitors: ", len(monitors))
+            return monitors
+        except Exception as e:
+            logger.error(f"Exception occurred while fetching monitors with error: {e}")
+            return None
+
+    def fetch_events(self, start_time_epoch, end_time_epoch, params: Dict = dict({})):
+        try:
+            response = api.Event.query(start=start_time_epoch, end=end_time_epoch, **params)
+            if not response:
+                logger.error(f"No events found")
+            events = response['events']
+            print("Found total events: ", len(events))
+            return events
+        except Exception as e:
+            logger.error(f"Exception occurred while fetching monitors with error: {e}")
+            return None
+
+    def fetch_metric(self, start_time_epoch, end_time_epoch, params: Dict = dict({})):
+        # api.Metric.query(start=start_time_epoch, end=end_time_epoch, **params)
+        return None

requirements.txt

@@ -14,6 +14,7 @@ click==8.1.7
 click-didyoumean==0.3.0
 click-plugins==1.1.1
 click-repl==0.3.0
+datadog==0.47.0
 db==0.1.1
 db-sqlite3==0.0.1
 Flask==3.0.0

route_handlers/datadog_route_handler.py

@@ -0,0 +1,26 @@
+from flask import jsonify
+
+from processors.datadog_webclient_processor import DatadogRestApiProcessor
+
+
+def get_monitors(dd_api_key, dd_app_key, dd_monitor_id=None):
+    datadog_webclient_processor = DatadogRestApiProcessor(dd_api_key, dd_app_key)
+    monitors = datadog_webclient_processor.fetch_monitors(dd_monitor_id)
+    if monitors:
+        return jsonify(monitors)
+    return jsonify({'success': False})
+
+
+def get_alert_events(dd_api_key, dd_app_key, start_time_epoch, end_time_epoch, params=None):
+    if not params:
+        params = {}
+    params['sources'] = 'alert'
+    datadog_webclient_processor = DatadogRestApiProcessor(dd_api_key, dd_app_key)
+    events = datadog_webclient_processor.fetch_events(start_time_epoch, end_time_epoch, params)
+    if events:
+        return jsonify(events)
+    return jsonify({'success': False})
+
+
+def get_metric_timeseries(dd_api_key, dd_app_key, start_time_epoch, end_time_epoch, query_string):
+    return