feat: json log format and W3C trace context propagation

admin/config/config.yml

@@ -27,7 +27,8 @@ sn_ram: 3g # memory for SN container
 target_sn_count: 0 # desired number of SN containers
 target_dn_count: 0 # desire number of DN containers
 log_level: INFO    # log level.  One of ERROR, WARNING, INFO, DEBUG
-log_timestamps: false # emit timestamp with log messages
+log_format: text   # log output format.  One of: text, json (json lines always include a timestamp)
+log_timestamps: true # emit timestamp with log messages (text format)
 log_prefix: null # Prefix text to append to log entries
 max_tcp_connections: 100 # max number of inflight tcp connections
 head_sleep_time: 10 # max sleep time between health checks for head node

hsds/attr_dn.py

@@ -197,7 +197,7 @@ async def GET_Attributes(request):
     log.debug(f"GET attributes obj_id: {obj_id} got json")
     if "attributes" not in obj_json:
         msg = f"unexpected data for obj id: {obj_id}"
-        msg.error(msg)
+        log.error(msg)
         raise HTTPInternalServerError()
 
     # return a list of attributes based on sorted dictionary keys
@@ -309,7 +309,7 @@ async def POST_Attributes(request):
     log.debug(f"Get attributes obj_id: {obj_id} got json")
     if "attributes" not in obj_json:
         msg = f"unexpected data for obj id: {obj_id}"
-        msg.error(msg)
+        log.error(msg)
         raise HTTPInternalServerError()
 
     # return a list of attributes based on sorted dictionary keys
@@ -577,7 +577,7 @@ async def DELETE_Attributes(request):
     log.debug(f"DELETE attributes obj_id: {obj_id} got json")
     if "attributes" not in obj_json:
         msg = f"unexpected data for obj id: {obj_id}"
-        msg.error(msg)
+        log.error(msg)
         raise HTTPInternalServerError()
 
     # return a list of attributes based on sorted dictionary keys

hsds/basenode.py

@@ -551,13 +551,15 @@ def baseInit(node_type):
     # setup log config
     log_level = config.get("log_level")
     prefix = config.get("log_prefix")
-    log_timestamps = config.get("log_timestamps", default=False)
+    log_timestamps = config.get("log_timestamps", default=True)
+    log_format = config.get("log_format", default="text")
 
     # Make stdout/stderr encoding consistent across all operating systems
     sys.stdout.reconfigure(encoding="utf-8")
     sys.stderr.reconfigure(encoding="utf-8")
 
-    log.setLogConfig(log_level, prefix=prefix, timestamps=log_timestamps)
+    kwargs = {"prefix": prefix, "timestamps": log_timestamps, "log_format": log_format}
+    log.setLogConfig(log_level, **kwargs)
 
     # create the app object
     log.info("Application baseInit")

hsds/chunk_sn.py

@@ -385,7 +385,7 @@ async def _getRequestData(request, http_streaming=True):
         log.debug(f"getRequestData - got json: {body}")
         if "value" in body:
             input_data = body["value"]
-            log.debug("input_data: {input_data}")
+            log.debug(f"input_data: {input_data}")
         elif "value_base64" in body:
             base64_data = body["value_base64"]
             base64_data = base64_data.encode("ascii")

hsds/chunklocator.py

@@ -187,8 +187,10 @@ def main():
     # setup log config
     log_level = config.get("log_level")
     prefix = config.get("log_prefix")
-    log_timestamps = config.get("log_timestamps", default=False)
-    log.setLogConfig(log_level, prefix=prefix, timestamps=log_timestamps)
+    log_timestamps = config.get("log_timestamps", default=True)
+    log_format = config.get("log_format", default="text")
+    kwargs = {"prefix": prefix, "timestamps": log_timestamps, "log_format": log_format}
+    log.setLogConfig(log_level, **kwargs)
     start_time = time.time()
     log.info(f"chunklocator start: {start_time:.2f}")
 

hsds/ctype_dn.py

@@ -91,7 +91,7 @@ async def POST_Datatype(request):
 
     ctype_id = get_obj_id(request, body=body)
     if not isValidUuid(ctype_id, obj_class="datatype"):
-        log.error("Unexpected type_id: {ctype_id}")
+        log.error(f"Unexpected type_id: {ctype_id}")
         raise HTTPInternalServerError()
 
     # verify the id doesn't already exist

hsds/group_dn.py

@@ -319,7 +319,7 @@ async def POST_Root(request):
         try:
             timestamp = int(params["timestamp"])
         except ValueError:
-            log.error("unexpected value for timestamp: {params}")
+            log.error(f"unexpected value for timestamp: {params}")
             raise HTTPInternalServerError()
     else:
         timestamp = getNow(app)

hsds/headnode.py

@@ -475,6 +475,12 @@ async def init():
 
     # set a bunch of global state
     app["id"] = createNodeId("head")
+    app["node_type"] = "head"
+    # the head node has no INITIALIZING->READY lifecycle (SN/DN register
+    # with it); it is ready as soon as it is listening.  Without this,
+    # log.request's admission check 503s the head's own status routes
+    # (e.g. /, /nodestate/{nodetype}, /nodeinfo/{statkey}).
+    app["node_state"] = "READY"
 
     bucket_name = config.get("bucket_name")
     if bucket_name:

hsds/hsds_app.py

@@ -11,28 +11,11 @@
 
 
 def _enqueue_output(out, queue, loglevel):
+    # loglevel is unused: level filtering happens in the node processes
+    # themselves (hsds_logger), lines are passed through verbatim
     try:
         for line in iter(out.readline, b""):
-            # filter lines by loglevel
-            words = line.split()
-            put_line = True
-
-            if loglevel != logging.DEBUG:
-                if len(words) >= 2:
-                    # format should be "node_name log_level> msg"
-                    level = words[1][:-1]
-                    if loglevel == logging.INFO:
-                        if level == "DEBUG":
-                            put_line = False
-                    elif loglevel == logging.WARN or loglevel == logging.WARNING:
-                        if not level.startswith("WARN") and level != "ERROR":
-                            put_line = False
-                    elif loglevel == logging.ERROR:
-                        if level != "ERROR":
-                            put_line = False
-            put_line = True
-            if put_line:
-                queue.put(line)
+            queue.put(line)
         logging.debug("_enqueue_output close()")
         out.close()
     except ValueError as ve:

hsds/hsds_logger.py

@@ -10,11 +10,19 @@
 # request a copy from help@hdfgroup.org.                                     #
 ##############################################################################
 #
-# Simple looger for hsds
+# Simple logger for hsds
+#
+# Supports "text" and "json" output formats and W3C trace context
+# (traceparent header) propagation so that log lines can be correlated
+# per-request across SN and DN nodes (and with upstream clients).
 #
 
 import asyncio
+import json
+import re
+import secrets
 import time
+from contextvars import ContextVar
 from aiohttp.web_exceptions import HTTPServiceUnavailable
 from .util.domainUtil import getDomainFromRequest
 
@@ -27,7 +35,14 @@
 req_count = {"GET": 0, "POST": 0, "PUT": 0, "DELETE": 0, "num_tasks": 0}
 log_count = {"DEBUG": 0, "INFO": 0, "WARN": 0, "ERROR": 0}
 # the following defaults will be adjusted by the app
-config = {"log_level": DEBUG, "prefix": "", "timestamps": False}
+config = {"log_level": DEBUG, "prefix": "", "timestamps": False, "log_format": "text"}
+
+# (trace_id, span_id, trace_flags) for the request being handled by the
+# current asyncio task tree - see newTraceContext()
+_trace_ctx = ContextVar("hsds_trace_ctx", default=None)
+
+_TRACE_ID_RGX = re.compile(r"[0-9a-f]{32}")
+_TRACE_FLAGS_RGX = re.compile(r"[0-9a-f]{2}")
 
 
 def _getLevelName(level):
@@ -44,7 +59,7 @@ def _getLevelName(level):
     return name
 
 
-def setLogConfig(level, prefix=None, timestamps=None):
+def setLogConfig(level, prefix=None, timestamps=None, log_format=None):
     if level == "DEBUG":
         config["log_level"] = DEBUG
     elif level == "INFO":
@@ -57,11 +72,55 @@ def setLogConfig(level, prefix=None, timestamps=None):
         config["log_level"] = ERROR
     else:
         raise ValueError(f"unexpected log_level: {level}")
-    # print(f"setLogConfig - level={level}")
     if prefix is not None:
         config["prefix"] = prefix
     if timestamps is not None:
         config["timestamps"] = timestamps
+    if log_format is not None:
+        if log_format not in ("text", "json"):
+            raise ValueError(f"unexpected log_format: {log_format}")
+        config["log_format"] = log_format
+
+
+def newTraceContext(traceparent=None):
+    """Set the trace context for the current asyncio task tree and return
+    the trace id.
+
+    If traceparent is a valid W3C traceparent header ("00-<trace_id>-
+    <parent_span_id>-<flags>"), the incoming trace id and flags are adopted;
+    otherwise a new trace id is generated.  A new span id is generated
+    either way to identify this node's hop.
+    """
+    trace_id = None
+    flags = "01"
+    if traceparent:
+        parts = traceparent.strip().lower().split("-")
+        if len(parts) == 4 and _TRACE_ID_RGX.fullmatch(parts[1]):
+            if parts[1] != "0" * 32:
+                trace_id = parts[1]
+                if _TRACE_FLAGS_RGX.fullmatch(parts[3]):
+                    flags = parts[3]
+    if trace_id is None:
+        trace_id = secrets.token_hex(16)
+    span_id = secrets.token_hex(8)
+    _trace_ctx.set((trace_id, span_id, flags))
+    return trace_id
+
+
+def getTraceId():
+    """Return the trace id for the current request, or None."""
+    ctx = _trace_ctx.get()
+    return ctx[0] if ctx else None
+
+
+def getTraceParent():
+    """Return a W3C traceparent header value for outgoing requests made on
+    behalf of the current request, or None if no trace context is set."""
+    ctx = _trace_ctx.get()
+    if ctx is None:
+        return None
+    trace_id, span_id, flags = ctx
+    return f"00-{trace_id}-{span_id}-{flags}"
 
 
 def _activeTaskCount():
@@ -72,29 +131,52 @@ def _activeTaskCount():
     return count
 
 
-def _timestamp():
+def _isotime():
+    now = time.time()
+    ms = int(now * 1000) % 1000
+    s = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(now))
+    return f"{s}.{ms:03d}Z"
+
 
+def _timestamp():
     if config["timestamps"]:
-        now = time.time()
-        ts = f"{now:.3f} "
+        ts = _isotime() + " "
     else:
         ts = ""
 
     return ts
 
 
+def _emit(level_name, msg, tag=None, extra=None):
+    """Print one log line in the configured format.  tag replaces the level
+    name in text format (used for the REQ/RSP access log lines); extra
+    fields are only emitted in json format."""
+    prefix = config["prefix"]
+    trace_id = getTraceId()
+    if config["log_format"] == "json":
+        obj = {"time": _isotime(), "level": level_name}
+        if prefix:
+            obj["node"] = prefix.strip()
+        if trace_id:
+            obj["trace_id"] = trace_id
+        if extra:
+            obj.update(extra)
+        obj["msg"] = msg
+        print(json.dumps(obj, default=str))
+    else:
+        if tag is None:
+            tag = level_name
+        ts = _timestamp()
+        trace = f"[{trace_id}] " if trace_id else ""
+        print(f"{prefix}{ts}{tag}> {trace}{msg}")
+
+
 def _logMsg(level, msg):
     if config["log_level"] > level:
         return  # ignore
 
-    ts = _timestamp()
-
-    prefix = config["prefix"]
-
     level_name = _getLevelName(level)
-
-    print(f"{prefix}{ts}{level_name}> {msg}")
-
+    _emit(level_name, msg)
     log_count[level_name] += 1
 
 
@@ -121,13 +203,16 @@ def error(msg):
 def request(req):
     app = req.app
     domain = getDomainFromRequest(req, validate=False)
-    prefix = config["prefix"]
-    ts = _timestamp()
+    # adopt the caller's trace context (or start a new trace) so all log
+    # lines emitted while handling this request carry the same trace id
+    newTraceContext(req.headers.get("traceparent"))
 
-    msg = f"{prefix}{ts}REQ> {req.method}: {req.path}"
+    msg = f"{req.method}: {req.path}"
+    extra = {"method": req.method, "path": req.path}
     if domain:
         msg += f" [{domain}]"
-    print(msg)
+        extra["domain"] = domain
+    _emit("INFO", msg, tag="REQ", extra=extra)
 
     INFO_METHODS = (
         "/about",
@@ -178,30 +263,7 @@ def response(req, resp=None, code=None, message=None):
         else:
             level = ERROR
 
-    log_level = config["log_level"]
-
-    if log_level == DEBUG:
-        prefix = config["prefix"]
-        ts = _timestamp()
-
-        num_tasks = len(asyncio.all_tasks())
-        active_tasks = _activeTaskCount()
-
-        debug(f"rsp - num tasks: {num_tasks} active tasks: {active_tasks}")
-
-        s = "{}{} RSP> <{}> ({}): {}"
-        print(s.format(prefix, ts, code, message, req.path))
-
-    elif log_level <= level:
-        prefix = config["prefix"]
-        ts = _timestamp()
-
-        num_tasks = len(asyncio.all_tasks())
-        active_tasks = _activeTaskCount()
-
-        debug(f"num tasks: {num_tasks} active tasks: {active_tasks}")
-
-        s = "{}{} RSP> <{}> ({}): {}"
-        print(s.format(prefix, ts, code, message, req.path))
-    else:
-        pass
+    if config["log_level"] <= level:
+        msg = f"<{code}> ({message}): {req.path}"
+        extra = {"status": code, "reason": message, "path": req.path}
+        _emit(_getLevelName(level), msg, tag="RSP", extra=extra)

hsds/link_dn.py

@@ -117,7 +117,7 @@ async def GET_Links(request):
 
     log.debug(f"for id: {group_id} got group json: {group_json}")
     if "links" not in group_json:
-        msg.error(f"unexpected group data for id: {group_id}")
+        log.error(f"unexpected group data for id: {group_id}")
         raise HTTPInternalServerError()
 
     # return a list of links based on sorted dictionary keys
@@ -130,7 +130,7 @@ async def GET_Links(request):
             titles = [x for x in titles if globmatch(x, pattern)]
         except ValueError:
             log.error(f"exception getting links using pattern: {pattern}")
-            raise HTTPBadRequest(reason=msg)
+            raise HTTPBadRequest(reason=f"invalid link name pattern: {pattern}")
         msg = f"getLinks with pattern: {pattern} returning {len(titles)} "
         msg += f"links from {len(link_dict)}"
         log.debug(msg)