Access log and JSON logging mode for Kibana

README.md

@@ -84,6 +84,7 @@ docker compose up -d
 
 * [Configuation](docs/Config.md) - how to use `config.yaml` to configure the service
 * [Development](docs/Development.md) - notes on developing the service codebase
+* [Structured Logging](docs/StructuredLogging.md) - JSON/ECS log output for Elasticsearch and Kibana
 
 
 # Attributions

config.template.yaml

@@ -3,6 +3,14 @@
 #
 log_level: INFO
 
+#
+# Log output format: 'text' for human-readable logs, or 'json' for one
+# ECS-shaped JSON object per line on stdout, ready for a log shipper to
+# forward to Elasticsearch/Kibana. Access lines carry event.duration
+# (nanoseconds), http.response.status_code and labels.target.
+#
+log_format: text
+
 #
 # Present an HTML web UI for browsing and downloading files.
 # Disable this to instead expose the ListBuckets API at the root.

docs/Config.md

@@ -3,6 +3,7 @@
 The `config.yaml` file configures the service. You can specify the following properties:
 
 * `log_level`: The logging level (ERROR, WARNING, INFO, DEBUG, TRACE)
+* `log_format`: `text` (default) for human-readable logs, or `json` for structured logs (see [Structured Logging](StructuredLogging.md))
 * `ui`: By default, the root shows an HTML UI listing of the buckets, with navigation. This disables the UI and restores the [ListBuckets](https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListBuckets.html) functionality at the root.
 * `virtual_buckets`: If true, then the buckets can be browsed like subdomains of the base URL, like 'https://bucketname.yourdomain.org'. This requires wildcard SSL certificates and additional configuration at the Nginx level, and requires that the `base_url` is set.
 * `base_url`: The base URL for your service. Only needed when using `virtual_buckets`.

docs/StructuredLogging.md

@@ -0,0 +1,48 @@
+# Structured Logging
+
+Setting `log_format: json` (or `X2S3_LOG_FORMAT=json`) makes x2s3 write one JSON object per line to stdout instead of human-readable text. The field names follow the [Elastic Common Schema](https://www.elastic.co/guide/en/ecs/current/index.html) (ECS), so Kibana's built-in visualizations understand them without any mapping work.
+
+The default remains `text`; nothing changes unless you turn this on.
+
+## What gets logged
+
+Every log record carries `@timestamp`, `log.level`, `log.logger`, `message`, `service.name`, `service.version` and `process.pid`. Records emitted while serving a request also carry `trace.id`, which is the same value returned to the client in the `x-amz-request-id` response header — so a client can report an id and you can pull up every line for that request.
+
+One access line is logged per request, tagged `event.dataset: x2s3.access`, with:
+
+| Field | Notes |
+|---|---|
+| `event.duration` | Request duration in **nanoseconds** (the ECS unit). Measured until the last byte of the body is sent, so it reflects real transfer time for large objects. |
+| `http.request.method`, `http.version` | |
+| `http.response.status_code` | |
+| `http.response.body.bytes` | Bytes actually written to the socket. |
+| `url.path`, `url.query`, `url.domain` | Path is logged as the client encoded it. |
+| `client.ip`, `client.port` | Corrected for the reverse proxy when running with `--proxy-headers`. |
+| `user_agent.original` | Separates Neuroglancer, browser and boto traffic. |
+| `labels.target` | Which configured target served the request. The dimension worth slicing latency by. |
+| `error.type`, `error.message`, `error.stack_trace` | On records logged with an exception. |
+
+## Shipping to Elasticsearch
+
+The container logs to stdout, so a shipper reading the Docker log files needs only to parse the JSON. For Filebeat:
+
+```yaml
+filebeat.inputs:
+  - type: container
+    paths:
+      - /var/lib/docker/containers/*/*.log
+    json.keys_under_root: true
+    json.add_error_key: true
+```
+
+Since ECS field names are written as dotted keys, Elasticsearch expands them into the usual nested ECS document on ingest.
+
+## Kibana
+
+`event.duration` is a plain number, so a percentile aggregation over it gives p95/p99 directly. Set the field's format to *Duration* (input: nanoseconds) in the data view to have Kibana render it readably. Break it down by `labels.target` to find a slow storage backend, or by `http.response.status_code` for error rates.
+
+## Known limitation
+
+x2s3 runs with several uvicorn workers sharing one stdout. Log lines longer than the ~4KB atomic-write limit can interleave between workers; `process.pid` is on every line so those are identifiable. If it becomes a problem, give each worker its own file sink.
+
+Uvicorn logs two lines ("Started server process", "Waiting for application startup") before it imports the app, so those stay plain text on every start. Everything after that is JSON.

tests/test_log.py

@@ -0,0 +1,81 @@
+"""Tests for the access log and the JSON (ECS) logging mode."""
+
+import json
+import logging
+
+import pytest
+from fastapi.testclient import TestClient
+
+from x2s3.app import create_app
+from x2s3.log import configure_logging
+from x2s3.settings import Settings, Target
+
+
+@pytest.fixture(autouse=True)
+def restore_logging():
+    """The app configures logging globally on startup; put it back afterwards.
+
+    JSON mode also reroutes stdlib logging into loguru, which would otherwise
+    leak into whatever test runs next.
+    """
+    root = logging.getLogger()
+    handlers, level = root.handlers[:], root.level
+    yield
+    configure_logging("INFO", "text")
+    root.handlers, root.level = handlers, level
+
+
+def _client(log_format):
+    settings = Settings()
+    settings.log_format = log_format
+    settings.targets = [Target(name='local-files', client='file', options={'path': '.'})]
+    return TestClient(create_app(settings))
+
+
+def _access_lines(captured):
+    """The access log lines out of everything the app logged."""
+    lines = []
+    for line in captured.out.splitlines():
+        record = json.loads(line)
+        if record.get("event.dataset") == "x2s3.access":
+            lines.append(record)
+    return lines
+
+
+def test_json_mode_logs_one_ecs_object_per_request(capsys):
+    with _client("json") as client:
+        client.get("/local-files/README.md")
+
+    lines = _access_lines(capsys.readouterr())
+    assert len(lines) == 1, lines
+    record = lines[0]
+
+    assert record["service.name"] == "x2s3"
+    assert record["http.request.method"] == "GET"
+    assert record["http.response.status_code"] == 200
+    assert record["url.path"] == "/local-files/README.md"
+    assert record["labels.target"] == "local-files"
+    assert record["trace.id"]
+    # The whole point: a number Kibana can take percentiles of.
+    assert isinstance(record["event.duration"], int) and record["event.duration"] > 0
+    # Counted off the wire, so it covers streamed bodies.
+    assert record["http.response.body.bytes"] > 0
+
+
+def test_json_mode_reports_error_status(capsys):
+    with _client("json") as client:
+        client.get("/no-such-target/key")
+
+    record = _access_lines(capsys.readouterr())[0]
+    assert record["http.response.status_code"] == 404
+    assert record["log.level"] == "warning"
+
+
+def test_text_mode_stays_human_readable(capsys):
+    """Default mode must remain a plain access line, not JSON."""
+    with _client("text") as client:
+        client.get("/local-files/README.md")
+
+    err = capsys.readouterr().err
+    assert '"GET /local-files/README.md HTTP/1.1" 200' in err
+    assert "target=local-files" in err

tests/test_xss_prevention.py

@@ -173,4 +173,71 @@ def test_xml_control_characters_are_handled(self):
         assert 'test' in key_text
         assert 'tab' in key_text
         assert 'newline' in key_text
-        assert 'carriage' in key_text
+        assert 'carriage' in key_text
+
+class TestMissingValues:
+    """Error responses must survive values the upstream did not supply.
+
+    escape() takes a string, so anything that reaches these templates as None
+    raised AttributeError while the error response was being built, turning a
+    handled upstream error into an unhandled 500 with a traceback. Real trigger:
+    an S3 backend answering with an empty <Resource/>, which botocore parses to
+    None.
+    """
+
+    def test_error_response_with_no_resource(self):
+        response = get_error_response(500, "InternalError", "Something failed", None)
+        assert response.status_code == 500
+        root = parse_xml(response.body.decode())
+        assert root.find('Resource').text is None
+
+    def test_error_response_with_no_message(self):
+        response = get_error_response(500, "InternalError", None, "/bucket/key")
+        root = parse_xml(response.body.decode())
+        assert root.find('Resource').text == "/bucket/key"
+
+    def test_nosuchkey_response_with_no_key(self):
+        """list_objects_v2 forwards its prefix as the key, and prefix is optional."""
+        response = get_nosuchkey_response(None)
+        assert response.status_code == 404
+        assert parse_xml(response.body.decode()).find('Code').text == 'NoSuchKey'
+
+    def test_error_response_with_non_string_detail(self):
+        """HTTPException.detail is not required to be a string."""
+        response = get_error_response(400, "InvalidArgument", {"msg": "bad"}, "/x")
+        root = parse_xml(response.body.decode())
+        assert "bad" in root.find('Message').text
+
+    def test_client_error_with_null_resource_is_handled(self):
+        """The reported failure, end to end through handle_s3_exception."""
+        from botocore.exceptions import ClientError
+        from x2s3.client_aioboto import handle_s3_exception
+
+        error = ClientError({
+            "ResponseMetadata": {"HTTPStatusCode": 500},
+            "Error": {"Code": "InternalError", "Message": "We encountered an "
+                      "internal error, please try again.", "Resource": None},
+        }, "ListObjectsV2")
+
+        response = handle_s3_exception(error, key=None)
+        assert response.status_code == 500
+        assert parse_xml(response.body.decode()).find('Code').text == 'InternalError'
+
+    def test_client_error_falls_back_to_the_key_for_the_resource(self):
+        """A null Resource must name the key, not report an empty one.
+
+        dict.get's default is skipped when the key is present but null, so the
+        fallback that was meant to run here never did.
+        """
+        from botocore.exceptions import ClientError
+        from x2s3.client_aioboto import handle_s3_exception
+
+        error = ClientError({
+            "ResponseMetadata": {"HTTPStatusCode": 403},
+            "Error": {"Code": "AccessDenied", "Message": "Access Denied",
+                      "Resource": None},
+        }, "ListObjectsV2")
+
+        response = handle_s3_exception(error, key="some/prefix/")
+        root = parse_xml(response.body.decode())
+        assert root.find('Resource').text == "some/prefix/"

x2s3/app.py

@@ -19,16 +19,20 @@
 
 from x2s3.utils import *
 from x2s3 import client_registry
+from x2s3.log import AccessLogMiddleware, configure_logging, disable_uvicorn_access_log
 from x2s3.settings import get_settings, Target
 
 # Use uvloop for better async performance
+# Reported from the lifespan rather than here: at import time no sink is
+# configured yet, so in JSON mode this would be the one plain-text line on the
+# stream.
 try:
     import uvloop
     import asyncio
     asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-    logger.info("uvloop event loop policy installed")
+    _UVLOOP_STATUS = ("info", "uvloop event loop policy installed")
 except ImportError:
-    logger.warning("uvloop not available, using default asyncio event loop")
+    _UVLOOP_STATUS = ("warning", "uvloop not available, using default asyncio event loop")
 
 class RequestIdMiddleware:
     """Pure ASGI middleware that attaches an S3-style x-amz-request-id header
@@ -59,7 +63,10 @@ async def send_with_request_id(message):
                 headers.append((b"x-amz-request-id", request_id.encode("latin-1")))
             await send(message)
 
-        await self.app(scope, receive, send_with_request_id)
+        # Bind it as ECS trace.id as well, so every line logged while serving
+        # this request can be grouped with its access log line in Kibana.
+        with logger.contextualize(**{"trace.id": request_id}):
+            await self.app(scope, receive, send_with_request_id)
 
 
 class PrivateNetworkAccessMiddleware:
@@ -123,8 +130,8 @@ async def lifespan(app: FastAPI):
             app.settings = settings
 
         # Configure logging
-        logger.remove()
-        logger.add(sys.stderr, level=app.settings.log_level)
+        configure_logging(app.settings.log_level, app.settings.log_format)
+        logger.log(_UVLOOP_STATUS[0].upper(), _UVLOOP_STATUS[1])
 
         logger.trace("Available protocols:")
         for proto in client_registry.available_protocols():
@@ -193,6 +200,10 @@ async def lifespan(app: FastAPI):
     # Echo Access-Control-Allow-Private-Network on PNA preflights. Added after
     # (i.e. outside) CORSMiddleware so it wraps the preflight response CORS emits.
     app.add_middleware(PrivateNetworkAccessMiddleware)
+    # Added last, so it is the outermost layer and times the whole request,
+    # including responses produced by the CORS and PNA middlewares.
+    app.add_middleware(AccessLogMiddleware)
+    disable_uvicorn_access_log()
     app.mount("/static", StaticFiles(directory="static"), name="static")
     templates = Jinja2Templates(directory="templates")
 
@@ -247,6 +258,9 @@ def get_target(request, path):
                 target_name, target_path = None, ''
 
         logger.trace(f"target_name={target_name}, target_path={target_path}, is_virtual={is_virtual}")
+        # Leave it where AccessLogMiddleware can pick it up once the response
+        # is done; it is the dimension worth slicing latency by.
+        request.scope.setdefault("state", {})["target_name"] = target_name
         return target_name, target_path, is_virtual
 
 

x2s3/client_aioboto.py

@@ -57,7 +57,7 @@ def handle_s3_exception(e, key=None):
             return get_nosuchkey_response(key)
         else:
             message = error.get('Message', 'Unknown error')
-            resource = error.get('Resource', key or 'Unknown')
+            resource = error.get('Resource') or key or 'Unknown'
             return get_error_response(status_code, error_code, message, resource)
     else:
         logger.opt(exception=sys.exc_info()).error("Error communicating with AWS S3")