Parse Startup Message

src/alogamous/log_line_parser.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+
+class LineType:
+    HEADER_LINE = "header line"
+    LOG_LINE = "log line"
+    UNSTRUCTURED_LINE = "unstructured line"
+
+
+class LogLineParser:
+    def __init__(self, header_line: str, expected_fields: list[str], seperator: str):
+        self.header_line = header_line
+        self.expected_fields = expected_fields
+        self.separator = seperator
+        self.separator_count = len(self.expected_fields) - 1
+
+    def parse(self, line):
+        if line == self.header_line:
+            return {"type": LineType.HEADER_LINE, "line": line}
+        if line.count(self.separator) == self.separator_count:
+            parsed_line = {"type": LineType.LOG_LINE}
+            separated_line = line.split(self.separator)
+            for index in range(len(self.expected_fields)):
+                parsed_line[self.expected_fields[index]] = separated_line[index]
+            return parsed_line
+        return {"type": LineType.UNSTRUCTURED_LINE, "line": line}

src/alogamous/startup_header_analyzer.py

@@ -0,0 +1,24 @@
+from alogamous import analyzer, log_line_parser
+
+parser = log_line_parser.LogLineParser(
+    "====================================================", ["datetime", "source", "level", "message"], " - "
+)
+
+
+class StartupHeaderAnalyzer(analyzer.Analyzer):
+    def __init__(self):
+        self.startup_block = False
+        self.startup_lines = []
+
+    def read_log_line(self, line):
+        line_type = parser.parse(line)["type"]
+        if self.startup_block is False and line_type == log_line_parser.LineType.HEADER_LINE:
+            self.startup_block = True
+        elif self.startup_block is True:
+            if line_type == log_line_parser.LineType.UNSTRUCTURED_LINE:
+                self.startup_lines.append(line)
+            elif line_type == log_line_parser.LineType.HEADER_LINE:
+                self.startup_block = False
+
+    def report(self, out_stream):
+        out_stream.write("\n".join(self.startup_lines))

tests/log_line_parser_test.py

@@ -0,0 +1,45 @@
+from alogamous import log_line_parser
+
+
+def test_parse_header_line():
+    parser = log_line_parser.LogLineParser(
+        "====================================================", ["datetime", "source", "level", "message"], " - "
+    )
+    line = "===================================================="
+    assert parser.parse(line) == {"type": "header line", "line": "===================================================="}
+
+
+def test_parse_log_line():
+    parser = log_line_parser.LogLineParser(
+        "====================================================", ["datetime", "source", "level", "message"], " - "
+    )
+    line = (
+        "2024-06-20 11:00:18,172 - aiokafka.consumer.subscription_state - INFO - Updating subscribed topics to: "
+        "frozenset({'internal'})"
+    )
+    assert parser.parse(line) == {
+        "type": "log line",
+        "datetime": "2024-06-20 11:00:18,172",
+        "source": "aiokafka.consumer.subscription_state",
+        "level": "INFO",
+        "message": "Updating subscribed topics to: frozenset({'internal'})",
+    }
+
+
+def test_parse_deviant_log_line():
+    parser = log_line_parser.LogLineParser(
+        "====================================================", ["datetime", "source", "level", "message"], " - "
+    )
+    line = "Hello I am a bad log line"
+    assert parser.parse(line) == {"type": "unstructured line", "line": "Hello I am a bad log line"}
+
+
+def test_parse_start_header_content():
+    parser = log_line_parser.LogLineParser(
+        "====================================================", ["datetime", "source", "level", "message"], " - "
+    )
+    line = "    Start time: 2024-06-20 09:00:00.001550+00:00"
+    assert parser.parse(line) == {
+        "type": "unstructured line",
+        "line": "    Start time: 2024-06-20 09:00:00.001550+00:00",
+    }

tests/startup_header_analyzer_test.py

@@ -0,0 +1,29 @@
+import io
+
+from alogamous import startup_header_analyzer
+
+
+def test_report():
+    startup_analyzer = startup_header_analyzer.StartupHeaderAnalyzer()
+    in_stream = """
+====================================================
+STARTING Tracking service
+    Start time: 2024-06-20 09:00:00.001550+00:00
+    Version: 2729a
+    Command line: ['.venv/bin/python3', '-m', 'app.tracking_service', '--market', 'US', '--version', '2729a']
+====================================================
+2024-06-20 11:00:17,983 - root - INFO - Adding subscription for pid None
+2024-06-20 11:00:18,115 - root - INFO - Initialized Influx DB Client to host
+2024-06-20 11:00:18,115 - root - INFO - Scheduling Error Handler in 150.0 seconds
+"""
+    out_stream = io.StringIO()
+    for line in in_stream.splitlines():
+        startup_analyzer.read_log_line(line)
+    startup_analyzer.report(out_stream)
+    assert (
+        out_stream.getvalue()
+        == """STARTING Tracking service
+    Start time: 2024-06-20 09:00:00.001550+00:00
+    Version: 2729a
+    Command line: ['.venv/bin/python3', '-m', 'app.tracking_service', '--market', 'US', '--version', '2729a']"""
+    )