perf: store logs as bytes in buffer

Author: ErikKalkoken

src/vlogs_handler/__init__.py

@@ -6,6 +6,5 @@
 __version__ = "0.1.0dev7"
 
 # TODO
-# - Consider storing logs as strings or bytes to reduce memory footprint
-# - Consider using a persistent queue
 # - Enable logging and switch to full debug logging in prod for testing
+# - Consider using a persistent queue

src/vlogs_handler/handler.py

@@ -18,13 +18,44 @@
 import threading
 import traceback
 import urllib.parse
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Callable, List, Optional, Tuple
+
+import orjson
 
 from . import request
 
 logger = logging.getLogger(__name__)
 
 
+_STANDARD_ATTRS: frozenset[str] = frozenset(
+    {
+        "args",
+        "asctime",
+        "created",
+        "exc_info",
+        "exc_text",
+        "filename",
+        "funcName",
+        "levelname",
+        "levelno",
+        "lineno",
+        "message",
+        "module",
+        "msecs",
+        "msg",
+        "name",
+        "pathname",
+        "process",
+        "processName",
+        "relativeCreated",
+        "stack_info",
+        "taskName",
+        "thread",
+        "threadName",
+    }
+)
+
+
 class VictoriaLogsHandler(logging.Handler):
     """VictoriaLogsHandler dispatches log events to a VictoriaLogs server.
 
@@ -34,7 +65,7 @@ class VictoriaLogsHandler(logging.Handler):
         buffer_size: Maximum number of logs the buffer can hold.
             If buffer_size <= 0, the size is unlimited (not recommended).
             When the buffer is full any new logs will be discarded.
-            100 000 logs approximately consume 100 MB of RAM in the buffer.
+            100 000 logs consume approx. 80-100 MB of RAM.
         chunk_size: Maximum number of logs send per request to the log server.
         record_to_stream: function that returns the stream value for a record.
             The default will return the name of the top package.
@@ -45,34 +76,6 @@ class VictoriaLogsHandler(logging.Handler):
         url: URL of the vlogs server, e.g. `"http://localhost:9428"`
     """
 
-    _STANDARD_ATTRS: frozenset[str] = frozenset(
-        {
-            "args",
-            "asctime",
-            "created",
-            "exc_info",
-            "exc_text",
-            "filename",
-            "funcName",
-            "levelname",
-            "levelno",
-            "lineno",
-            "message",
-            "module",
-            "msecs",
-            "msg",
-            "name",
-            "pathname",
-            "process",
-            "processName",
-            "relativeCreated",
-            "stack_info",
-            "taskName",
-            "thread",
-            "threadName",
-        }
-    )
-
     def __init__(
         self,
         batch_size: int = 125,
@@ -160,47 +163,31 @@ def close(self):
 
     def emit(self, record: logging.LogRecord) -> None:
         try:
-            log_entry = self._format_log_record(record)
-            self._buffer.put_nowait(log_entry)
+            log = _serialize_log_to_json(record, self._record_to_stream)
+
+        except Exception:
+            logger.exception("serializing log: %s", record)
+            self.handleError(record)
+            return
+
+        try:
+            self._buffer.put_nowait(log)
 
         except queue.Full:
-            logger.error("Buffer full. Discarding new log.")
+            logger.error("Buffer full. Discarding new log: %s", log)
             return
 
         except Exception:
-            logger.exception("Emitting record")
+            logger.exception("Emitting record: %s", log)
             self.handleError(record)
             return
 
         with self._lock:
             self._added_count += 1
             if self._added_count > self._batch_size:
+                self._added_count = 0
                 self._worker_run.set()
 
-    def _format_log_record(self, record: logging.LogRecord) -> Dict[str, Any]:
-        log = {
-            "stream": self._record_to_stream(record),
-            "timestamp": record.created,
-            "level": record.levelname,
-            "logger": record.name,
-            "module": record.module,
-            "function": record.funcName,
-            "line_number": record.lineno,
-            "message": record.getMessage(),
-            "process_name": record.processName,
-            "process": record.process,
-            "thread_name": record.threadName,
-            "thread": record.thread,
-        }
-        if record.exc_info:
-            log["exception_name"], log["exception"] = _format_exception(record.exc_info)
-
-        for k, v in record.__dict__.items():
-            if k not in self._STANDARD_ATTRS:
-                log[k] = v
-
-        return log
-
     def start(self):
         """Start the worker. Do nothing when the worker is already running."""
         with self._lock:
@@ -215,19 +202,17 @@ def start(self):
     def _worker(self):
         while not self._worker_shutdown.is_set():
             self._worker_run.wait(timeout=self._flush_interval)
-            with self._lock:
-                self._added_count = 0
-            self.flush()
             self._worker_run.clear()
+            self.flush()
 
         logger.debug("Worker stopped")
 
     def flush(self):
         """Flush the buffer and send all logs to the log server."""
-        failed: List[Dict[str, Any]] = []
+        failed: List[bytes] = []
         done = False
         while not done:
-            logs: List[Dict[str, Any]] = []
+            logs: List[bytes] = []
             while len(logs) < self._chunk_size:
                 try:
                     logs.append(self._buffer.get_nowait())
@@ -237,13 +222,13 @@ def flush(self):
 
             if logs:
                 ok = request.post_ndjson(
-                    url=self._vlogs_url, data=logs, timeout=self._request_timeout
+                    url=self._vlogs_url, objs=logs, timeout=self._request_timeout
                 )
                 if not ok:
                     failed += logs
-                    logger.warning("Logs failed to send to log server: %d", len(logs))
+                    logger.warning("Failed to send logs to log server: %d", len(logs))
                 else:
-                    logger.debug("Logs submitted to log server: %d", len(logs))
+                    logger.debug("Logs send to log server: %d", len(logs))
 
         if not failed:
             return
@@ -263,6 +248,35 @@ def flush(self):
         logger.debug("Saved logs to buffer after failed send: %d", n)
 
 
+def _serialize_log_to_json(
+    record: logging.LogRecord, record_to_stream: Callable[[logging.LogRecord], str]
+) -> bytes:
+    """Serialize a log record into a JSON object and return it."""
+    obj = {
+        "stream": record_to_stream(record),
+        "timestamp": record.created,
+        "level": record.levelname,
+        "logger": record.name,
+        "module": record.module,
+        "function": record.funcName,
+        "line_number": record.lineno,
+        "message": record.getMessage(),
+        "process_name": record.processName,
+        "process": record.process,
+        "thread_name": record.threadName,
+        "thread": record.thread,
+    }
+    if record.exc_info:
+        obj["exception_name"], obj["exception"] = _format_exception(record.exc_info)
+
+    for k, v in record.__dict__.items():
+        if k not in _STANDARD_ATTRS:
+            obj[k] = v
+
+    log = orjson.dumps(obj, default=str)
+    return log
+
+
 def _create_filter(name: str):
     def filter_logic(record: logging.LogRecord) -> bool:
         return not record.name.startswith(name)

src/vlogs_handler/request.py

@@ -4,9 +4,7 @@
 import urllib.error
 import urllib.parse
 import urllib.request
-from typing import Any, List, Optional
-
-import orjson  # significantly better performance than standard json library
+from typing import List, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -20,30 +18,21 @@ def is_url(url: str) -> bool:
         return False
 
 
-def post_ndjson(*, url: str, data: List[Any], timeout: Optional[float] = None) -> bool:
+def post_ndjson(
+    *, url: str, objs: List[bytes], timeout: Optional[float] = None
+) -> bool:
     """Send a POST request with the ndjson protocol
     and report whether it was successful.
 
     Args:
         url: request URL
-        data: list of objects to send
+        data: list of JSON objects to send
         timeout: request timeout in seconds.
             Settings it to None will disable the timeout.
     """
 
-    chunks: List[bytes] = []
-    for obj in data:
-        try:
-            chunks.append(orjson.dumps(obj, option=orjson.OPT_APPEND_NEWLINE))
-        except (TypeError, ValueError):
-            logger.exception(
-                "Could not convert obj to JSON. Discarded", extra={"log": obj}
-            )
-            continue
-
-    data_bytes = b"".join(chunks)
-
-    req = urllib.request.Request(url, data=data_bytes, method="POST")
+    data = b"\n".join(objs)
+    req = urllib.request.Request(url, data=data, method="POST")
     req.add_header("Content-Type", "application/x-ndjson")
 
     try:

tests/test_encoder.py

@@ -31,18 +31,11 @@ def test_should_encode(self):
         got = orjson.loads(orjson.dumps(data, default=str))
 
         # then
-        self.assertEqual(got["class"], "<class 'tests.test_encoder.MyClass'>")
+        self.assertIn("MyClass", got["class"])
         self.assertEqual(got["date"], "2026-01-11")
         self.assertEqual(got["datetime"], "2026-01-11T12:15:42.000099+00:00")
         self.assertEqual(got["float"], 1.23)
         self.assertIn("my_func at", got["func"])
         self.assertEqual(got["integer"], 1)
         self.assertEqual(got["set"], "{1, 2, 3}")
         self.assertEqual(got["text"], "Alpha")
-        self.assertEqual(got["text"], "Alpha")
-        self.assertEqual(got["text"], "Alpha")
-        self.assertEqual(got["text"], "Alpha")
-        self.assertEqual(got["text"], "Alpha")
-        self.assertEqual(got["text"], "Alpha")
-        self.assertEqual(got["text"], "Alpha")
-        self.assertEqual(got["text"], "Alpha")

tests/test_handler.py

@@ -5,6 +5,8 @@
 import unittest
 from unittest.mock import MagicMock, patch
 
+import orjson
+
 from vlogs_handler import VictoriaLogsHandler, handler
 
 MODULE_PATH = "vlogs_handler.handler"
@@ -93,7 +95,7 @@ def test_should_send_a_log(self, m: MagicMock):
         self.assertTrue(called_in_time)
 
         self.assertEqual(m.call_count, 1)
-        got = m.call_args.kwargs["data"][0]
+        got = orjson.loads(m.call_args.kwargs["objs"][0])
         self.assertEqual(got["stream"], "test_logger")
         self.assertEqual(got["level"], "INFO")
         self.assertEqual(got["logger"], "test_logger.module_1.module_2")
@@ -113,10 +115,10 @@ def test_should_send_log_with_extras(self, m: MagicMock):
         self.assertTrue(called_in_time)
 
         self.assertEqual(m.call_count, 1)
-        got = m.call_args.kwargs["data"][0]
+        got = orjson.loads(m.call_args.kwargs["objs"][0])
         self.assertEqual(got["message"], "Alpha")
         self.assertEqual(got["planet"], "Jupiter")
-        self.assertEqual(got["deadline"], my_date)
+        self.assertEqual(got["deadline"], "2026-01-11T12:15:42.000099+00:00")
 
     def test_should_send_exception_log(self, m: MagicMock):
         # given
@@ -133,7 +135,7 @@ def test_should_send_exception_log(self, m: MagicMock):
         self.assertTrue(called_in_time)
 
         self.assertEqual(m.call_count, 1)
-        got = m.call_args.kwargs["data"][0]
+        got = orjson.loads(m.call_args.kwargs["objs"][0])
         self.assertEqual(got["stream"], "test_logger")
         self.assertEqual(got["level"], "ERROR")
         self.assertEqual(got["logger"], "test_logger.module_1.module_2")
@@ -185,11 +187,11 @@ def test_handler_should_send_multiple_logs_in_single_request(self, m: MagicMock)
         self.assertTrue(called_in_time)
 
         self.assertEqual(m.call_count, 1)
-        got = m.call_args.kwargs["data"]
+        got = m.call_args.kwargs["objs"]
         self.assertEqual(len(got), 2)
 
-        self.assertEqual(got[0]["message"], "Alpha")
-        self.assertEqual(got[1]["message"], "Bravo")
+        self.assertEqual(orjson.loads(got[0])["message"], "Alpha")
+        self.assertEqual(orjson.loads(got[1])["message"], "Bravo")
 
 
 @patch(MODULE_PATH + ".request.post_ndjson")
@@ -237,7 +239,7 @@ def test_handler_should_send_immediately_when_batch_size_reached(
 
         self.assertEqual(m.call_count, 1)
 
-        self.assertEqual(len(m.call_args.kwargs["data"]), 4)
+        self.assertEqual(len(m.call_args.kwargs["objs"]), 4)
 
     def test_handler_should_shutdown_gracefully(self, m: MagicMock):
         # given
@@ -251,9 +253,9 @@ def test_handler_should_shutdown_gracefully(self, m: MagicMock):
 
         # then
         self.assertEqual(m.call_count, 1)
-        got = m.call_args.kwargs["data"]
+        got = m.call_args.kwargs["objs"]
         self.assertEqual(len(got), 1)
-        self.assertEqual(got[0]["message"], "Alpha")
+        self.assertEqual(orjson.loads(got[0])["message"], "Alpha")
 
 
 @patch(MODULE_PATH + ".request.post_ndjson")
@@ -299,8 +301,8 @@ def test_should_send_logs_in_chunks(self, m: MagicMock):
         # then
         self.assertEqual(handler._buffer.qsize(), 0)
         self.assertEqual(m.call_count, 2)
-        self.assertEqual(len(m.call_args_list[0].kwargs["data"]), 3)
-        self.assertEqual(len(m.call_args_list[1].kwargs["data"]), 1)
+        self.assertEqual(len(m.call_args_list[0].kwargs["objs"]), 3)
+        self.assertEqual(len(m.call_args_list[1].kwargs["objs"]), 1)
 
 
 def add_to_queue(queue: queue.Queue, items):
@@ -338,5 +340,3 @@ def test_empty_string_name(self):
         self.mock_record.name = ""
         result = handler._top_package_name(self.mock_record)
         self.assertEqual(result, "")
-        self.assertEqual(result, "")
-        self.assertEqual(result, "")