buildservice: Move the old stuff into the new wrapper

[people/ms/pakfire.git] / src / pakfire / daemon.py

#!/usr/bin/python3

import asyncio
import functools
import glob
import io
import json
import logging
import logging.handlers
import multiprocessing
import os.path
import setproctitle
import signal
import socket
import tempfile

from . import _pakfire
from . import buildservice
from . import config
from . import logger

from pakfire.constants import *
from pakfire.i18n import _

# Setup logging
log = logging.getLogger("pakfire.daemon")

class Daemon(object):
	def __init__(self, config_file="daemon.conf", debug=False, verbose=False):
		self.config  = config.Config(config_file)
		self.debug   = debug
		self.verbose = verbose

		# Setup logger
		self.log = logger.setup(
			"pakfire",
			syslog_identifier="pakfire-daemon",
			enable_console=self.verbose,
			debug=self.debug,
		)

		# Initialize the connection to the buildservice
		self.service = buildservice.BuildService()

		# Set when this process receives a shutdown signal
		self._shutdown_signalled = None

		# List of worker processes.
		self.workers = []

		# Stats Connection
		self.stats = None

	@property
	def ccache_path(self):
		"""
			Returns the ccache path
		"""
		return self.config.get("daemon", "ccache_path", "/var/cache/pakfire/ccache")

	async def run(self):
		"""
			Main loop.
		"""
		# Register signal handlers.
		self.register_signal_handlers()

		# Initialize shutdown signal
		self._shutdown_signalled = asyncio.Event()

		# Create the control connection
		self.control = await self.service.control(daemon=self)

		# Run main loop
		while True:
			# Submit stats
			await self.control.submit_stats()

			# Check if we are running by awaiting the shutdown signal
			try:
				await asyncio.wait_for(self._shutdown_signalled.wait(), timeout=5)
				break
			except asyncio.TimeoutError:
				pass

		# Main loop has ended, but we wait until all workers have finished.
		self.terminate_all_workers()

	def shutdown(self):
		"""
			Terminates all workers and exists the daemon.
		"""
		# Ignore if the main method has never been called
		if not self._shutdown_signalled:
			return

		# Ignore, if we are already shutting down
		if self._shutdown_signalled.is_set():
			return

		self.log.info(_("Shutting down..."))
		self._shutdown_signalled.set()

		# Close the control connection
		if self.control:
			self.control.close()

	def terminate_all_workers(self):
		"""
			Terminates all workers.
		"""
		self.log.debug("Sending SIGTERM to all workers")

		# First send SIGTERM to all processes
		for worker in self.workers:
			worker.terminate()

		self.log.debug("Waiting for workers to terminate")

		# Then wait until they all have finished.
		for worker in self.workers:
			worker.join()

		self.log.debug("All workers have finished")

	# Signal handling.

	def register_signal_handlers(self):
		signal.signal(signal.SIGCHLD, self.handle_SIGCHLD)
		signal.signal(signal.SIGINT,  self.handle_SIGTERM)
		signal.signal(signal.SIGTERM, self.handle_SIGTERM)

	def handle_SIGCHLD(self, signum, frame):
		"""
			Handle signal SIGCHLD.
		"""
		# Find the worker process that has terminated
		for worker in self.workers:
			# Skip any workers that are still alive
			if worker.is_alive():
				continue

			self.log.debug("Worker %s has terminated with status %s" % \
				(worker.pid, worker.exitcode))

			# Remove the worker from the list
			try:
				self.workers.remove(worker)
			except ValueError:
				pass

			# Close the process
			worker.close()

			# We finish after handling one worker. If multiple workers have finished
			# at the same time, this handler will be called again to handle it.
			break

	def handle_SIGTERM(self, signum, frame):
		"""
			Handle signal SIGTERM.
		"""
		# Just shut down.
		self.shutdown()

	def job_received(self, job):
		"""
			Called when this builder was assigned a new job
		"""
		# Launch a new worker
		worker = Worker(self, job)
		self.workers.append(worker)

		# Run it
		worker.start()

		self.log.debug("Spawned a new worker process as PID %s" % worker.pid)


class Worker(multiprocessing.Process):
	def __init__(self, daemon, data):
		multiprocessing.Process.__init__(self)
		self.daemon = daemon

		# The job that has been received
		self.data = data

	@property
	def service(self):
		return self.daemon.service

	@property
	def log(self):
		return self.daemon.log

	def run(self):
		self.log.debug("Worker %s has launched" % self.pid)

		# Register signal handlers
		self.register_signal_handlers()

		# Run everything from here asynchronously
		asyncio.run(self._work())

		self.log.debug("Worker %s terminated gracefully" % self.pid)

	def is_test(self):
		"""
			Returns True if this job is a test job
		"""
		return self.data.get("test", False)

	@property
	def job_id(self):
		return self.data.get("id")

	@property
	def ccache(self):
		"""
			ccache settings
		"""
		return self.data.get("ccache", {})

	@property
	def ccache_enabled(self):
		return self.ccache.get("enabled", False)

	@property
	def ccache_path(self):
		"""
			The ccache path for this job
		"""
		path = self.ccache.get("path", None)

		if path:
			return os.path.join(self.daemon.ccache_path, path)

	async def _work(self):
		"""
			Called from the async IO loop doing all the work
		"""
		success = False

		# Check if we have received some useful data
		if not self.job_id:
			raise ValueError("Did not receive a job ID")

		# Set the process title
		setproctitle.setproctitle("pakfire-worker job %s" % self.job_id)

		# Fetch the build architecture
		arch = self.data.get("arch")

		# Fetch the package URL
		pkg = self.data.get("pkg")
		if not pkg:
			raise ValueError("Did not received a package URL")

		# Connect to the service
		self.job = await self.service.job(self.job_id, worker=self)

		# Setup build logger
		logger = BuildLogger(self.log, self.job)

		# Create a temporary directory in which the built packages will be copied
		with tempfile.TemporaryDirectory(prefix="pakfire-packages-") as target:
			packages = []

			# Run the build
			try:
				build = self.build(pkg, arch=arch, target=target,
					logger=logger._log, build_id=self.job_id,

					# Always disable using snapshots
					disable_snapshot=True,

					# ccache
					disable_ccache=not self.ccache_enabled,
					ccache_path=self.ccache_path,
				)

				# Wait until the build process is done and stream the log in the meantime
				while not build.done():
					await logger.stream(timeout=1)

				# Await the build task (which would raise any exceptions)
				await build

			# Catch any other Exception
			except Exception as e:
				raise e

			# The build has finished successfully
			else:
				success = True

				# Find any packages
				if not self.is_test():
					packages = glob.glob("%s/*.pfm" % target)

			# Notify the service that the job has finished
			finally:
				await self.job.finished(
					success=success,
					logfile=logger.logfile.name,
					packages=packages,
				)

	def build(self, *args, **kwargs):
		"""
			Sets up a new Pakfire instance and runs it in a new thread.

			This method returns an async.Task() object which can be used to track
			if this job is still running.
		"""
		thread = asyncio.to_thread(self._build, *args, **kwargs)

		return asyncio.create_task(thread)

	def _build(self, pkg, arch=None, logger=None, **kwargs):
		# Setup Pakfire instance
		p = _pakfire.Pakfire(arch=arch, conf=self.pakfire_conf, logger=logger)

		# Run the build
		return p.build(pkg, **kwargs)

	def shutdown(self):
		self.log.debug("Shutting down worker %s" % self.pid)

		# XXX figure out what to do, when a build is running

	def abort(self, *args, **kwargs):
		"""
			Called to abort a running build immediately
		"""
		log.warning("Build job has been aborted")

		# XXX TODO

	# Signal handling.

	def register_signal_handlers(self):
		signal.signal(signal.SIGCHLD, self.handle_SIGCHLD)
		signal.signal(signal.SIGINT,  self.handle_SIGTERM)
		signal.signal(signal.SIGTERM, self.handle_SIGTERM)

	def handle_SIGCHLD(self, signum, frame):
		"""
			Handle signal SIGCHLD.
		"""
		# Must be here so that SIGCHLD won't be propagated to
		# PakfireDaemon.
		pass

	def handle_SIGTERM(self, signum, frame):
		"""
			Handle signal SIGTERM.
		"""
		self.shutdown()

	@functools.cached_property
	def pakfire_conf(self):
		"""
			Writes the pakfire configuration to file and returns its path
		"""
		conf = self.data.get("conf")

		# Dump pakfire configuration
		log.debug("Pakfire configuration:\n%s" % conf)

		return io.StringIO(conf)


class BuildLogger(object):
	"""
		This class groups together all sorts of logging.
	"""
	def __init__(self, log, job):
		self.log = log
		self.job = job

		# Create a logfile
		self.logfile = tempfile.NamedTemporaryFile(mode="w")

		# Create a FIFO queue to buffer any log messages
		self.queue = asyncio.Queue()

		# Create a new logger
		self.logger = self.log.getChild(self.job.id)
		self.logger.setLevel(logging.DEBUG)

		# Log everything to the queue
		handler = logging.handlers.QueueHandler(self.queue)
		handler.setLevel(logging.INFO)
		self.logger.addHandler(handler)

		# Log everything to the file
		handler = logging.StreamHandler(self.logfile)
		handler.setLevel(logging.INFO)
		self.logger.addHandler(handler)

	def _log(self, level, message):
		# Remove any trailing newline (but only one)
		if message:
			message = message.removesuffix("\n")

		return self.logger.log(level, message)

	async def stream(self, timeout=0):
		while True:
			# Fetch a message from the queue
			try:
				message = await asyncio.wait_for(self.queue.get(), timeout=timeout)

			# If we did not receive any messages within the timeout,
			# we return control back to the caller
			except asyncio.TimeoutError as e:
				break

			# Ignore any empty messages
			if message is None:
				continue

			# Send message to the service
			await self.job.log(message.created, message.levelno, message.getMessage())