494 lines
16 KiB
Python
494 lines
16 KiB
Python
import argparse
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import sqlite3
|
|
|
|
from io import TextIOWrapper
|
|
from typing import Iterable, Iterator, Optional
|
|
|
|
level_ordinals = {
|
|
"Diag": 1,
|
|
"Info": 2,
|
|
"Warn": 3,
|
|
"Error": 4,
|
|
"Fatal": 5,
|
|
}
|
|
log_msg_start_re = re.compile('^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}.\d{6}\|')
|
|
|
|
|
|
def _get_arg_parser():
|
|
parser = argparse.ArgumentParser(
|
|
description="Builds the Spark server (A.K.A sparkd, the daemon)")
|
|
parser.add_argument("--pattern", help="Regex pattern to filter logs")
|
|
parser.add_argument("--file", help="The log file to dump")
|
|
parser.add_argument(
|
|
"--category",
|
|
help=
|
|
"The category (subsystem) to grep (e.g., Gen, Http, Md, OrderMan etc.)"
|
|
)
|
|
parser.add_argument("--level",
|
|
help="The minimum log level to include",
|
|
choices=["Diag", "Info", "Warn", "Error", "Fatal"])
|
|
parser.add_argument(
|
|
"--source-file",
|
|
help="Source file(s) whose logs to include (includes all by default)")
|
|
parser.add_argument("--line-num",
|
|
help="The line number in the source file(s) to grep")
|
|
parser.add_argument("--count",
|
|
help="Count the number of log messages",
|
|
action="store_true",
|
|
dest="count")
|
|
parser.add_argument(
|
|
"--count-only",
|
|
help="Count the number of log messages without printing them",
|
|
action="store_true",
|
|
dest="count_only")
|
|
parser.add_argument(
|
|
"--count-all",
|
|
help="Count the number of log messages in each category",
|
|
action="store_true",
|
|
dest="count_all")
|
|
parser.add_argument(
|
|
"--dump-db",
|
|
help="Dump logs to a sqlite3 db (named <--file>.sqlite)",
|
|
action="store_true")
|
|
parser.add_argument("--dump-db-slog",
|
|
help="Dump structured logs to sqlite3 db",
|
|
action="store_true")
|
|
parser.add_argument("--dump-db-to",
|
|
help="Dump logs to a specific sqlite3 db",
|
|
metavar="SQLITE_FILE")
|
|
parser.add_argument("--dump-db-table-to",
|
|
help="Dump logs to a specific table in a sqlite3 db",
|
|
metavar="SQLITE_TABLE")
|
|
parser.add_argument("-y, --yes",
|
|
help="Overwrite existing db without asking",
|
|
action="store_true",
|
|
dest="yes")
|
|
parser.set_defaults(
|
|
pattern=None,
|
|
file=None,
|
|
level=0,
|
|
category=None,
|
|
count=False,
|
|
count_only=False,
|
|
dump_db=False,
|
|
dump_db_to=None,
|
|
dump_db_slog=False,
|
|
)
|
|
return parser
|
|
|
|
|
|
class ChunkIterator:
|
|
def __init__(self, source: Iterator, chunk_size: int):
|
|
self.source = source
|
|
self.chunk_size = chunk_size
|
|
|
|
def __iter__(self):
|
|
return self
|
|
|
|
def __next__(self):
|
|
chunk = []
|
|
try:
|
|
while len(chunk) < self.chunk_size:
|
|
chunk.append(self.source.__next__())
|
|
except StopIteration:
|
|
pass
|
|
if chunk:
|
|
return chunk
|
|
raise StopIteration()
|
|
|
|
|
|
class LogRecord:
|
|
def __init__(self, msg: str):
|
|
self.full_msg = msg
|
|
self.parts = []
|
|
field_start = 0
|
|
for _ in range(0, 9):
|
|
field_end = msg.find("|", field_start)
|
|
field = msg[field_start:field_end].strip()
|
|
field_start = field_end + 1
|
|
self.parts.append(field)
|
|
self.parts.append(msg[field_end + 1:].rstrip())
|
|
|
|
@property
|
|
def ts(self):
|
|
return self.parts[0]
|
|
|
|
@property
|
|
def ts_nanos(self):
|
|
return self.parts[1]
|
|
|
|
@property
|
|
def category(self):
|
|
return self.parts[2]
|
|
|
|
@property
|
|
def level(self):
|
|
return self.parts[3]
|
|
|
|
@property
|
|
def thread_id(self):
|
|
return self.parts[4]
|
|
|
|
@property
|
|
def thread_name(self):
|
|
return self.parts[5]
|
|
|
|
@property
|
|
def pid(self):
|
|
return self.parts[6]
|
|
|
|
@property
|
|
def src_file(self):
|
|
i = self.parts[7].find(":")
|
|
return self.parts[7][:i]
|
|
|
|
@property
|
|
def line_num(self):
|
|
i = self.parts[7].find(":")
|
|
return self.parts[7][i + 1:]
|
|
|
|
@property
|
|
def deployment_type(self):
|
|
return self.parts[8]
|
|
|
|
@property
|
|
def msg(self):
|
|
return self.parts[9]
|
|
|
|
def __str__(self):
|
|
return self.full_msg.rstrip()
|
|
|
|
|
|
class LogRecordIterator:
|
|
def __init__(self, log_file: TextIOWrapper):
|
|
self.log_file = log_file
|
|
self.count = 0
|
|
self.curr_log = None
|
|
self.next_ln = None
|
|
self.next_log_start = None
|
|
|
|
def __iter__(self):
|
|
return self
|
|
|
|
def _read_until_next_log(self):
|
|
curr_log = self.next_log_start
|
|
next_ln = self.log_file.readline()
|
|
while next_ln and not log_msg_start_re.match(next_ln):
|
|
curr_log += next_ln
|
|
next_ln = self.log_file.readline()
|
|
self.next_log_start = next_ln
|
|
return LogRecord(curr_log)
|
|
|
|
# 45932673469
|
|
def __next__(self):
|
|
if self.next_log_start is None:
|
|
self.next_log_start = self.log_file.readline()
|
|
if not self.next_log_start:
|
|
self.log_file.close()
|
|
raise StopIteration
|
|
return self._read_until_next_log()
|
|
|
|
|
|
class LogFilterCriteria:
|
|
def __init__(
|
|
self,
|
|
level: Optional[int | str] = None,
|
|
category: Optional[str] = None,
|
|
source_file: Optional[str] = None,
|
|
line_num: Optional[int] = None,
|
|
pattern: Optional[str] = None,
|
|
):
|
|
self.level = level_ordinals[level] if isinstance(level, str) else level
|
|
self.category = category
|
|
self.source_file = source_file
|
|
self.line_num = line_num
|
|
self.pattern_re = re.compile(pattern) if pattern else None
|
|
|
|
def include(self, log_rec: LogRecord):
|
|
if level_ordinals[log_rec.level] < self.level:
|
|
return False
|
|
if self.source_file:
|
|
if self.source_file != log_rec.src_file:
|
|
return False
|
|
if self.line_num and self.line_num != log_rec.line_num:
|
|
return False
|
|
if self.category and self.category != log_rec.category:
|
|
return False
|
|
if (self.pattern_re and not self.pattern_re.match(log_rec.msg)):
|
|
return False
|
|
return True
|
|
|
|
|
|
def dump_to_db(log_recs: Iterable[LogRecord], db_out: str, table_name: str):
|
|
con = sqlite3.connect(db_out)
|
|
cur = con.cursor()
|
|
columns = [
|
|
"ts TEXT",
|
|
"ts_nanos INTEGER",
|
|
"category TEXT",
|
|
"level INTEGER",
|
|
"thread_id INTEGER",
|
|
"thread_name TEXT",
|
|
"pid INTEGER",
|
|
"src_file TEXT",
|
|
"line_num TEXT",
|
|
"deployment_type TEXT",
|
|
"msg TEXT",
|
|
]
|
|
columns_expr = ", ".join(columns)
|
|
cur.execute(f"CREATE TABLE IF NOT EXISTS {table_name}({columns_expr})")
|
|
insert_expr = f"INSERT INTO {table_name} VALUES ({', '.join(['?' for _ in columns])})"
|
|
chunks = ChunkIterator([(lr.ts, lr.ts_nanos, lr.category, lr.level,
|
|
lr.thread_id, lr.thread_name, lr.pid, lr.src_file,
|
|
lr.line_num, lr.deployment_type, lr.msg.rstrip())
|
|
for lr in log_recs].__iter__(), 256)
|
|
for chunk in chunks:
|
|
cur.executemany(insert_expr, chunk)
|
|
print(f"Commiting sqlite transactions")
|
|
con.commit()
|
|
|
|
|
|
class StructuredLogTable:
|
|
def __init__(self, table_name: str, columns: list[(str, str)],
|
|
cur: sqlite3.Cursor):
|
|
self.table_name = table_name
|
|
self.columns = columns
|
|
self.create_expr = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join([n + ' ' + t for n, t in columns])})"
|
|
self.insert_expr = f"INSERT INTO {self.table_name} VALUES ({', '.join(['?' for _ in columns])})"
|
|
self.buffered_rows = []
|
|
self.cur = cur
|
|
|
|
def create_table(self):
|
|
self.cur.execute(self.create_expr)
|
|
|
|
def flush(self):
|
|
self.cur.executemany(self.insert_expr, self.buffered_rows)
|
|
self.buffered_rows = []
|
|
|
|
def add(self, log_msg: str):
|
|
j = json.loads(log_msg)
|
|
vals = []
|
|
for cname, _ in self.columns:
|
|
v = j.get(cname)
|
|
if cname == "d":
|
|
if v is None:
|
|
v = "{}"
|
|
elif isinstance(v, dict):
|
|
v = json.dumps(v)
|
|
vals.append(v)
|
|
self.buffered_rows.append(vals)
|
|
if len(self.buffered_rows) > 128:
|
|
self.flush()
|
|
|
|
|
|
def get_structured_log_tables(
|
|
cur: sqlite3.Cursor) -> dict[str, StructuredLogTable]:
|
|
def execution_report_table(cur: sqlite3.Cursor):
|
|
table_name = "execution_report"
|
|
columns = [
|
|
("ts", "INTEGER"),
|
|
("acc", "TEXT"),
|
|
("u", "TEXT"),
|
|
("id", "INTEGER"),
|
|
("sid", "INTEGER"),
|
|
("sec", "TEXT"),
|
|
("s", "TEXT"),
|
|
("exch", "TEXT"),
|
|
("eid", "TEXT"),
|
|
("p", "REAL"),
|
|
("q", "REAL"),
|
|
("fq", "REAL"),
|
|
("lq", "REAL"),
|
|
("typ", "TEXT"),
|
|
("tif", "TEXT"),
|
|
("nst", "TEXT"),
|
|
("st", "TEXT"),
|
|
("d", "TEXT"),
|
|
]
|
|
return StructuredLogTable(table_name, columns, cur)
|
|
|
|
def create_order_table(cur: sqlite3.Cursor):
|
|
table_name = "create_order"
|
|
columns = [
|
|
("ts", "INTEGER"),
|
|
("id", "INTEGER"),
|
|
("sid", "INTEGER"),
|
|
("sec", "TEXT"),
|
|
("s", "TEXT"),
|
|
("p", "REAL"),
|
|
("q", "REAL"),
|
|
("typ", "TEXT"),
|
|
("tif", "TEXT"),
|
|
("st", "TEXT"),
|
|
("d", "TEXT"),
|
|
]
|
|
return StructuredLogTable(table_name, columns, cur)
|
|
|
|
def modify_order_table(cur: sqlite3.Cursor):
|
|
table_name = "modify_order"
|
|
columns = [
|
|
("ts", "INTEGER"),
|
|
("id", "INTEGER"),
|
|
("sid", "INTEGER"),
|
|
("sec", "TEXT"),
|
|
("s", "TEXT"),
|
|
("p", "REAL"),
|
|
("q", "REAL"),
|
|
("typ", "TEXT"),
|
|
("eid", "TEXT"),
|
|
("st", "TEXT"),
|
|
("d", "TEXT"),
|
|
]
|
|
return StructuredLogTable(table_name, columns, cur)
|
|
|
|
def binance_spots_order_update_table(cur: sqlite3.Cursor):
|
|
table_name = "binance_spots_order_update"
|
|
columns = [
|
|
("ts", "INTEGER"),
|
|
("id", "INTEGER"),
|
|
("sid", "INTEGER"),
|
|
("sec", "TEXT"),
|
|
("s", "TEXT"),
|
|
("p", "REAL"),
|
|
("q", "REAL"),
|
|
("eid", "INTEGER"),
|
|
("rr", "TEXT"),
|
|
("leq", "REAL"),
|
|
("lep", "REAL"),
|
|
("fq", "REAL"),
|
|
("camt", "REAL"),
|
|
("cass", "TEXT"),
|
|
("cw", "INTEGER"),
|
|
("cm", "INTEGER"),
|
|
("ct", "INTEGER"),
|
|
("qatq", "REAL"),
|
|
("lqatq", "REAL"),
|
|
("st", "TEXT"),
|
|
("d", "TEXT"),
|
|
]
|
|
return StructuredLogTable(table_name, columns, cur)
|
|
|
|
tables = {}
|
|
|
|
def add_table(t: StructuredLogTable):
|
|
tables[t.table_name] = t
|
|
|
|
add_table(execution_report_table(cur))
|
|
add_table(create_order_table(cur))
|
|
add_table(modify_order_table(cur))
|
|
add_table(binance_spots_order_update_table(cur))
|
|
return tables
|
|
|
|
|
|
def create_order_timeline(cur: sqlite3.Cursor):
|
|
cols = ["ts", "id", "sid", "s", "p", "q", "st", "d"]
|
|
cols_str = ", ".join(cols)
|
|
expr = f"""
|
|
DROP VIEW IF EXISTS order_timeline;
|
|
DROP TABLE IF EXISTS order_timeline;
|
|
CREATE VIEW IF NOT EXISTS order_timeline AS select action, {cols_str} from (
|
|
select 'create' as action, {cols_str} from create_order UNION
|
|
select 'modify' as action, {cols_str} from modify_order
|
|
)
|
|
ORDER BY ts;
|
|
"""
|
|
expr = f"""
|
|
DROP VIEW IF EXISTS order_timeline;
|
|
DROP TABLE IF EXISTS order_timeline;
|
|
CREATE TEMPORARY VIEW order_timeline_1 AS select action, {cols_str} from (
|
|
select 'create' as action, {cols_str} from create_order UNION
|
|
select 'modify' as action, {cols_str} from modify_order
|
|
);
|
|
CREATE TEMPORARY VIEW order_timeline_2 AS select action, {cols_str} from (
|
|
select * from order_timeline_1 UNION
|
|
select 'execution_report' as action, {cols_str} from execution_report
|
|
);
|
|
CREATE TEMPORARY VIEW order_timeline_3 AS select action, {cols_str} from (
|
|
select * from order_timeline_2 UNION
|
|
select 'binance_spots_update' as action, {cols_str} from binance_spots_order_update
|
|
);
|
|
CREATE TABLE order_timeline AS select * from order_timeline_3 ORDER BY ts;
|
|
"""
|
|
cur.executescript(expr)
|
|
|
|
|
|
def dump_to_db_tables(log_recs: Iterable[LogRecord], db_out: str):
|
|
table_name_re = re.compile("^\[([a-zA-Z_0-9]+).([a-z]+)\](.*)$")
|
|
con = sqlite3.connect(db_out)
|
|
cur = con.cursor()
|
|
lr_chunks = ChunkIterator(log_recs, 256)
|
|
tables = get_structured_log_tables(cur)
|
|
for tname, t in tables.items():
|
|
t.create_table()
|
|
for lr_chunk in lr_chunks:
|
|
for lr in lr_chunk:
|
|
tmatch = table_name_re.match(lr.msg)
|
|
if not tmatch:
|
|
continue
|
|
tname, ttype, smsg = tmatch.group(1, 2, 3)
|
|
if ttype != "json":
|
|
# unsupported type/schema
|
|
continue
|
|
table = tables.get(tname)
|
|
if not table:
|
|
continue
|
|
table.add(smsg)
|
|
for tname, t in tables.items():
|
|
t.flush()
|
|
print(f"Commiting sqlite transactions")
|
|
con.commit()
|
|
|
|
|
|
def _main():
|
|
parser = _get_arg_parser()
|
|
args = parser.parse_args()
|
|
dump_db = args.dump_db or args.dump_db_to or args.dump_db_table_to
|
|
dump_db_slog = args.dump_db_slog
|
|
log_filter = LogFilterCriteria(
|
|
level=args.level,
|
|
category=args.category,
|
|
source_file=args.source_file,
|
|
line_num=args.line_num,
|
|
pattern=args.pattern,
|
|
)
|
|
with open(args.file) as f:
|
|
log_recs = filter(log_filter.include, LogRecordIterator(f))
|
|
if dump_db:
|
|
db_file = args.dump_db_to or f"{args.file}.sqlite"
|
|
if os.path.exists(db_file):
|
|
if not args.yes:
|
|
raise ValueError(f"'{db_file}' already exists")
|
|
os.remove(db_file)
|
|
db_table = args.dump_db_table_to or "log"
|
|
dump_to_db(log_recs, db_file, db_table)
|
|
elif dump_db_slog:
|
|
db_file = f"{args.file}.sqlite"
|
|
if os.path.exists(db_file):
|
|
if not args.yes:
|
|
raise ValueError(f"'{db_file}' already exists")
|
|
os.remove(db_file)
|
|
dump_to_db_tables(log_recs, db_file)
|
|
else:
|
|
n = 0
|
|
for lr in log_recs:
|
|
if not args.count_only:
|
|
print(lr)
|
|
n += 1
|
|
if args.count or args.count_only:
|
|
print(n)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# con = sqlite3.connect("/workspaces/spark/.tmp/logging.sparkd/slog_20250328-1152.sqlite")
|
|
# con = sqlite3.connect("/workspaces/spark/.tmp/logging.sparkd/slog_20250329-1308.sqlite")
|
|
con = sqlite3.connect("/workspaces/spark/.tmp/logging.sparkd/slog_20250329-1340.sqlite")
|
|
cur = con.cursor()
|
|
create_order_timeline(cur)
|
|
con.commit()
|
|
# s = '{ "id": 17004, "sid": 12, "sec": "BTCUSDT", "s": "BUY", "eid": 11601318, "est": CANCELED, "rr": NONE, "leq": 0, "lep": 0, "fq": 0, "camt": 0, "cass": "0", "cw": false, "cm": false, "ct": 1743268003035, "qatq": 0, "lqatq": 0, "qoq": 0, "st": "INVALID_STATE", "d": "exch id old or unresolved (expected for CancelReplace ops)", "ts": 1743268007350708793 }'
|
|
# _main()
|