Process logs before sending them to loki

services/logging/loki.yaml

@@ -43,6 +43,7 @@ compactor:
   retention_enabled: false
 
 limits_config:
+  discover_log_levels: false
   reject_old_samples: true
   reject_old_samples_max_age: 4h
   max_cache_freshness_per_query: 10m

services/logging/vector.yaml

@@ -25,11 +25,39 @@ transforms:
         .message = .short_message
       }
 
+      # Extract structured log fields from message using regex pattern
+      if exists(.message) {
+        # python service logs
+        parsed_fields, err = parse_regex(.message, r'log_level=(?P<log_level>[^|]*) \| log_timestamp=(?P<log_timestamp>[^|]*) \| log_source=(?P<log_source>[^|]*) \| log_uid=(?P<log_uid>[^|]*) \| log_oec=(?P<log_oec>[^|]*) \| log_trace_id=(?P<log_trace_id>[^|]*) \| log_span_id=(?P<log_span_id>[^|]*) \| log_msg=(?P<log_msg>.*)$')
+        if err == null {
+          .log_level = parsed_fields.log_level
+          .log_timestamp = parsed_fields.log_timestamp
+          .log_source = parsed_fields.log_source
+          .log_uid = parsed_fields.log_uid
+          .log_oec = parsed_fields.log_oec
+          .log_trace_id = parsed_fields.log_trace_id
+          .log_span_id = parsed_fields.log_span_id
+          .log_msg = parsed_fields.log_msg
+        }
+        # traefik logs
+        traefik_fields, traefik_err = parse_regex(.message, r'time="(?P<log_timestamp>[^"]*)" level=(?P<log_level>\w+) msg="(?P<log_msg>.*)"')
+        if traefik_err == null {
+          .log_timestamp = traefik_fields.log_timestamp
+          .log_level = traefik_fields.log_level
+          .log_msg = traefik_fields.log_msg
+        }
+      }
+
       # Handle container name - GELF uses _container_name (with underscore prefix)
       if exists(._container_name) {
         .container_name = ._container_name
+
+        # Extract Docker service name from container name (everything before first dot)
+        match = parse_regex!(.container_name, r'^(?P<service_name>[^.]+)')
+        .service_name = match.service_name
       } else {
         .container_name = "unknown"
+        .service_name = "unknown"
       }
 
       # Handle container ID
@@ -58,13 +86,13 @@ sinks:
     encoding:
       codec: json
     labels:
-      job: "docker"
       source: "vector"
-      # Pass through the GELF 'host' field from the original log event
       host: "{{ host }}"
       container_name: "{{ container_name }}"
-    # Remove label fields from the log line to avoid duplication
-    remove_label_fields: true
+      service_name: "{{ service_name }}"
+      # Override level label with extracted log_level from structured logs. This ensures correct color coding in Loki/Grafana
+      level: "{{ log_level }}"
+
     healthcheck:
       enabled: true