event-agent: enhance access log

etc/event-agent.conf-sample

@@ -49,6 +49,7 @@ handlers_conf = /etc/oio/sds/OPENIO/event-agent/event-handlers.conf
 log_facility = LOG_LOCAL0
 log_level = INFO
 log_address = /dev/log
+log_request_format=log_type:access	request_id:%(request_id)s	status_int:%(status)d	duration_float:%(duration)f	handler:%(handler)s	event:%(event_type)s	tube:%(tube)s	topic:%(topic)s	account:%(account)s	container:%(container)s	object:%(path)s	content_id:%(content)s	version_id:%(version)s
 syslog_prefix = OIO,OPENIO,event-agent,1
 
 #log all events that are handled (access logs)

oio/common/logger.py

@@ -314,7 +314,7 @@ def format(self, record, extras=None):
             "exc_filename": exc_filename,
             "exc_lineno": exc_lineno,
             "message": record.getMessage(),
-            **{k: v for k, v in self.get_extras()},
+            **self.get_extras(),
             **extras,
         }
 

oio/event/filters/base.py

@@ -14,25 +14,17 @@
 # You should have received a copy of the GNU Lesser General Public
 # License along with this library.
 from contextvars import ContextVar
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 from logging import makeLogRecord
 
 from oio.common.exceptions import OioException
 from oio.common.logger import LTSVFormatter, get_logger
+from oio.event.utils import MsgContext, log_context_from_msg
 
 
 @dataclass(init=True)
-class FilterContext:
-    request_id: str = None
-    event_type: str = None
-    user: str = None
-    container: str = None
-    bucket: str = None
-    path: str = None
-    version: str = None
-
-    def items(self):
-        return self.__dict__.items()
+class FilterContext(MsgContext):
+    filter_name: str = None
 
 
 ctx_filter = ContextVar("filter", default=FilterContext())
@@ -41,18 +33,21 @@ def items(self):
 class FilterLTSVFormater(LTSVFormatter):
     def get_extras(self):
         ctx = ctx_filter.get()
-        return ctx.items()
+        return asdict(ctx)
 
 
 class Filter(object):
     DEFAULT_LOG_FORMAT = "\t".join(
         (
             "pid:%(pid)d",
             "log_level:%(levelname)s",
+            "filter:%(filter_name)s",
             "event_type:%(event_type)s",
             "request_id:%(request_id)s",
+            "account:%(account)s",
             "container:%(container)s",
             "object:%(path)s",
+            "content_id:%(content)s",
             "version_id:%(version)s",
             "exc_text:%(exc_text)s",
             "exc_filename:%(exc_filename)s",
@@ -76,7 +71,7 @@ def __init__(self, app, conf, logger=None):
         formatter = FilterLTSVFormater(fmt=log_format)
         # Ensure log format can be populated
         record = makeLogRecord({})
-        formatter.format(record, extras=self.log_context_from_env({}).__dict__)
+        formatter.format(record, extras=asdict(self.log_context_from_env({})))
 
         self.logger = get_logger(
             conf,
@@ -89,16 +84,9 @@ def __init__(self, app, conf, logger=None):
     def init(self):
         pass
 
-    def log_context_from_env(self, env):
-        ctx = FilterContext()
-        ctx.request_id = env.get("request_id")
-        ctx.event_type = env.get("event")
-        url = env.get("url")
-        if url:
-            ctx.path = url.get("path")
-            ctx.container = url.get("user")
-            ctx.account = url.get("account")
-            ctx.version = url.get("version")
+    def log_context_from_env(self, env, context_class=FilterContext):
+        ctx = log_context_from_msg(env, context_class)
+        ctx.filter_name = self.__class__.__name__
         return ctx
 
     def process(self, env, cb):

oio/event/filters/lifecycle_actions.py

@@ -13,6 +13,7 @@
 # You should have received a copy of the GNU Lesser General Public
 # License along with this library.
 
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from enum import Enum
 from urllib.parse import quote
@@ -139,6 +140,7 @@ def rule_id(self):
         return self.event.data.get("rule_id")
 
 
+@dataclass(init=True)
 class LifecycleFilterContext(FilterContext):
     action: str = None
     rule_id: str = None
@@ -341,7 +343,7 @@ def _log_event(self, context: LifecycleActionContext):
         self._metrics = LifecycleMetricTracker(self.conf)
 
     def log_context_from_env(self, env):
-        ctx = super().log_context_from_env(env)
+        ctx = super().log_context_from_env(env, LifecycleFilterContext)
         data = env.get("data", {})
         ctx.action = data.get("action")
         ctx.rule_id = data.get("rule_id")

oio/event/kafka_agent.py

@@ -13,6 +13,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import time
+from dataclasses import asdict
 
 from oio.account.bucket_client import BucketClient
 from oio.account.client import AccountClient
@@ -25,10 +26,12 @@
 from oio.event.evob import is_retryable, is_success
 from oio.event.kafka_consumer import KafkaConsumerWorker, RejectMessage, RetryLater
 from oio.event.loader import loadhandlers
+from oio.event.utils import log_context_from_msg
 from oio.rdir.client import RdirClient
 
 
 class KafkaEventWorker(KafkaConsumerWorker):
+
     def __init__(
         self,
         topic,
@@ -112,17 +115,22 @@ def log_and_statsd(self, start, status, _extra):
             "request_id": "-",
             "tube": "-",
             "topic": "-",
-            "event": "-",
-            **_extra,
+            "event_type": "-",
+            "container": "-",
+            "account": "-",
+            "path": "-",
+            "content": "-",
+            "version": "-",
         }
+        extra.update({k: v for k, v in _extra.items() if v is not None})
 
         extra["duration"] = time.monotonic() - start
         extra["status"] = status
-        extra["event"] = str(extra["event"]).replace(".", "-")
+        extra["event_type"] = str(extra["event_type"]).replace(".", "-")
         if self.logger_request is not None:
             self.logger_request.info("", extra=extra)
         self.statsd.timing(
-            f"openio.event.{extra['topic']}.{extra['event']}.{extra['status']}"
+            f"openio.event.{extra['topic']}.{extra['event_type']}.{extra['status']}"
             ".duration",
             extra["duration"] * 1000,
         )
@@ -134,13 +142,14 @@ def process_message(self, message, _properties):
         start = time.monotonic()
         reqid = message.get("request_id")
         event = message.get("event")
+        ctx = log_context_from_msg(message)
         replacements = {
-            "request_id": reqid,
             "tube": self.topic,
             "topic": self.topic,
-            "event": event,
+            **asdict(ctx),
         }
         handler = self.handlers.get(event, None)
+        replacements["handler"] = handler.__class__.__name__ if handler else None
         if not handler:
             self.log_and_statsd(start, 404, replacements)
             raise RejectMessage(f"No handler for {message.get('event')}")

oio/event/utils.py

@@ -0,0 +1,45 @@
+# Copyright (C) 2025 OVH SAS
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 3.0 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library.
+from dataclasses import dataclass
+
+
+@dataclass(init=True)
+class MsgContext:
+    request_id: str = None
+    event_type: str = None
+    content: str = None
+    container: str = None
+    account: str = None
+    path: str = None
+    version: str = None
+
+    def items(self):
+        return self.__dict__.items()
+
+
+def log_context_from_msg(message, context_class=MsgContext):
+    ctx = context_class()
+    ctx.request_id = message.get("request_id")
+    ctx.event_type = message.get("event")
+    url = message.get("url")
+    match_ctx_name = {"user": "container"}
+    if url:
+        for key in ("path", "content", "version", "user", "account"):
+            if key in url:
+                value = url.get(key)
+                if key in match_ctx_name:
+                    key = match_ctx_name[key]
+                setattr(ctx, key, value)
+    return ctx