perf: better json performance

Author: ErikKalkoken

README.md

@@ -26,8 +26,6 @@ STATUS: In development
 
 - **Standard Configuration Support**: Fully compatible with `logging.config.dictConfig`, making it easy to drop into frameworks like Django, Flask, or FastAPI.
 
-- **Zero dependency**: Does not use 3rd party libraries to avoid potential conflicts when processing log events from any 3rd party library.
-
 ## Installation
 
 The handler can be installed with PIP from PyPI:

pyproject.toml

@@ -16,6 +16,7 @@ classifiers = [
     "Programming Language :: Python :: 3.13",
     "Topic :: System :: Logging",
 ]
+dependencies = ["orjson>3.10"]
 dynamic = ["version", "description"]
 license = "MIT"
 license-files = ["LICENSE"]

src/vlogs_handler/__init__.py

@@ -3,9 +3,9 @@
 from .handler import VictoriaLogsHandler  # noqa: F401
 
 __title__ = "VictoriaLogs Handler"
-__version__ = "0.1.0dev6"
+__version__ = "0.1.0dev7"
 
 # TODO
-# [ ] - Split very large requests
-# [ ] - Consider using a persistent queue
-# [ ] - Enable logging and switch to full debug logging in prod for testing
+# - 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

src/vlogs_handler/encoder.py

@@ -1,11 +1,12 @@
 """Module handler provides the implementation of the vlogs handler.
 
 - The handler collects log events and sends them asynchronously to the vlogs server
-- Log records are converted into JSON objects. The handler uses an extension
-of Python's default json encoder to serialize additional types and improve robustness.
+- Log records are converted into JSON objects.
+    The handler uses orjson for significantly better performance
+    and support of additional data types
 - The handler uses the vlogs's JSON Stream API for data ingestion.
 - Logs are submitted when the flush interval expires or when the batch size is exceeded
-- Multiple logs are batched together into a single request using the ndjson protocol
+- Multiple logs are chunked together into a single request using the ndjson protocol
     to minimize the number of requests.
 - When submission to the log server fails, logs are returns into the buffer
     for later retry.

src/vlogs_handler/handler.py

@@ -1,14 +1,12 @@
 """Module request provides the functionality to send HTTP requests."""
 
-import io
-import json
 import logging
 import urllib.error
 import urllib.parse
 import urllib.request
 from typing import Any, List, Optional
 
-from . import encoder
+import orjson  # significantly better performance than standard json library
 
 logger = logging.getLogger(__name__)
 
@@ -33,16 +31,17 @@ def post_ndjson(*, url: str, data: List[Any], timeout: Optional[float] = None) -
             Settings it to None will disable the timeout.
     """
 
-    with io.StringIO() as buffer:
-        for obj in data:
-            try:
-                json.dump(obj, buffer, cls=encoder.JSON)
-                buffer.write("\n")
-            except (TypeError, ValueError, RecursionError):
-                logger.exception("convert obj to JSON. Discarded", extra={"entry": obj})
-                continue
-
-        data_bytes = buffer.getvalue().encode("utf-8")
+    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")
     req.add_header("Content-Type", "application/x-ndjson")

src/vlogs_handler/request.py

@@ -1,19 +1,23 @@
 import datetime as dt
-import json
 import unittest
 
-from vlogs_handler import encoder
+import orjson
 
 
-class TestJSONEncoderPlus(unittest.TestCase):
+class MyClass:
+    pass
+
+
+def my_func():
+    pass
+
+
+class TestOrjsonEncoder(unittest.TestCase):
     def test_should_encode(self):
         # given
-        def my_func():
-            pass
-
         my_date = dt.datetime(2026, 1, 11, 12, 15, 42, 99, tzinfo=dt.timezone.utc)
         data = {
-            "class": encoder.JSON,
+            "class": MyClass,
             "date": my_date.date(),
             "datetime": my_date,
             "float": 1.23,
@@ -24,15 +28,21 @@ def my_func():
         }
 
         # when
-        got = json.loads(json.dumps(data, cls=encoder.JSON))
+        got = orjson.loads(orjson.dumps(data, default=str))
 
         # then
-        self.assertEqual(got["class"], "<class 'vlogs_handler.encoder.JSON'>")
+        self.assertEqual(got["class"], "<class 'tests.test_encoder.MyClass'>")
         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["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_encoder.py

@@ -4,7 +4,7 @@
 from contextlib import contextmanager
 from http.client import HTTPMessage
 from io import BytesIO
-from typing import NamedTuple
+from typing import List, NamedTuple
 from unittest.mock import MagicMock, patch
 
 from vlogs_handler import request
@@ -13,6 +13,7 @@
 
 
 def make_http_error(status=404, message="Not Found", url="https://example.com"):
+    """Create a HTTPError."""
     headers = HTTPMessage()
     fp = BytesIO(b"Error body content")
     return urllib.error.HTTPError(
@@ -21,15 +22,24 @@ def make_http_error(status=404, message="Not Found", url="https://example.com"):
 
 
 def make_urlopen_fake(exception=None):
+    """Create and return a fake for urlopen and the list of recorded requests.
+
+    The requests list is updated which every new request.
+    """
+    requests_history: List[urllib.request.Request] = []
+
     @contextmanager
     def urlopen_fake(req: urllib.request.Request, timeout):
         if not isinstance(req, urllib.request.Request):
             raise ValueError("req must be a Request")
+
+        requests_history.append(req)
+
         if exception:
             raise exception
         yield MagicMock()
 
-    return urlopen_fake
+    return urlopen_fake, requests_history
 
 
 class TestPostNdjson(unittest.TestCase):
@@ -40,40 +50,50 @@ def setUp(self) -> None:
     def test_should_submit_successfully(self):
         # when
         with patch(MODULE_PATH + ".urllib.request.urlopen") as m:
-            m.side_effect = make_urlopen_fake()
+            m.side_effect, requests_history = make_urlopen_fake()
             got = request.post_ndjson(url=self.url, data=self.data)
 
         # then
         self.assertTrue(got)
+        self.assertEqual(len(requests_history), 1)
+        req = requests_history[0]
+        self.assertEqual(req.full_url, self.url)
+        self.assertEqual(req.get_header("Content-type"), "application/x-ndjson")
+        self.assertEqual(req.data, b'{"event":"test"}\n{"event":"more_data"}\n')
 
     def test_should_handle_http_exception(self):
         # when
         with patch(MODULE_PATH + ".urllib.request.urlopen") as m:
-            m.side_effect = make_urlopen_fake(exception=make_http_error())
+            m.side_effect, requests_history = make_urlopen_fake(
+                exception=make_http_error()
+            )
             got = request.post_ndjson(url=self.url, data=self.data)
 
         # then
         self.assertFalse(got)
+        self.assertEqual(len(requests_history), 1)
 
     def test_should_handle_url_exception(self):
         # when
         with patch(MODULE_PATH + ".urllib.request.urlopen") as m:
-            m.side_effect = make_urlopen_fake(
+            m.side_effect, requests_history = make_urlopen_fake(
                 exception=urllib.error.URLError("Network is unreachable")
             )
             got = request.post_ndjson(url=self.url, data=self.data)
 
         # then
         self.assertFalse(got)
+        self.assertEqual(len(requests_history), 1)
 
     def test_should_handle_general_exception(self):
         # when
         with patch(MODULE_PATH + ".urllib.request.urlopen") as m:
-            m.side_effect = make_urlopen_fake(exception=RuntimeError)
+            m.side_effect, requests_history = make_urlopen_fake(exception=RuntimeError)
             got = request.post_ndjson(url=self.url, data=self.data)
 
         # then
         self.assertFalse(got)
+        self.assertEqual(len(requests_history), 1)
 
 
 class TestIsURL(unittest.TestCase):