From 8c6dc40519ba9d637a3390e0b8005dd44cb66b9a Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Feb 29 2024 15:06:05 +0000 Subject: wip Related: https://pagure.io/koji/issue/3176 --- diff --git a/kojihub/kojixmlrpc.py b/kojihub/kojixmlrpc.py index 025200b..c4c6d21 100644 --- a/kojihub/kojixmlrpc.py +++ b/kojihub/kojixmlrpc.py @@ -41,6 +41,7 @@ from koji.server import ServerError, BadRequest, RequestTimeout from koji.xmlrpcplus import ExtendedMarshaller, Fault, dumps, getparser from . import auth from . import db +from . import repos from . import scheduler @@ -866,9 +867,11 @@ def get_registry(opts, plugins): functions = kojihub.RootExports() hostFunctions = kojihub.HostExports() schedulerFunctions = scheduler.SchedulerExports() + reposFunctions = repos.ReposExports() registry.register_instance(functions) registry.register_module(hostFunctions, "host") registry.register_module(schedulerFunctions, "scheduler") + registry.register_module(reposFunctions, "repos") registry.register_function(auth.login) registry.register_function(auth.sslLogin) registry.register_function(auth.logout) diff --git a/kojihub/repos.py b/kojihub/repos.py new file mode 100644 index 0000000..463a8c1 --- /dev/null +++ b/kojihub/repos.py @@ -0,0 +1,357 @@ +import datetime +import koji +from koji.context import context +from .db import ( + DeleteProcessor, + InsertProcessor, + QueryProcessor, + UpdateProcessor, + _dml, +) +from .kojihub import ( + make_task, + get_tag, + RootExports, +) + +''' +Repo regen queue +================ + +Repo regen queue is now in the database. Simple ordering by priority sorts it. + +Tag can get into queue by: +1) kojira detecting that repo is no longer valid (priority 100) +2) By user requesting it via build with fresh repo (priority 90) +3) New target creation should add it to the queue (priority 80) +4) buildroot inheritance change (priority 100) + +Tag can leave the queue: +1) tag is deleted (ON CASCADE?) +2) newRepo is started + +Tag can be prioritized by tags.extra['repo_regen_priority'] + +Storing: + * tag + * last computed score + time of computation + * needed since + * expired since + * priority (static offset) + * awaited + * maven_support + + Build can: + 1) run as normal (use latest repo) + 2) --wait-repo + - if expired_event >= requested + - newRepo is running, wait + - expired_event < requested + if newRepo is running, queue new one + + +Random ideas: + - request_repo_regen can consult policy based on tag/user and set weight/priority + - add start requests to the queue (this repo should be regenerated in two hours + earliest if nobody will request it earlier + - tag option - regenerate always, daily, never if not requested + +repo_regen_method = + has req_regen_method :: req_regen_method + method runroot :: never + target match rhel-9.0.0-build :: always + target match rhel-6.8.z-build :: 1d + target match eng-fedora-34-* && bool has_external_repos :: check + all :: none + +Actions: always, 12h, 3d, check, never/manual + +repo_regen_priority = + perm match repo && has req_priority :: req_priority # via manual newRepo call + perm match kpatch :: adjust +1 + all :: heuristics +''' + + +def request_repo_regen(tag_id, priority=None, awaited=False): + query = QueryProcessor(tables=['repo_queue'], + columns=['tag_id', 'priority', 'awaited'], + clauses=['tag_id = %(tag_id)s'], + values={'tag_id': tag_id}, + opts={'rowlock': True}) + queued = query.executeOne() + if queued: + data = {} + if priority and priority < queued['priority']: + # better priority now + data['priority'] = priority + if awaited and not queued['awaited']: + data['awaited'] = True + + if data: + update = UpdateProcessor(table='repo_queue', + clauses=['tag_id = %(tag_id)s'], + values=queued) + data['updated_ts'] = datetime.datetime.now() + update.set(**data) + update.execute() + else: + tag = get_tag(tag_id, strict=True) + data = { + 'tag_id': tag_id, + 'maven_support': tag['maven_support'], + } + if awaited: + data['awaited'] = awaited + # TODO: improve getRepo import (put it out of RootExports) + repo = RootExports().getRepo(tag_id) + if repo: + data['expired_event'] = repo['create_event'] + data['expired_ts'] = datetime.datetime.fromtimestamp(repo['create_ts']) + if priority: + context.session.assertPerm('regen-repo') + data['priority'] = priority + insert = InsertProcessor(table='repo_queue', data=data) + insert.execute() + + +def set_repo_regen_priority(tag_id, priority): + context.session.assertPerm('regen-repo') + query = QueryProcessor( + tables=['repo_regen'], + columns=['priority'], + clauses=['tag_id = %(tag_id)s'], + values={'tag_id': tag_id}, + ) + queued = query.executeOne() + if queued and queued['priority'] != priority: + update = UpdateProcessor( + 'repo_queue', + clauses=['tag_id = %(tag_id)s'], + values={'tag_id': tag_id}, + data={'priority': priority}, + ) + update.execute() + + +def _get_top_repo(maven=None): + # pop from the queue + clauses = [] + values = {} + if maven is not None: + clauses.append('maven_support = %(maven)s') + values = {'maven': bool(maven)} + query = QueryProcessor(tables=['repo_queue'], + columns=['tag_id', 'priority', 'score'], + clauses=clauses, + values=values, + opts={ + 'order': 'priority,-score', + 'limit': 1, + 'rowlock': True, + }) + row = query.executeOne() + if not row: + return None + + remove_from_queue(row['tag_id']) + return row['tag_id'] + + +def remove_from_queue(tag_id): + """Delete from queue in case tag + - is no longer expired (task was scheduled) + - doesn't exist anymore or + - is no more a buildtag + """ + context.session.assertPerm('regen-repo') + delete = DeleteProcessor('repo_queue', + clauses=['tag_id = %(tag_id)s'], + values={'tag_id': tag_id}) + delete.execute() + + +def get_repo_queue(): + # TODO, placeholder (CLI/web) + columns = [ + 'tag_id', + 'score', + 'priority', + 'expired_event', + 'expired_ts', + 'updated_ts', + 'weight', + 'awaited', + #'max_n', + 'maven_support', + ] + query = QueryProcessor(tables=['repo_queue'], + columns=columns, + opts={'order': 'priority,-score'}) + return query.execute() + + +def update_score(): + """Set score for whole queue + + We score the tags by two factors + - age of current repo + - last use in a buildroot + + Having an older repo or a higher use count gives the tag a higher + priority for regen. The formula attempts to keep the last use factor + from overpowering, so that tags with very old repos still get priority + + Updating could happen periodically every few minutes, triggered by kojira + """ + + query = ''' + WITH subquery AS ( + WITH + queued AS ( + SELECT + tag_id, + CASE awaited + WHEN TRUE THEN 2.0 + ELSE 1.0 + END awaited, + EXTRACT(epoch FROM (NOW() - expired_ts)) AS age + FROM repo_queue), + n_recent_values AS ( + SELECT COUNT(*) AS n_recent, repo.tag_id AS tag_id + FROM buildroot + LEFT OUTER JOIN standard_buildroot ON standard_buildroot.buildroot_id = buildroot.id + LEFT OUTER JOIN repo ON repo.id = standard_buildroot.repo_id + LEFT OUTER JOIN events ON events.id = standard_buildroot.create_event + WHERE + events.time > NOW() - '1 year'::interval AND + repo.tag_id IN (SELECT tag_id FROM repo_queue) + GROUP BY repo.tag_id), + max_n AS ( + SELECT + CASE MAX(max_n) + WHEN 0 THEN 1 + ELSE MAX(max_n) + END max_n + FROM repo_queue + ) + SELECT + queued.tag_id, + CASE + WHEN n_recent IS NULL THEN age * awaited + ELSE age * awaited * (CAST(n_recent * 9.0 AS INTEGER) / max_n.max_n + 1) + END score + FROM queued + LEFT JOIN n_recent_values ON n_recent_values.tag_id = queued.tag_id + JOIN max_n ON TRUE + ) + UPDATE repo_queue + SET score = score_table.score + FROM (SELECT * FROM subquery) AS score_table + WHERE score_table.tag_id = repo_queue.tag_id + ''' + # TODO - maybe better as stored procedure? + return _dml(query, {}) + + +''' +def __update_score_original(tag_id): + """Set score for needed_tag entry + + We score the tags by two factors + - age of current repo + - last use in a buildroot + + Having an older repo or a higher use count gives the tag a higher + priority for regen. The formula attempts to keep the last use factor + from overpowering, so that tags with very old repos still get priority + + Updating could happen periodically every few minutes, triggered by kojira + """ + # get queue item + query = QueryProcessor(tables=['repo_queue'], + columns=['awaited', 'EXTRACT(epoch FROM (NOW() - expired_ts))'], + aliases=['awaited', 'age'], + clauses=['tag_id = %(tag_id)s'], + values={'tag_id': tag_id}) + queued = query.executeOne() + if not queued: + return None + + # get recent uses in last day + query = QueryProcessor( + tables=['buildroot'], + joins=[ + 'LEFT OUTER JOIN standard_buildroot ' + 'ON standard_buildroot.buildroot_id = buildroot.id', + 'LEFT OUTER JOIN repo ' + 'ON repo.id = standard_buildroot.repo_id', + 'LEFT OUTER JOIN events ' + 'ON events.id = standard_buildroot.create_event'], + clauses=[ + 'repo.tag_id = %(tag_id)s', + "events.time > NOW() - '1 day'::interval", + ], + values={'tag_id': tag_id}, + opts={'countOnly': True}) + n_recent = query.executeOne() + + # SELECT max(n_recent) FROM repo_queue + query = QueryProcessor(tables=['repo_queue'], + columns=['MAX(max_n)'], + aliases=['max']) + max_n = query.executeOne()['max'] or 1 + + adj = n_recent * 9.0 // max_n + 1 + age = queued['age'] + # XXX - need to make sure our times aren't far off, otherwise this + # scoring could have the opposite of the desired effect + if age < 0: + age = 0 + if queued['awaited']: + score_adjust = 2.0 + else: + score_adjust = 1.0 + score = age * adj * score_adjust + import logging + logging.error(f"SCORE: {score}: age {age} adj {adj} score_adjust {score_adjust}") + # so a day old unused repo gets about the regen same score as a + # 2.4-hour-old, very popular repo + + # update queue + update = UpdateProcessor( + table='repo_queue', + clauses=['tag_id = %(tag_id)s'], + values={'tag_id': tag_id}, + data={'score': score}, + rawdata={'updated_ts': 'NOW()'} + ) + update.execute() + return score +''' + + +def start_newrepo_from_queue(task_priority=15, maven=None): + """ + TODO: config rate-limiting (probably better in kojira?) + could be done by non/maven task limits as arguments + """ + # TODO - some kojira-only permission instead + context.session.assertPerm('admin') + while True: + tag_id = _get_top_repo() + if not tag_id: + return + if not get_tag(tag_id, strict=False): + # deleted tag + continue + args = koji.encode_args(tag_id) + return make_task('newRepo', args, priority=task_priority, channel='createrepo') + + +class ReposExports(): + request = staticmethod(request_repo_regen) + setPriority = staticmethod(set_repo_regen_priority) + removeFromQueue = staticmethod(remove_from_queue) + startNewRepoFromQueue = staticmethod(start_newrepo_from_queue) + updateScore = staticmethod(update_score) diff --git a/schemas/schema.sql b/schemas/schema.sql index e5f3462..2ff8943 100644 --- a/schemas/schema.sql +++ b/schemas/schema.sql @@ -1052,4 +1052,21 @@ CREATE TABLE locks ( INSERT INTO locks(name) VALUES('protonmsg-plugin'); INSERT INTO locks(name) VALUES('scheduler'); +-- repo queue +CREATE TABLE repo_queue ( + tag_id INTEGER NOT NULL REFERENCES tag (id) PRIMARY KEY, + score FLOAT DEFAULT 0.0, + priority INTEGER DEFAULT 100, + --needed_since TIMESTAMPTZ NOT NULL DEFAULT NOW(), + expired_event INTEGER REFERENCES events(id), + expired_ts TIMESTAMPTZ, + updated_ts TIMESTAMPTZ NOT NULL DEFAULT NOW(), + weight FLOAT DEFAULT 1.0 CHECK (NOT weight <= 0), + awaited BOOL DEFAULT FALSE, + --max_n INTEGER NOT NULL DEFAULT 0, + maven_support BOOLEAN NOT NULL DEFAULT FALSE +); +CREATE INDEX repo_queue_score_priority ON repo_queue(priority, score DESC); +CREATE INDEX repo_queue_tag_id ON repo_queue(tag_id); + COMMIT WORK;