diff --git a/admin/config/config.yml b/admin/config/config.yml index 756be465..c7562e00 100755 --- a/admin/config/config.yml +++ b/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 diff --git a/hsds/attr_dn.py b/hsds/attr_dn.py index 456e9854..9a4e00f4 100755 --- a/hsds/attr_dn.py +++ b/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 diff --git a/hsds/basenode.py b/hsds/basenode.py index 69a00531..954ae998 100644 --- a/hsds/basenode.py +++ b/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") diff --git a/hsds/chunk_sn.py b/hsds/chunk_sn.py index 8a86c8a8..ee77b1a2 100755 --- a/hsds/chunk_sn.py +++ b/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") diff --git a/hsds/chunklocator.py b/hsds/chunklocator.py index 6727de9e..287fe469 100644 --- a/hsds/chunklocator.py +++ b/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}") diff --git a/hsds/ctype_dn.py b/hsds/ctype_dn.py index f06b98b3..d1e8dcee 100755 --- a/hsds/ctype_dn.py +++ b/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 diff --git a/hsds/group_dn.py b/hsds/group_dn.py index 0a6bb937..ef827f61 100755 --- a/hsds/group_dn.py +++ b/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) diff --git a/hsds/headnode.py b/hsds/headnode.py index 55547294..b81a0c64 100755 --- a/hsds/headnode.py +++ b/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: diff --git a/hsds/hsds_app.py b/hsds/hsds_app.py index e690b68d..a3274636 100644 --- a/hsds/hsds_app.py +++ b/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: diff --git a/hsds/hsds_logger.py b/hsds/hsds_logger.py index 3421d981..d2cb4dd1 100644 --- a/hsds/hsds_logger.py +++ b/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-- + -"), 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) diff --git a/hsds/link_dn.py b/hsds/link_dn.py index f7ec5956..316aaafe 100755 --- a/hsds/link_dn.py +++ b/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) diff --git a/hsds/util/httpUtil.py b/hsds/util/httpUtil.py index 0d43ae4a..4edf63a1 100644 --- a/hsds/util/httpUtil.py +++ b/hsds/util/httpUtil.py @@ -280,6 +280,15 @@ async def request_read(request, count=None) -> bytes: return bytes(body) +def _addTraceHeader(kwargs): + """Add a traceparent header to outgoing request kwargs when the current + request has a trace context (see hsds_logger.newTraceContext), so log + lines can be correlated across SN and DN nodes.""" + traceparent = log.getTraceParent() + if traceparent: + kwargs["headers"] = {"traceparent": traceparent} + + async def http_get(app, url, params=None, client=None): """ Helper function - async HTTP GET @@ -289,10 +298,11 @@ async def http_get(app, url, params=None, client=None): client = get_http_client(app, url=url) url = get_http_std_url(url) status_code = None - timeout = config.get("timeout") + kwargs = {"params": params, "timeout": config.get("timeout")} + _addTraceHeader(kwargs) # TBD: use read_bufsize parameter to optimize read for large responses try: - async with client.get(url, params=params, timeout=timeout) as rsp: + async with client.get(url, **kwargs) as rsp: log.info(f"http_get status: {rsp.status} for req: {url}") status_code = rsp.status if rsp.status == 200: @@ -366,6 +376,7 @@ async def http_post(app, url, data=None, params=None, client=None): kwargs["timeout"] = timeout if params: kwargs["params"] = params + _addTraceHeader(kwargs) try: async with client.post(url, **kwargs) as rsp: @@ -439,6 +450,7 @@ async def http_put(app, url, data=None, params=None, client=None): timeout = config.get("timeout") if timeout: kwargs["timeout"] = timeout + _addTraceHeader(kwargs) try: async with client.put(url, **kwargs) as rsp: @@ -501,6 +513,7 @@ async def http_delete(app, url, data=None, params=None, client=None): kwargs["timeout"] = timeout if params: kwargs["params"] = params + _addTraceHeader(kwargs) try: async with client.delete(url, **kwargs) as rsp: diff --git a/tests/unit/logger_test.py b/tests/unit/logger_test.py new file mode 100644 index 00000000..340dd13f --- /dev/null +++ b/tests/unit/logger_test.py @@ -0,0 +1,181 @@ +############################################################################## +# Copyright by The HDF Group. # +# All rights reserved. # +# # +# This file is part of HSDS (HDF5 Scalable Data Service), Libraries and # +# Utilities. The full HSDS copyright notice, including # +# terms governing use, modification, and redistribution, is contained in # +# the file COPYING, which can be found at the root of the source code # +# distribution tree. If you do not have access to this file, you may # +# request a copy from help@hdfgroup.org. # +############################################################################## +import contextvars +import io +import json +import re +import sys +import unittest +from contextlib import redirect_stdout + +sys.path.append("../..") +from hsds import hsds_logger as log + +ISO_TS = r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z" +HEX32 = "0123456789abcdef01234567890abcde" +HEX16 = "0123456789abcdef" + + +def capture(func, *args, **kwargs): + """Run func in a copied context (so trace context doesn't leak between + tests) and return whatever it printed to stdout.""" + buf = io.StringIO() + ctx = contextvars.copy_context() + with redirect_stdout(buf): + ctx.run(func, *args, **kwargs) + return buf.getvalue() + + +class LoggerTest(unittest.TestCase): + def setUp(self): + log.setLogConfig("DEBUG", prefix="", timestamps=False, log_format="text") + + def testTextFormat(self): + out = capture(log.info, "hello") + self.assertEqual(out, "INFO> hello\n") + out = capture(log.warn, "uh oh") + self.assertEqual(out, "WARN> uh oh\n") + + def testTextPrefixAndTimestamp(self): + log.setLogConfig("INFO", prefix="sn1 ", timestamps=True) + out = capture(log.info, "hello") + self.assertTrue(re.fullmatch(f"sn1 {ISO_TS} INFO> hello\n", out), out) + + def testLevelFiltering(self): + log.setLogConfig("WARNING") + self.assertEqual(capture(log.debug, "nope"), "") + self.assertEqual(capture(log.info, "nope"), "") + self.assertEqual(capture(log.error, "yep"), "ERROR> yep\n") + + def testJsonFormat(self): + log.setLogConfig("INFO", log_format="json") + out = capture(log.warning, "watch out") + obj = json.loads(out) + self.assertEqual(obj["level"], "WARN") + self.assertEqual(obj["msg"], "watch out") + self.assertTrue(re.fullmatch(ISO_TS, obj["time"]), obj) + self.assertNotIn("trace_id", obj) + + def testJsonNodePrefix(self): + log.setLogConfig("INFO", prefix="dn2 ", log_format="json") + obj = json.loads(capture(log.info, "hi")) + self.assertEqual(obj["node"], "dn2") + + def testBadConfig(self): + with self.assertRaises(ValueError): + log.setLogConfig("VERBOSE") + with self.assertRaises(ValueError): + log.setLogConfig("INFO", log_format="xml") + + def testNewTraceContextFromHeader(self): + def check(): + traceparent = f"00-{HEX32}-{HEX16}-01" + trace_id = log.newTraceContext(traceparent) + self.assertEqual(trace_id, HEX32) + self.assertEqual(log.getTraceId(), HEX32) + outgoing = log.getTraceParent() + # same trace, but a new span id for this hop + parts = outgoing.split("-") + self.assertEqual(len(parts), 4) + self.assertEqual(parts[0], "00") + self.assertEqual(parts[1], HEX32) + self.assertNotEqual(parts[2], HEX16) + self.assertTrue(re.fullmatch(r"[0-9a-f]{16}", parts[2])) + self.assertEqual(parts[3], "01") + + contextvars.copy_context().run(check) + + def testNewTraceContextGenerated(self): + def check(): + trace_id = log.newTraceContext(None) + self.assertTrue(re.fullmatch(r"[0-9a-f]{32}", trace_id)) + other = log.newTraceContext(None) + self.assertNotEqual(trace_id, other) + + contextvars.copy_context().run(check) + + def testNewTraceContextInvalidHeader(self): + def check(): + for bad in ("junk", "00-zz-yy-01", f"00-{'0' * 32}-{HEX16}-01"): + trace_id = log.newTraceContext(bad) + self.assertTrue(re.fullmatch(r"[0-9a-f]{32}", trace_id), bad) + + contextvars.copy_context().run(check) + + def testNoTraceContext(self): + self.assertIsNone(log.getTraceId()) + self.assertIsNone(log.getTraceParent()) + + def testTraceIdInTextOutput(self): + def emit(): + log.newTraceContext(f"00-{HEX32}-{HEX16}-01") + log.info("with trace") + + out = capture(emit) + self.assertEqual(out, f"INFO> [{HEX32}] with trace\n") + + def testTraceIdInJsonOutput(self): + log.setLogConfig("INFO", log_format="json") + + def emit(): + log.newTraceContext(f"00-{HEX32}-{HEX16}-01") + log.info("with trace") + + obj = json.loads(capture(emit)) + self.assertEqual(obj["trace_id"], HEX32) + + def testLogCount(self): + before = log.log_count["INFO"] + capture(log.info, "counted") + self.assertEqual(log.log_count["INFO"], before + 1) + + +class TracePropagationTest(unittest.IsolatedAsyncioTestCase): + """The traceparent header is forwarded on all outgoing http_* requests + (SN -> DN hops) when a trace context is set.""" + + async def test_http_methods_send_traceparent(self): + import aiohttp + from aiohttp import web + from aiohttp.test_utils import TestServer + from hsds.util.httpUtil import http_get, http_post, http_put, http_delete + + log.setLogConfig("ERROR") # quiet the http_* info logging + seen = {} + + async def handler(request): + seen[request.method] = request.headers.get("traceparent") + return web.json_response({"ok": True}) + + app = web.Application() + app.router.add_route("*", "/", handler) + server = TestServer(app) + await server.start_server() + try: + async with aiohttp.ClientSession() as session: + log.newTraceContext(None) + url = str(server.make_url("/")) + await http_get({}, url, client=session) + await http_post({}, url, data={}, client=session) + await http_put({}, url, data={}, client=session) + await http_delete({}, url, client=session) + finally: + await server.close() + + expected = log.getTraceParent() + self.assertIsNotNone(expected) + for method in ("GET", "POST", "PUT", "DELETE"): + self.assertEqual(seen.get(method), expected, method) + + +if __name__ == "__main__": + unittest.main()