Add logging to InfluxDB

girder_wholetale/__init__.py

@@ -12,7 +12,7 @@
 import jsonschema
 import six
 import validators
-from girder import events
+from girder import auditLogger, events
 from girder.api import access
 from girder.api.describe import Description, autoDescribeRoute, describeRoute
 from girder.api.rest import RestException, boundHandler, loadmodel
@@ -52,9 +52,9 @@
 from .lib.path_mapper import PathMapper
 from .models.instance import Instance as InstanceModel
 from .models.lock import Lock as LockModel
-from .models.version_hierarchy import VersionHierarchyModel
 from .models.session import Session as SessionModel
 from .models.transfer import Transfer as TransferModel
+from .models.version_hierarchy import VersionHierarchyModel
 from .rest.account import Account
 from .rest.dataset import Dataset
 from .rest.dm import DM
@@ -77,6 +77,7 @@
     external_auth_providers_schema,
     repository_to_provider_schema,
 )
+from .utils import add_influx_handler
 
 logger = logging.getLogger(__name__)
 
@@ -231,6 +232,10 @@ def _validateLogo(doc):
         PluginSettings.CONTACT_HREF,
         PluginSettings.BUG_HREF,
         PluginSettings.MOUNTS,
+        PluginSettings.INFLUXDB_URL,
+        PluginSettings.INFLUXDB_TOKEN,
+        PluginSettings.INFLUXDB_ORG,
+        PluginSettings.INFLUXDB_BUCKET,
     }
 )
 def validateHref(doc):
@@ -807,6 +812,7 @@ def load(self, info):
 
         pathMapper = PathMapper()
         from .lib.transfer_manager import SimpleTransferManager
+
         transferManager = SimpleTransferManager(pathMapper)
 
         # a GC that does nothing
@@ -825,7 +831,10 @@ def load(self, info):
                 LRUSortingScheme(),
             ),
         )
-        from .lib.cache_manager import SimpleCacheManager  # Needs models to be registered
+        from .lib.cache_manager import (
+            SimpleCacheManager,
+        )
+
         cacheManager = SimpleCacheManager(
             Setting(), transferManager, fileGC, pathMapper
         )
@@ -964,7 +973,11 @@ def sessionDeleted(event):
         metricsLogger.setLevel(logging.INFO)
         metricsLogger.addHandler(_MetricsHandler())
         VersionHierarchyModel().resetCrashedCriticalSections()
-
+        if Setting().get(PluginSettings.INFLUXDB_BUCKET):
+            add_influx_handler(
+                auditLogger,
+                Setting().get(PluginSettings.INFLUXDB_BUCKET),
+            )
         registerPluginStaticContent(
             plugin="wholetale",
             css=["/style.css"],

girder_wholetale/constants.py

@@ -51,6 +51,10 @@ class PluginSettings:
     GC_RUN_INTERVAL = "dm.gc_run_interval"
     GC_COLLECT_START_FRACTION = "dm.gc_collect_start_fraction"
     GC_COLLECT_END_FRACTION = "dm.gc_collect_end_fraction"
+    INFLUXDB_URL = "wholetale.influxdb_url"
+    INFLUXDB_TOKEN = "wholetale.influxdb_token"
+    INFLUXDB_ORG = "wholetale.influxdb_org"
+    INFLUXDB_BUCKET = "wholetale.influxdb_bucket"
 
 
 SettingDefault.defaults.update(
@@ -165,6 +169,10 @@ class PluginSettings:
         PluginSettings.RUNS_DIRS_ROOT: "/tmp/wt/runs-dirs",
         PluginSettings.VERSIONS_DIRS_ROOT: "/tmp/wt/versions-dirs",
         PluginSettings.DAV_SERVER: False,
+        PluginSettings.INFLUXDB_URL: "http://images.local.xarthisius.xyz:8086",
+        PluginSettings.INFLUXDB_TOKEN: "",
+        PluginSettings.INFLUXDB_ORG: "my_org",
+        PluginSettings.INFLUXDB_BUCKET: "",
     }
 )
 

girder_wholetale/utils.py

@@ -1,18 +1,108 @@
-import os
 import datetime
+import json
+import logging
+import os
 import pathlib
-import six.moves.urllib as urllib
+from urllib.parse import quote_plus
 
-from girder.utility.model_importer import ModelImporter
+try:
+    import influxdb_client as influxdb
+except ImportError:
+    influxdb = None
+
+import cherrypy
 from girder.models.notification import Notification
-from girder.models.user import User
 from girder.models.setting import Setting
+from girder.models.user import User
+from girder.utility.model_importer import ModelImporter
 
+from .constants import PluginSettings
 
 NOTIFICATION_EXP_HOURS = 1
 WT_EVENT_EXP_SECONDS = int(os.environ.get("GIRDER_WT_EVENT_EXP_SECONDS", 5))
 
 
+class InfluxHandler(logging.Handler):
+    def __init__(self, bucket):
+        if influxdb is None:
+            raise ImportError("InfluxDB client not installed")
+
+        super().__init__()
+        self.client = influxdb.InfluxDBClient(
+            url=Setting().get(PluginSettings.INFLUXDB_URL),
+            token=Setting().get(PluginSettings.INFLUXDB_TOKEN),
+            org=Setting().get(PluginSettings.INFLUXDB_ORG),
+        )
+        self.write_api = self.client.write_api()
+        self.bucket = bucket
+
+    @staticmethod
+    def _document_create(record):
+        return (
+            influxdb.Point("document")
+            .tag("level", record.levelname)
+            .field("collection", record.details.get("collection"))
+            .field("id", str(record.details.get("id")))
+            .time(int(record.created * 1e9), influxdb.WritePrecision.NS)
+        )
+
+    @staticmethod
+    def _download(record):
+        return (
+            influxdb.Point("download")
+            .tag("level", record.levelname)
+            .field("fileId", str(record.details.get("fileId")))
+            .field("startBytes", record.details.get("startBytes"))
+            .field("endBytes", record.details.get("endBytes"))
+            .field("extraParameters", record.details.get("extraParameters"))
+            .time(int(record.created * 1e9), influxdb.WritePrecision.NS)
+        )
+
+    @staticmethod
+    def _rest_request(record):
+        return (
+            influxdb.Point("rest_request")
+            .tag("level", record.levelname)
+            .field("method", record.details.get("method"))
+            .field("status", record.details.get("status"))
+            .field("route", "/".join(record.details.get("route", "")))
+            .field("params", json.dumps(record.details.get("params")))
+            .field("message", record.getMessage())
+            .time(int(record.created * 1e9), influxdb.WritePrecision.NS)
+        )
+
+    def emit(self, record):
+        match record.getMessage():
+            case "document.create":
+                point = self._document_create(record)
+            case "file.download":
+                point = self._download(record)
+            case "rest.request":
+                point = self._rest_request(record)
+        self.write_api.write(bucket=self.bucket, record=point)
+
+    def close(self):
+        self.write_api._write_options.write_scheduler.executor.shutdown(wait=False)
+        self.write_api.close()
+        self.client.close()
+        super().close()
+
+
+def add_influx_handler(logger, bucket):
+    def stop_influx_client():
+        for handler in logger.handlers:
+            if isinstance(handler, InfluxHandler):
+                logger.removeHandler(handler)
+                handler.close()
+
+    stop_influx_client()  # Remove any existing handlers
+    logger.addHandler(InfluxHandler(bucket))
+    stop_influx_client.priority = 10
+    cherrypy.engine.subscribe(
+        "stop", stop_influx_client
+    )  # ensure we clean up on restart
+
+
 def get_tale_dir_root(tale: dict, root_path_setting: str) -> pathlib.Path:
     root = Setting().get(root_path_setting)
     return pathlib.Path(root) / str(tale["_id"])[0:2] / str(tale["_id"])
@@ -41,7 +131,7 @@ def esc(value):
     :return: The escaped string
     :rtype: str
     """
-    return urllib.parse.quote_plus(value)
+    return quote_plus(value)
 
 
 def notify_event(users, event, affectedIds):

requirements.txt

@@ -12,3 +12,4 @@ httpio>=0.3.0
 fs
 pathvalidate
 bdbag
+influxdb-client

setup.py

@@ -33,6 +33,7 @@
         "rdflib",
         "celery[redis]",
         "pathvalidate",
+        "influxdb-client",
         "python-magic",
         "requests",
         "validators",