+ import subprocess

+ from .helpers import (get_redis_logger, silent_remove, ensure_dir_exists,

+                       get_chroot_arch, cmd_debug, format_filename,

+                       uses_devel_repo)

+ from backend.worker_manager import WorkerManager

+     def __init__(self, opts, action, log=None):

+         self.log = log if log else get_redis_logger(self.opts, "backend.actions", "actions")

+                     self.log.info("Empty repo so far, creating the directory")

+                     createrepo(path=path, username=ownername,

+                                projectname=projectname)

+                 createrepo(path=chroot_path, username=data["user"],

+                            projectname=data["copr"])

+         devel = uses_devel_repo(self.front_url, ownername,

+                                 projectname)

+ 

+                 createrepo(path=chroot_path, base_url=result_base_url,

+                            username=ownername, projectname=projectname,

+                            devel=devel)

+                 self.log.exception("Error making local repo: %s", chroot_path)

+             createrepo(path=chrootdir, username=data["ownername"],

+                        projectname=data["projectname"])

+                     createrepo(path=destdir, username=ownername,

+                                projectname=projectname)

+         # TODO: we don't need Munch() here, drop it

+         return result

+ 

+ 

+ class ActionQueueTask():

+     def __init__(self, id):

+         self.id = id

+     def __repr__(self):

+         return str(self.id)

+ 

+ 

+ class ActionWorkerManager(WorkerManager):

+     frontend_client = None

+     worker_prefix = 'action_worker'

+ 

+     def start_task(self, worker_id, task):

+         command = [

+             'copr-backend-process-action',

+             '--daemon',

+             '--task-id', repr(task),

+             '--worker-id', worker_id,

+         ]

+         # TODO: mark as started on FE, and let user know in UI

+         subprocess.check_call(command)

+ 

+     def has_worker_ended(self, worker_id, task_info):

+         return 'status' in task_info

+ 

+     def finish_task(self, worker_id, task_info):

+         task_id = self.get_task_id_from_worker_id(worker_id)

+ 

+         result = Munch()

+         result.id = int(task_id)

+         result.result = int(task_info['status'])

+         result.job_ended_on = time.time()

+ 

+         try:

+             self.frontend_client.update({"actions": [result]})

+         except RequestException:

+             self.log.exception("can't post to frontend, retrying indefinitely")

+             return False

+         return True

+ 

+     def is_worker_alive(self, worker_id, task_info):

+         if not 'PID' in task_info:

+             return False

+         pid = int(task_info['PID'])

+         try:

+             # Send signal=0 to the process to check whether it still exists.

+             # This is just no-op if the signal was successfully delivered to

+             # existing process, otherwise exception is raised.

+             os.kill(pid, 0)

+         except OSError:

+             return False

+         return True

+ def createrepo(path, username, projectname, devel=False, base_url=None):

+     Creates repodata.  Depending on the "auto_createrepo" parameter it either

+     creates the repodata directory in `path`, or in `path/devel`.

+     :param devel: create the repository in 'devel' subdirectory

+     if not devel:

+ 

+     # Automatic createrepo disabled.  Even so, we still need to createrepo in

+     # special "devel" directory so we can later build packages against it.

+ from ..actions import ActionWorkerManager, ActionQueueTask

+ from ..helpers import get_redis_logger, get_redis_connection

+     def get_frontend_actions(self):

+         Get unfiltered list of actions from frontend, both running and pending.

+ 

+         try:

+             url = "{0}/backend/pending-actions/".format(self.opts.frontend_base_url)

+             request = get(url, auth=("user", self.opts.frontend_auth))

+             raw_actions = request.json()

+         except (RequestException, ValueError) as error:

+             self.log.exception(

+                 "Retrieving an action tasks failed with error: %s",

+                 error)

+             return []

+ 

+         return [ActionQueueTask(action['id']) for action in raw_actions]

+ 

+         redis = get_redis_connection(self.opts)

+         worker_manager = ActionWorkerManager(

+             redis_connection=redis,

+             log=self.log,

+             max_workers=self.opts.actions_max_workers)

+         worker_manager.frontend_client = FrontendClient(self.opts, self.log)

+ 

+         timeout = self.opts.sleeptime

+ 

+             self.log.info("getting actions from frontend")

+             start = time.time()

+             for task in self.get_frontend_actions():

+                 worker_manager.add_task(task)

+ 

+             # Execute the actions.

+             worker_manager.run(timeout=timeout)

+ 

+             sleep_more = timeout - (time.time() - start)

+             if sleep_more > 0:

+                 time.sleep(sleep_more)

+ from requests import post, get, RequestException

+ RETRY_TIMEOUT = 5

+     def get_reliably(self, url_path):

+         """

+         Get the URL response from frontend, try indefinitely till the server

+         gives us answer.

+         """

+         url = "{}/{}/".format(self.frontend_url, url_path)

+         auth = ("user", self.frontend_auth)

+ 

+         attempt = 0

+         while True:

+             attempt += 1

+             try:

+                 response = get(url, auth=auth)

+             except RequestException as ex:

+                 self.msg = "Get request {} failed: {}".format(attempt, ex)

+                 time.sleep(RETRY_TIMEOUT)

+                 continue

+ 

+             return response

+ 

+ 

+         opts.actions_max_workers = _get_conf(

+             cp, "builder", "actions_max_workers",

+             default=10, mode="int")

+ 

+ def uses_devel_repo(front_url, username, projectname):

+         return not bool(result.data["detail"]["auto_createrepo"])

+     return False

+ from ..helpers import uses_devel_repo

+         project_owner = self.job.project_owner

+         project_name = self.job.project_name

+ 

+         base_url = "/".join([self.opts.results_baseurl, project_owner,

+                              project_name, self.job.chroot])

+                       .format(project_owner, project_name,

+         devel = uses_devel_repo(self.opts.frontend_base_url,

+                                 project_owner, project_name)

+                 username=project_owner,

+                 projectname=project_name,

+                 devel=devel,

+ import time

+ from heapq import heappop, heappush

+ import itertools

+ import redis

+ import logging

+ 

+ class JobQueue():

+     """

+     Priority "task" queue for WorkerManager.  Taken from:

+     https://docs.python.org/3/library/heapq.html#priority-queue-implementation-notes

+     The higher the 'priority' is, the later the task is taken.

+     """

+ 

+     def __init__(self, removed='<removed-task>'):

+         self.prio_queue = []             # list of entries arranged in a heap

+         self.entry_finder = {}           # mapping of tasks to entries

+         self.removed = removed           # placeholder for a removed task

+         self.counter = itertools.count() # unique sequence count

+ 

+     def add_task(self, task, priority=0):

+         'Add a new task or update the priority of an existing task'

+         if repr(task) in self.entry_finder:

+             self.remove_task(task)

+         count = next(self.counter)

+         entry = [priority, count, task]

+         self.entry_finder[repr(task)] = entry

+         heappush(self.prio_queue, entry)

+ 

+     def remove_task(self, task):

+         'Mark an existing task as removed.  Raise KeyError if not found.'

+         entry = self.entry_finder.pop(repr(task))

+         entry[-1] = self.removed

+ 

+     def pop_task(self):

+         'Remove and return the lowest priority task. Raise KeyError if empty.'

+         while self.prio_queue:

+             priority, count, task = heappop(self.prio_queue)

+             if task is not self.removed:

+                 del self.entry_finder[repr(task)]

+                 return task

+         raise KeyError('pop from an empty priority queue')

+ 

+ 

+ class WorkerManager():

+     """

+     Automatically process 'self.tasks' priority queue, and start background jobs

+     to handle them.

+ 

+     :cvar worker_prefix: Unique string across all the WorkerManager child

+             classes, this is used as prefix for the workers in redis database

+             and to easily determine to which WorkerManager the particlar worker

+             belongs to.  So it can be anything reasonable, just make sure it is

+             unique.

+     :cvar worker_timeout_start: The time period we give the background process

+             successfully start and identify itself (see has_worker_started()

+             method).  If the background process isn't indentified after this

+             timeout, we drop it from database and consider it failed.  And the

+             task is re-scheduled.  Float value in seconds.

+     :cvar worker_timeout_deadcheck: If the worker successfully identified itself

+             after start (the has_worker_started() returns True) we know that the

+             worker process at least started.  But after worker_timeout_deadcheck

+             timeout we'll also keep an eye on the process by asking the

+             is_worker_alive() method - whether the task is really still doing

+             something on background or not (== unexpected failure cleanup).

+             Fill float value in seconds.

+     """

+     worker_prefix = 'worker' # make sure this is unique in each class

+     worker_timeout_start = 10

+     worker_timeout_deadcheck = 60

+ 

+     def __init__(self, redis_connection=None, max_workers=8, log=None):

+         self.tasks = JobQueue()

+         self.log = log if log else logging.getLogger()

+         self.redis = redis_connection

+         self.max_workers = max_workers

+ 

+     def start_task(self, worker_id, task):

+         """

+         Start background job using the 'task' object taken from the 'tasks'

+         queue.  The background task should _on its own_ and ASAP let the manager

+         know that it successfully started (e.g. mark the job 'started' in redis

+         DB), so the has_worker_started() method later gives us valid info.

+         """

+         raise NotImplementedError

+ 

+     def finish_task(self, worker_id, task):

+         """

+         This is called once the worker manager consider the task to be done,

+         because the `has_worker_ended()` method already returns True.  Override

+         this function and use it to let Frontend know that the task finished,

+         and how (whether it succeeded, etc.).

+         """

+         raise NotImplementedError

+ 

+     def has_worker_started(self, worker_id, task_info):

+         """

+         The background task process should somehow notify manager that it

+         already started (so we can have has_worker_started() implemented).

+         By default we expect it sets 'started' attribute in redis DB, but feel

+         free to override this method and invent different notification

+         mechanism.

+         """

+         return 'started' in task_info

+ 

+     def has_worker_ended(self, worker_id, task_info):

+         """

+         Check 'task_info' (dictionary output from redis) whether the task is

+         already finished by worker.  If yes, do whatever is needed with the

+         result (contact frontend) and return True.  If the task is still

+         processed, return False.

+         """

+         raise NotImplementedError

+ 

+     def is_worker_alive(self, worker_id, task_info):

+         """

+         Check staled jobs on background, whether they haven't died before they

+         notified us about the status.  We'll keep asking after

+         worker_timeout_deadcheck seconds left since we tried to spawn the

+         worker.

+         """

+         raise NotImplementedError

+ 

+     def get_worker_id(self, task_id):

+         """

+         Given the unique task representation form (usually ID), generate worker

+         identificator (redis key).

+         """

+         return '{}:{}'.format(self.worker_prefix, task_id)

+ 

+     def get_task_id_from_worker_id(self, worker_id):

+         """

+         Given the unique task representation form (usually ID), generate worker

+         identificator (redis key).

+         """

+         prefix, task_id = worker_id.rsplit(':', 1)

+         assert prefix == self.worker_prefix

+         return task_id

+ 

+     def has_worker(self, task_id):

+         worker_id = self.get_worker_id(task_id)

+         return worker_id in self.worker_ids()

+ 

+     def add_task(self, task):

+         task_id = repr(task)

+         if self.has_worker(task_id):

+             # No need to re-add this to queue.

+             self.log.warning("Task %s has worker, skipped", task_id)

+             return

+ 

+         self.log.info("Adding task %s to queue", task_id)

+         self.tasks.add_task(task)

+ 

+     def worker_ids(self):

+         """

+         Return the redis keys representing workers running on background.

+         """

+         return self.redis.keys(self.worker_prefix + ':*')

+ 

+     def run(self, timeout=float('inf')):

+         """

+         Process the task (priority) queue.

+         """

+         start_time = time.time()

+ 

+         while time.time() - start_time < timeout:

+             self._cleanup_workers()

+ 

+             worker_count = len(self.worker_ids())

+             if worker_count >= self.max_workers:

+                 time.sleep(1)

+                 continue

+ 

+             # We can allocate some workers, if there's something to do.

+             try:

+                 task = self.tasks.pop_task()

+             except KeyError:

+                 # Empty queue!

+                 if worker_count:

+                     # It still makes sense to cycle to finish the workers.

+                     time.sleep(1)

+                     continue

+                 # Optimization part, nobody is working now, and there's nothing

+                 # to do.  Just simply wait till the end of the cycle.

+                 break

+ 

+             self._start_worker(task)

+ 

+     def _start_worker(self, task):

+         worker_id = self.get_worker_id(repr(task))

+         self.redis.hset(worker_id, 'allocated', time.time())

+         self.log.info("Starting worker %s", worker_id)

+         self.start_task(worker_id, task)

+ 

+     def clean_tasks(self):

+         'remove all tasks from queue'

+         self.tasks = JobQueue()

+ 

+     def _cleanup_workers(self):

+         now = time.time()

+ 

+         for worker_id in self.worker_ids():

+             info = self.redis.hgetall(worker_id)

+             allocated = float(info.get('allocated'))

+ 

+             if self.has_worker_ended(worker_id, info):

+                 # finished worker

+                 self.log.info("Finished worker %s", worker_id)

+                 self.finish_task(worker_id, info)

+                 self.redis.delete(worker_id)

+                 continue

+ 

+             if info.get('delete'):

+                 self.log.warning("worker %s deleted", worker_id)

+                 self.redis.delete(worker_id)

+                 continue

+ 

+             if not self.has_worker_started(worker_id, info):

+                 if now - allocated > self.worker_timeout_start:

+                     # This worker failed to start?

+                     self.log.error("worker %s failed to start", worker_id)

+                     self.redis.delete(worker_id)

+                 continue

+ 

+             checked = info.get('checked', allocated)

+ 

+             if now - float(checked) > self.worker_timeout_deadcheck:

+                 self.log.info("checking worker %s", worker_id)

+                 self.redis.hset(worker_id, 'checked', now)

+                 if self.is_worker_alive(worker_id, info):

+                     continue

+                 self.log.error("dead worker %s", worker_id)

+ 

+                 # The worker could finish in the meantime, make sure we

+                 # hgetall() once more.

+                 self.redis.hset(worker_id, 'delete', 1)

+ # How many workers (background processes) can Backend spawn for handling actions

+ # concurrently.

+ #actions_max_workers=8

+ 

+ #! /usr/bin/python3

+ 

+ import os

+ import sys

+ import time

+ import daemon

+ import argparse

+ import logging

+ import contextlib

+ 

+ sys.path.append("/usr/share/copr/")

+ 

+ from backend.helpers import (BackendConfigReader, get_redis_logger,

+                              get_redis_connection)

+ from backend.frontend import FrontendClient

+ from backend.actions import Action, ActionResult

+ 

+ 

+ def get_arg_parser():

+     'return argument parser object'

+ 

+     parser = argparse.ArgumentParser(

+         description="Process single copr action",

+     )

+     parser.add_argument(

+         "--task-id",

+         type=int,

+         required=True,

+         help="Task ID to process",

+     )

+     parser.add_argument(

+         "--worker-id",

+         help="Worker ID already exists in DB (used by WorkerManager only)",

+     )

+     parser.add_argument(

+         "--daemon",

+         action='store_true',

+         help="run on background, as daemon process"

+     )

+     return parser

+ 

+ 

+ def handle_task(opts, args, log):

+     "Handle the task, executed on background in DaemonContext"

+ 

+     task_id = args.task_id

+ 

+     frontend_client = FrontendClient(opts, log)

+     redis = get_redis_connection(opts)

+ 

+     log.info("Handling action %s", task_id)

+ 

+     if args.worker_id:

+         # Identify ourselves.

+         redis.hset(args.worker_id, 'started', 1)

+         redis.hset(args.worker_id, 'PID', os.getpid())

+ 

+     resp = frontend_client.get_reliably('action/{}'.format(task_id))

+     if resp.status_code != 200:

+         log.error("failed to download task, apache code %s", resp.status_code)

+         sys.exit(1)

+ 

+     action_task = resp.json()

+     action = Action(opts, action_task, log=log)

+     result = ActionResult.FAILURE

+     try:

+         action_result = action.run()

+         result = action_result.result

+     except Exception:

+         log.exception("action failed for unknown error")

+ 

+     log.info("Action %s ended with status=%s", action_task, result)

+ 

+     # Let the manager know what's the result.

+     if args.worker_id:

+         redis.hset(args.worker_id, 'status', str(result))

+ 

+ 

+ def main():

+     'handle the task, the main function'

+ 

+     if os.getuid() == 0:

+         sys.stderr.write("this needs to be run as 'copr' user\n")

+         sys.exit(1)

+ 

+     config = '/etc/copr/copr-be.conf'

+     opts = BackendConfigReader(config).read()

+     args = get_arg_parser().parse_args()

+ 

+     context = contextlib.nullcontext()

+     if args.daemon:

+         context = daemon.DaemonContext()

+ 

+     with context:

+         logger_name = '{}.{}.pid-{}'.format(

+             sys.argv[0],

+             'managed' if args.worker_id else 'manual',

+             os.getpid(),

+         )

+         log = get_redis_logger(opts, logger_name, "actions")

+         try:

+             if not args.daemon:

+                 # when executing from commandline - on foreground - we want to print

+                 # something to stderr as well

+                 log.addHandler(logging.StreamHandler())

+             handle_task(opts, args, log)

+         except Exception as exc: # pylint: disable=W0703

+             log.exception("unexpected failure %s", str(exc))

+ 

+ 

+ if __name__ == "__main__":

+     main()

+ from backend.helpers import uses_devel_repo, get_persistent_status, get_auto_prune_status

+             if uses_devel_repo(self.opts.frontend_base_url, username, projectname):