From 4085cdea27970eed56e227182532957521af2f2f Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 08 2022 12:52:38 +0000 Subject: [PATCH 1/3] Scheduler logger --- diff --git a/docs/schema.sql b/docs/schema.sql index 39fdbf7..982818e 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -955,4 +955,21 @@ CREATE TABLE proton_queue ( ) WITHOUT OIDS; +-- Scheduler tables +CREATE TYPE logger_level AS ENUM ('NOTSET', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'); +CREATE TABLE scheduler_log_messages ( + id SERIAL NOT NULL PRIMARY KEY, + task_id INTEGER REFERENCES task (id), + host_id INTEGER REFERENCES host (id), + msg_time TIMESTAMPTZ NOT NULL DEFAULT NOW(), + logger_name VARCHAR(200) NOT NULL, + level logger_level NOT NULL, + location VARCHAR(200), + msg TEXT NOT NULL +) WITHOUT OIDS; +CREATE INDEX scheduler_log_messages_task_id ON scheduler_log_messages(task_id); +CREATE INDEX scheduler_log_messages_host_id ON scheduler_log_messages(host_id); +CREATE INDEX scheduler_log_messages_msg_time ON scheduler_log_messages(msg_time); +CREATE INDEX scheduler_log_messages_level ON scheduler_log_messages(level); + COMMIT WORK; diff --git a/kojihub/kojixmlrpc.py b/kojihub/kojixmlrpc.py index 1881a88..c194741 100644 --- a/kojihub/kojixmlrpc.py +++ b/kojihub/kojixmlrpc.py @@ -42,6 +42,8 @@ from koji.context import context from koji.server import ServerError, BadRequest, RequestTimeout from koji.xmlrpcplus import ExtendedMarshaller, Fault, dumps, getparser +from . import scheduler + class Marshaller(ExtendedMarshaller): diff --git a/kojihub/scheduler.py b/kojihub/scheduler.py new file mode 100644 index 0000000..b0b7b3f --- /dev/null +++ b/kojihub/scheduler.py @@ -0,0 +1,42 @@ +import functools +import logging + +from koji.db import InsertProcessor + + +class DBLogger(object): + """DBLogger class for encapsulating scheduler logging. It is thread-safe + as both logging parts do this per se (loggind + DB handler via context)""" + + def __init__(self, logger_name=None): + if logger_name: + self.logger = logger_name + else: + self.logger = 'koji.scheduler' + + def log(self, msg, logger_name=None, level=logging.NOTSET, + task_id=None, host_id=None, location=None): + if not logger_name: + logger_name = self.logger + # log to regular log + text = f"task: {task_id}, host: {host_id}, location: {location}, message: {msg}" + logging.getLogger(logger_name).log(level, text) + # log to db + insert = InsertProcessor( + 'scheduler_log_messages', + data={ + 'logger_name': logger_name, + 'level': logging._levelToName[level], + 'task_id': task_id, + 'host_id': host_id, + 'location': location, + 'msg': msg, + } + ) + insert.execute() + + debug = functools.partialmethod(log, level=logging.DEBUG) + info = functools.partialmethod(log, level=logging.INFO) + warning = functools.partialmethod(log, level=logging.WARNING) + error = functools.partialmethod(log, level=logging.ERROR) + critical = functools.partialmethod(log, level=logging.CRITICAL) diff --git a/tests/test_hub/test_scheduler.py b/tests/test_hub/test_scheduler.py new file mode 100644 index 0000000..a1359bd --- /dev/null +++ b/tests/test_hub/test_scheduler.py @@ -0,0 +1,67 @@ +import logging +import mock +import unittest + +import scheduler + +IP = scheduler.InsertProcessor + + +class TestDBLogger(unittest.TestCase): + def setUp(self): + self.InsertProcessor = mock.patch('scheduler.InsertProcessor', + side_effect=self.getInsert).start() + self.inserts = [] + + def tearDown(self): + mock.patch.stopall() + + def getInsert(self, *args, **kwargs): + insert = IP(*args, **kwargs) + insert.execute = mock.MagicMock() + self.inserts.append(insert) + return insert + + def test_defaults(self): + logger = scheduler.DBLogger() + self.assertEqual(logger.logger, 'koji.scheduler') + self.assertEqual(len(self.inserts), 0) + + def test_basic(self): + logger = scheduler.DBLogger() + logger.log("text") + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.table, 'scheduler_log_messages') + self.assertEqual(insert.data, { + 'task_id': None, + 'host_id': None, + 'logger_name': 'koji.scheduler', + 'level': 'NOTSET', + 'location': None, + 'text': 'text', + }) + + def test_all(self): + logger = scheduler.DBLogger() + logger.log("text", logger_name="logger_name", level=logging.ERROR, + task_id=123, host_id=456, location="location") + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.data, { + 'task_id': 123, + 'host_id': 456, + 'logger_name': 'logger_name', + 'level': 'ERROR', + 'location': 'location', + 'text': 'text', + }) + + def test_levels(self): + logger = scheduler.DBLogger() + for level in ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'): + m = getattr(logger, level.lower()) + m("") + insert = self.inserts[0] + self.assertEqual(insert.data['level'], level) + self.inserts = [] From 1ec990131fd30118d488a3b3a2050807ad3962bc Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 08 2022 12:52:38 +0000 Subject: [PATCH 2/3] delete old scheduler log messages --- diff --git a/util/koji-sweep-db b/util/koji-sweep-db index 0c54e85..de4a57d 100755 --- a/util/koji-sweep-db +++ b/util/koji-sweep-db @@ -147,6 +147,21 @@ def clean_buildroots(cursor, vacuum, test): cursor.execute("VACUUM ANALYZE buildroot") +def clean_scheduler_logs(cursor, vacuum, test, age): + clauses = f"(msg_time < NOW() - '{age:d} days'::interval)" + if options.verbose: + query = QueryProcessor(tables=["scheduler_log_messages"], + clauses=clauses, + opts={'countOnly': True}) + rows = query.execute() + print(f"Deleting {rows} scheduler log messages") + if not test: + delete = DeleteProcessor(table="scheduler_log_messages", clauses=clauses)\ + delete.execute() + if vacuum: + cursor.execute("VACUUM ANALYZE scheduler_log_messages") + + if __name__ == "__main__": global options parser = OptionParser("%prog cleans koji database") @@ -180,6 +195,9 @@ if __name__ == "__main__": parser.add_option('--scratch-builds-age', type=int, dest="scratch_age", action="store", default=730, metavar="DAYS", help="Delete scratch builds' tasks older than this (default: 2 years") + parser.add_option('--logs-age', type=int, + action="store", default=7, metavar="DAYS", + help="Delete scheduler log messages older than this (default: 7 days)") parser.add_option('--buildroots', action="store_true", help="Delete unreferenced buildroots") parser.add_option('-f', '--force', action="store_true", @@ -240,6 +258,7 @@ if __name__ == "__main__": clean_sessions(cursor, options.vacuum, options.test, options.sessions_age, options.sessions_absolute_age) clean_reservations(cursor, options.vacuum, options.test, options.reservations_age) + clean_scheduler_logs(cursor, options.vacuum, options.test, options.logs_age) if options.tag_notifications: clean_notification_tasks(cursor, options.vacuum, options.test, age=options.tag_notifications_age) From b3ad76096c427968493ba1b92addc9c2dfb5c34b Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 08 2022 12:52:38 +0000 Subject: [PATCH 3/3] basic CLI interface for scheduler logs --- diff --git a/cli/koji_cli/commands.py b/cli/koji_cli/commands.py index 0100616..81483e9 100644 --- a/cli/koji_cli/commands.py +++ b/cli/koji_cli/commands.py @@ -7737,3 +7737,62 @@ def anon_handle_userinfo(goptions, session, args): print("Number of tasks: %d" % tasks.result) print("Number of builds: %d" % builds.result) print('') + + +def handle_scheduler_logs(goptions, session, args): + "[monitor] Query scheduler logs" + usage = "usage: %prog scheduler-logs " + parser = OptionParser(usage=get_usage_str(usage)) + parser.add_option("--task", type="int", action="store", + help="Filter by task ID") + parser.add_option("--host", type="str", action="store", + help="Filter by host (name/ID)") + parser.add_option("--level", type="str", action="store", + choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], + help="Filter by message level") + parser.add_option("--from", type="float", action="store", dest="from_ts", + help="Logs from given timestamp") + parser.add_option("--to", type="float", action="store", dest="to_ts", + help="Logs until given timestamp (included)") + parser.add_option("--logger", type="str", action="store", + help="Filter by logger name") + (options, args) = parser.parse_args(args) + if len(args) != 0: + parser.error("There are no arguments for this command") + + kwargs = {} + if options.task: + kwargs['taskID'] = options.task + if options.host: + try: + kwargs['hostID'] = int(options.host) + except ValueError: + kwargs['hostID'] = session.getHost(options.host)['id'] + if options.level: + kwargs['level'] = options.level + if options.from_ts: + kwargs['from_ts'] = options.from_ts + if options.to_ts: + kwargs['to_ts'] = options.to_ts + if options.logger: + kwargs['logger_name'] = options.logger + + logs = session.scheduler.getLogs(**kwargs) + + mask = ("%(task_id)s\t%(host_name)s\t%(msg_time)s\t%(logger_name)s" + "\t%(level)s\t%(location)s\t%(msg)s") + if not goptions.quiet: + h = mask % { + 'task_id': 'Task', + 'host_name': 'Host', + 'msg_time': 'Time', + 'logger_name': 'Logger', + 'level': 'Level', + 'location': 'Location', + 'msg': 'Message', + } + print(h) + print('-' * len(h)) + + for log in logs: + print(mask % log) diff --git a/kojihub/kojixmlrpc.py b/kojihub/kojixmlrpc.py index c194741..eb4ee58 100644 --- a/kojihub/kojixmlrpc.py +++ b/kojihub/kojixmlrpc.py @@ -847,8 +847,10 @@ def get_registry(opts, plugins): registry = HandlerRegistry() functions = kojihub.RootExports() hostFunctions = kojihub.HostExports() + schedulerFunctions = scheduler.SchedulerExports() registry.register_instance(functions) registry.register_module(hostFunctions, "host") + registry.register_module(schedulerFunctions, "scheduler") registry.register_function(koji.auth.login) registry.register_function(koji.auth.sslLogin) registry.register_function(koji.auth.logout) diff --git a/kojihub/scheduler.py b/kojihub/scheduler.py index b0b7b3f..eebb6d7 100644 --- a/kojihub/scheduler.py +++ b/kojihub/scheduler.py @@ -1,7 +1,61 @@ import functools import logging -from koji.db import InsertProcessor +from koji.db import InsertProcessor, QueryProcessor + + +class SchedulerExports(): + def getLogs(self, taskID=None, hostID=None, level=None, + from_ts=None, to_ts=None, logger_name=None): + """Return all related log messages + + :param int taskID: filter by task + :param int hostID: filter by host + :param str level: filter by message level + :param float from_ts: filter from earliest time + :param float to_ts: filter to latest time (from_ts < ts <= to_ts) + :param str logger_name: filter by logger name + :return [dict]: list of messages + """ + fields = ( + ('scheduler_log_messages.id', 'id'), + ('task_id', 'task_id'), + ('host_id', 'host_id'), + ('msg_time', 'msg_time'), + ('logger_name', 'logger_name'), + ('level', 'level'), + ('location', 'location'), + ('msg', 'msg'), + ('hosts.name', 'host_name'), + ) + clauses = [] + values = {} + if taskID is not None: + clauses.append("taskID = %(taskID)") + values['taskID'] = taskID + if hostID is not None: + clauses.append("hostID = %(hostID)") + values['hostID'] = hostID + if level is not None: + clauses.append("level = %(level)s") + values['level'] = level.upper() + if from_ts is not None: + clauses.append("msg_time > %(from_ts)s") + values['from_ts'] = float(from_ts) + if to_ts is not None: + clauses.append("msg_time <= %(to_ts)s") + values['to_ts'] = float(to_ts) + if logger_name is not None: + clauses.append("logger_name = %(to_ts)s") + values['logger_name'] = logger_name + + columns, aliases = zip(*fields) + query = QueryProcessor(tables=['scheduler_log_messages'], + columns=columns, aliases=aliases, + joins=['hosts ON host_id = hosts.id'], + clauses=clauses, values=values, + opts={'order': 'msg_time'}) + return query.execute() class DBLogger(object):