add support for fluent logging

be/levity/settings.py

@@ -139,3 +139,5 @@
 )
 
 OCPP_HEARTBEAT_INTERVAL = 3600
+
+FLUENTD_HOST = os.environ.get("FLUENTD_HOST")

be/ocpp/models/message.py

@@ -61,7 +61,9 @@ def to_ocpp(self):
         else:
             raise ValueError("Unknown message type {}".format(self.message_type))
 
-        return dict(id=self.charge_point.id, message=ocpp_message)
+        return dict(
+            id=self.charge_point.id, actor=str(self.actor), message=ocpp_message
+        )
 
     @staticmethod
     def from_occp(charge_point: ChargePoint, ocpp_message: dict):

be/ocpp/receivers/__init__.py

@@ -1 +1,2 @@
 from .metrics.event_counter import count_websocket_events  # noqa: F401
+from .logging.fluent_logger import log_messages, log_websocket_events  # noqa: F401

be/ocpp/receivers/logging/fluent_logger.py

@@ -0,0 +1,27 @@
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+from fluent import sender
+
+from levity import settings
+from ocpp.models import WebsocketEvent, Message
+
+
+logger = (
+    sender.FluentSender("app", host=settings.FLUENTD_HOST, port=24224)
+    if settings.FLUENTD_HOST
+    else None
+)
+
+
+@receiver(post_save, sender=WebsocketEvent)
+def log_websocket_events(instance: WebsocketEvent, created, **kwargs):
+    if not logger:
+        return
+    logger.emit(f"ws.{instance.type}", dict(id=instance.charge_point.id))
+
+
+@receiver(post_save, sender=Message)
+def log_messages(instance: Message, created, **kwargs):
+    if not logger:
+        return
+    logger.emit("message", instance.to_ocpp())

be/pyproject.toml

@@ -16,6 +16,7 @@ psycopg2-binary = "^2.9.5"
 prometheus-client = "^0.16.0"
 pyyaml = "6.0.1"
 pytz = "^2023.3"
+fluent-logger = "^0.10.0"
 
 
 [tool.poetry.group.dev.dependencies]

fluentd/Dockerfile

@@ -0,0 +1,20 @@
+FROM fluent/fluentd:v1.16-1
+
+# Use root account to use apk
+USER root
+
+# below RUN includes plugin as examples elasticsearch is not required
+# you may customize including plugins as you wish
+RUN apk add --no-cache --update --virtual .build-deps \
+        sudo build-base ruby-dev \
+ && sudo gem install fluent-plugin-s3 \
+ && sudo gem sources --clear-all \
+ && apk del .build-deps \
+ && rm -rf /tmp/* /var/tmp/* /usr/lib/ruby/gems/*/cache/*.gem
+
+COPY fluent.conf /fluentd/etc/
+
+RUN mkdir -p /var/log/fluent
+RUN chown -R fluent /var/log/fluent
+
+USER fluent

fluentd/fluent.conf

@@ -0,0 +1,20 @@
+<source>
+  @type forward
+</source>
+
+<match *>
+  @type s3
+
+  s3_bucket levity
+  s3_region us-west-2
+  path logs/
+  <instance_profile_credentials>
+  </instance_profile_credentials>
+  <buffer tag,time>
+    @type file
+    path /var/log/fluent/s3
+    timekey 3600 # 1 hour partition
+    timekey_wait 10m
+    timekey_use_utc true # use utc
+  </buffer>
+</match>