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

#!/usr/bin/python

import hashlib
import json
import multiprocessing
import os
import signal
import sys
import tempfile
import time

import pakfire.base
import pakfire.builder
import pakfire.config
import pakfire.downloader
import pakfire.system
import pakfire.util
from pakfire.system import system

import base
import transport

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

import logging
log = logging.getLogger("pakfire.daemon")

class BuildJob(dict):
	"""
		Wrapper class for build jobs, that are received from the hub.

		This makes accessing attributes more easy.
	"""
	def __getattr__(self, key):
		try:
			return self[key]
		except KeyError:
			raise AttributeError, key


class PakfireDaemon(object):
	def __init__(self, config):
		self.config = config

		# Indicates if this daemon is in running mode.
		self.__running = True

		# Create daemon that sends keep-alive messages.
		self.keepalive = PakfireDaemonKeepalive(self.config)

		# List of worker processes.
		self.__workers = [self.keepalive]

		### Configuration
		# Number of workers in waiting state.
		self.max_waiting = 1

		# Number of running workers.
		self.max_running = system.cpu_count * 2

	def run(self, heartbeat=30):
		"""
			Main loop.
		"""
		# Register signal handlers.
		self.register_signal_handlers()

		# Start keepalive process.
		self.keepalive.start()

		# Run main loop.
		while self.__running:
			time_started = time.time()

			# Spawn a sufficient number of worker processes.
			self.spawn_workers_if_needed()

			# Get runtime of this loop iteration.
			time_elapsed = time.time() - time_started

			# Wait if the heartbeat time has not been reached, yet.
			if time_elapsed < heartbeat:
				time.sleep(heartbeat - time_elapsed)

		# 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.
		"""
		if not self.__running:
			return

		log.info(_("Shutting down..."))
		self.__running = False

	def spawn_workers_if_needed(self, *args, **kwargs):
		"""
			Spawns more workers if needed.
		"""
		# Do not create any more processes when the daemon is shutting down.
		if not self.__running:
			return

		# Cleanup all other workers.
		self.cleanup_workers()

		# Do not create more workers if there are already enough workers
		# active.
		if len(self.workers) >= self.max_running:
			log.warning("More workers running than allowed")
			return

		# Do nothing, if there is are already enough workers waiting.
		wanted_waiting_workers = self.max_waiting - len(self.waiting_workers)
		if wanted_waiting_workers <= 0:
			return

		# Spawn a new worker.
		for i in range(wanted_waiting_workers):
			self.spawn_worker(*args, **kwargs)

	def spawn_worker(self, *args, **kwargs):
		"""
			Spawns a new worker process.
		"""
		worker = PakfireWorker(config=self.config, *args, **kwargs)
		worker.start()

		log.debug("Spawned new worker process: %s" % worker)
		self.__workers.append(worker)

	def terminate_worker(self, worker):
		"""
			Terminates the given worker.
		"""
		log.warning(_("Terminating worker process: %s") % worker)

		worker.terminate()

	def terminate_all_workers(self):
		"""
			Terminates all workers.
		"""
		# First send SIGTERM to all processes.
		self.terminate_worker(self.keepalive)
		for worker in self.workers:
			self.terminate_worker(worker)

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

	def remove_worker(self, worker):
		"""
			Removes a worker from the internal list of worker processes.
		"""
		assert not worker.is_alive(), "Remove alive worker?"

		log.debug("Removing worker: %s" % worker)
		try:
			self.__workers.remove(worker)
		except:
			pass

	def cleanup_workers(self):
		"""
			Remove workers that are not alive any more.
		"""
		for worker in self.workers:
			if worker.is_alive():
				continue

			self.remove_worker(worker)

	@property
	def workers(self):
		return [w for w in self.__workers if isinstance(w, PakfireWorker)]

	@property
	def running_workers(self):
		workers = []

		for worker in self.workers:
			if worker.waiting.is_set():
				continue

			workers.append(worker)

		return workers

	@property
	def waiting_workers(self):
		workers = []

		for worker in self.workers:
			if worker.waiting.is_set():
				workers.append(worker)

		return workers

	# 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.
		"""
		# Spawn new workers if necessary.
		self.spawn_workers_if_needed()

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


class PakfireDaemonKeepalive(multiprocessing.Process):
	def __init__(self, config):
		multiprocessing.Process.__init__(self)

		# Save config.
		self.config = config

	def run(self, heartbeat=30):
		# Register signal handlers.
		self.register_signal_handlers()

		# Create connection to the hub.
		self.transport = transport.PakfireHubTransport(self.config)

		# Send our profile to the hub.
		self.send_builder_info()

		while True:
			time_started = time.time()

			# Send keepalive message.
			self.send_keepalive()

			# Get runtime of this loop iteration.
			time_elapsed = time.time() - time_started

			# Wait if the heartbeat time has not been reached, yet.
			if time_elapsed < heartbeat:
				time.sleep(heartbeat - time_elapsed)

	def shutdown(self):
		"""
			Ends this process immediately.
		"""
		sys.exit(1)

	# 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.
		"""
		# Just shut down.
		self.shutdown()

	# Talking to the hub.

	def send_builder_info(self):
		log.info(_("Sending builder information to hub..."))

		data = {
			# CPU info
			"cpu_model"       : system.cpu_model,
			"cpu_count"       : system.cpu_count,
			"cpu_arch"        : system.native_arch,
			"cpu_bogomips"    : system.cpu_bogomips,

			# Memory + swap
			"mem_total"       : system.memory,
			"swap_total"      : system.swap_total,

			# Pakfire + OS
			"pakfire_version" : PAKFIRE_VERSION,
			"host_key"        : self.config.get("signatures", "host_key", None),
			"os_name"         : system.distro.pretty_name,

			# Supported arches
			"supported_arches" : ",".join(system.supported_arches),
		}
		self.transport.post("/builders/info", data=data)

	def send_keepalive(self):
		log.debug("Sending keepalive message to hub...")

		data = {
			# Load average.
			"loadavg1"   : system.loadavg1,
			"loadavg5"   : system.loadavg5,
			"loadavg15"  : system.loadavg15,

			# Memory
			"mem_total"  : system.memory_total,
			"mem_free"   : system.memory_free,

			# Swap
			"swap_total" : system.swap_total,
			"swap_free"  : system.swap_free,

			# Disk space
			"space_free" : self.free_space,
		}
		self.transport.post("/builders/keepalive", data=data)

	@property
	def free_space(self):
		mp = system.get_mountpoint(BUILD_ROOT)

		return mp.space_left


class PakfireWorker(multiprocessing.Process):
	def __init__(self, config, waiting=None):
		multiprocessing.Process.__init__(self)

		# Save config.
		self.config = config

		# Waiting event. Clear if this worker is running a build.
		self.waiting = multiprocessing.Event()
		self.waiting.set()

		# Indicates if this worker is running.
		self.__running = True

	def run(self):
		# Register signal handlers.
		self.register_signal_handlers()

		# Create connection to the hub.
		self.transport = transport.PakfireHubTransport(self.config)

		while self.__running:
			# Try to get a new build job.
			job = self.get_new_build_job()
			if not job:
				continue

			# If we got a job, we are not waiting anymore.
			self.waiting.clear()

			# Run the job and return.
			return self.execute_job(job)

	def shutdown(self):
		self.__running = False

		# When we are just waiting, we can edit right away.
		if self.waiting.is_set():
			log.debug("Exiting immediately")
			sys.exit(1)

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

	# 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()

	def get_new_build_job(self, timeout=600):
		log.debug("Requesting new job...")

		try:
			job = self.transport.get_json("/builders/jobs/queue",
				data={ "timeout" : timeout, }, timeout=timeout)

			if job:
				return BuildJob(job)

		# As this is a long poll request, it is okay to time out.
		except TransportMaxTriesExceededError:
			pass

	def execute_job(self, job):
		log.debug("Executing job: %s" % job)

		# Call the function that processes the build and try to catch general
		# exceptions and report them to the server.
		# If everything goes okay, we tell this the server, too.
		try:
			# Create a temporary file and a directory for the resulting files.
			tmpdir  = tempfile.mkdtemp()
			tmpfile = os.path.join(tmpdir, os.path.basename(job.source_url))
			logfile = os.path.join(tmpdir, "build.log")

			# Create pakfire configuration instance.
			config = pakfire.config.ConfigDaemon()
			config.parse(job.config)

			# Create pakfire instance.
			p = None
			try:
				p = pakfire.base.PakfireBuilder(config=config, arch=job.arch)

				# Download the source package.
				grabber = pakfire.downloader.PackageDownloader(p)
				grabber.urlgrab(job.source_url, filename=tmpfile)

				# Check if the download checksum matches (if provided).
				if job.source_hash_sha512:
					h = hashlib.new("sha512")
					f = open(tmpfile, "rb")
					while True:
						buf = f.read(BUFFER_SIZE)
						if not buf:
							break

						h.update(buf)
					f.close()

					if not job.source_hash_sha512 == h.hexdigest():
						raise DownloadError, "Hash check did not succeed."

				# Create a new instance of a build environment.
				build = pakfire.builder.BuildEnviron(p, tmpfile,
					release_build=True, build_id=job.id, logfile=logfile)

				try:
					# Create the build environment.
					build.start()

					# Update the build status on the server.
					self.upload_buildroot(job, build.installed_packages)
					self.update_state(job, "running")

					# Run the build (without install test).
					build.build(install_test=False)

					# Copy the created packages to the tempdir.
					build.copy_result(tmpdir)

				finally:
					# Cleanup the build environment.
					build.stop()

				# Jippie, build is finished, we are going to upload the files.
				self.update_state(job, "uploading")

				# Walk through the result directory and upload all (binary) files.
				# Skip that for test builds.
				if not job.type == "test":
					for dir, subdirs, files in os.walk(tmpdir):
						for file in files:
							file = os.path.join(dir, file)
							if file in (logfile, tmpfile,):
								continue

							self.upload_file(job, file, "package")

			except DependencyError, e:
				message = "%s: %s" % (e.__class__.__name__, e)
				self.update_state(job, "dependency_error", message)
				raise

			except DownloadError, e:
				message = "%s: %s" % (e.__class__.__name__, e)
				self.update_state(job, "download_error", message)
				raise

			finally:
				if p:
					p.destroy()

				# Upload the logfile in any case and if it exists.
				if os.path.exists(logfile):
					self.upload_file(job, logfile, "log")

				# Cleanup the files we created.
				pakfire.util.rm(tmpdir)

		except DependencyError:
			# This has already been reported.
			raise

		except (DownloadError,):
			# Do not take any further action for these exceptions.
			pass

		except (KeyboardInterrupt, SystemExit):
			self.update_state(job, "aborted")

		except Exception, e:
			# Format the exception and send it to the server.
			message = "%s: %s" % (e.__class__.__name__, e)

			self.update_state(job, "failed", message)
			raise

		else:
			self.update_state(job, "finished")

	def update_state(self, job, state, message=None):
		"""
			Update state of the build job on the hub.
		"""
		data = {
			"message" : message or "",
		}

		self.transport.post("/builders/jobs/%s/state/%s" % (job.id, state),
			data=data)

	def upload_file(self, job, filename, type):
		assert os.path.exists(filename)
		assert type in ("package", "log")

		# First upload the file data and save the upload_id.
		upload_id = self.transport.upload_file(filename)

		data = {
			"type" : type,
		}

		# Add the file to the build.
		self.transport.post("/builders/jobs/%s/addfile/%s" % (job.id, upload_id),
			data=data)

	def upload_buildroot(self, job, installed_packages):
		pkgs = []
		for pkg in installed_packages:
			pkgs.append((pkg.friendly_name, pkg.uuid))

		data = { "buildroot" : json.dumps(pkgs) }

		self.transport.post("/builders/jobs/%s/buildroot" % job.id, data=data)
Commit	Line	Data
aa14071d MT	1	#!/usr/bin/python
	2
	3	import hashlib
	4	import json
	5	import multiprocessing
	6	import os
	7	import signal
	8	import sys
	9	import tempfile
	10	import time
	11
	12	import pakfire.base
	13	import pakfire.builder
	14	import pakfire.config
	15	import pakfire.downloader
	16	import pakfire.system
	17	import pakfire.util
	18	from pakfire.system import system
	19
	20	import base
	21	import transport
	22
	23	from pakfire.constants import *
	24	from pakfire.i18n import _
	25
	26	import logging
	27	log = logging.getLogger("pakfire.daemon")
	28
	29	class BuildJob(dict):
	30	"""
	31	Wrapper class for build jobs, that are received from the hub.
	32
	33	This makes accessing attributes more easy.
	34	"""
	35	def __getattr__(self, key):
	36	try:
	37	return self[key]
	38	except KeyError:
	39	raise AttributeError, key
	40
	41
	42	class PakfireDaemon(object):
	43	def __init__(self, config):
	44	self.config = config
	45
	46	# Indicates if this daemon is in running mode.
	47	self.__running = True
	48
9a9926fd MT	49	# Create daemon that sends keep-alive messages.
9a9926fd MT	50	self.keepalive = PakfireDaemonKeepalive(self.config)
aa14071d	51
9a9926fd MT	52	# List of worker processes.
9a9926fd MT	53	self.__workers = [self.keepalive]
aa14071d MT	54
	55	### Configuration
	56	# Number of workers in waiting state.
	57	self.max_waiting = 1
	58
	59	# Number of running workers.
	60	self.max_running = system.cpu_count * 2
	61
	62	def run(self, heartbeat=30):
	63	"""
	64	Main loop.
	65	"""
	66	# Register signal handlers.
	67	self.register_signal_handlers()
	68
9a9926fd MT	69	# Start keepalive process.
9a9926fd MT	70	self.keepalive.start()
aa14071d	71
9a9926fd	72	# Run main loop.
aa14071d MT	73	while self.__running:
	74	time_started = time.time()
	75
aa14071d MT	76	# Spawn a sufficient number of worker processes.
	77	self.spawn_workers_if_needed()
	78
	79	# Get runtime of this loop iteration.
	80	time_elapsed = time.time() - time_started
	81
	82	# Wait if the heartbeat time has not been reached, yet.
	83	if time_elapsed < heartbeat:
	84	time.sleep(heartbeat - time_elapsed)
	85
	86	# Main loop has ended, but we wait until all workers have finished.
	87	self.terminate_all_workers()
	88
	89	def shutdown(self):
	90	"""
	91	Terminates all workers and exists the daemon.
	92	"""
	93	if not self.__running:
	94	return
	95
	96	log.info(_("Shutting down..."))
	97	self.__running = False
	98
	99	def spawn_workers_if_needed(self, args, *kwargs):
	100	"""
	101	Spawns more workers if needed.
	102	"""
	103	# Do not create any more processes when the daemon is shutting down.
	104	if not self.__running:
	105	return
	106
	107	# Cleanup all other workers.
	108	self.cleanup_workers()
	109
	110	# Do not create more workers if there are already enough workers
	111	# active.
	112	if len(self.workers) >= self.max_running:
	113	log.warning("More workers running than allowed")
	114	return
	115
	116	# Do nothing, if there is are already enough workers waiting.
	117	wanted_waiting_workers = self.max_waiting - len(self.waiting_workers)
	118	if wanted_waiting_workers <= 0:
	119	return
	120
	121	# Spawn a new worker.
	122	for i in range(wanted_waiting_workers):
	123	self.spawn_worker(args, *kwargs)
	124
	125	def spawn_worker(self, args, *kwargs):
	126	"""
	127	Spawns a new worker process.
	128	"""
	129	worker = PakfireWorker(config=self.config, args, *kwargs)
	130	worker.start()
	131
	132	log.debug("Spawned new worker process: %s" % worker)
	133	self.__workers.append(worker)
	134
	135	def terminate_worker(self, worker):
	136	"""
	137	Terminates the given worker.
	138	"""
	139	log.warning(_("Terminating worker process: %s") % worker)
140
141	worker.terminate()
142
143	def terminate_all_workers(self):
144	"""
145	Terminates all workers.
146	"""
fa736f5a MT	147	# First send SIGTERM to all processes.
fa736f5a MT	148	self.terminate_worker(self.keepalive)
aa14071d MT	149	for worker in self.workers:
	150	self.terminate_worker(worker)
	151
fa736f5a	152	# Then wait until they all have finished.
9a9926fd	153	self.keepalive.join()
fa736f5a MT	154	for worker in self.workers:
fa736f5a MT	155	worker.join()
9a9926fd	156
aa14071d MT	157	def remove_worker(self, worker):
	158	"""
	159	Removes a worker from the internal list of worker processes.
	160	"""
	161	assert not worker.is_alive(), "Remove alive worker?"
	162
	163	log.debug("Removing worker: %s" % worker)
	164	try:
	165	self.__workers.remove(worker)
	166	except:
	167	pass
	168
	169	def cleanup_workers(self):
	170	"""
	171	Remove workers that are not alive any more.
	172	"""
	173	for worker in self.workers:
	174	if worker.is_alive():
	175	continue
	176
	177	self.remove_worker(worker)
	178
	179	@property
	180	def workers(self):
9a9926fd	181	return [w for w in self.__workers if isinstance(w, PakfireWorker)]
aa14071d MT	182
	183	@property
	184	def running_workers(self):
	185	workers = []
	186
	187	for worker in self.workers:
	188	if worker.waiting.is_set():
	189	continue
	190
	191	workers.append(worker)
	192
	193	return workers
	194
	195	@property
	196	def waiting_workers(self):
	197	workers = []
	198
	199	for worker in self.workers:
	200	if worker.waiting.is_set():
	201	workers.append(worker)
	202
	203	return workers
	204
	205	# Signal handling.
	206
	207	def register_signal_handlers(self):
	208	signal.signal(signal.SIGCHLD, self.handle_SIGCHLD)
	209	signal.signal(signal.SIGINT, self.handle_SIGTERM)
	210	signal.signal(signal.SIGTERM, self.handle_SIGTERM)
	211
	212	def handle_SIGCHLD(self, signum, frame):
	213	"""
	214	Handle signal SIGCHLD.
	215	"""
	216	# Spawn new workers if necessary.
	217	self.spawn_workers_if_needed()
	218
	219	def handle_SIGTERM(self, signum, frame):
	220	"""
	221	Handle signal SIGTERM.
	222	"""
	223	# Just shut down.
	224	self.shutdown()
	225
9a9926fd MT	226
	227	class PakfireDaemonKeepalive(multiprocessing.Process):
	228	def __init__(self, config):
	229	multiprocessing.Process.__init__(self)
	230
	231	# Save config.
	232	self.config = config
	233
	234	def run(self, heartbeat=30):
	235	# Register signal handlers.
	236	self.register_signal_handlers()
	237
	238	# Create connection to the hub.
	239	self.transport = transport.PakfireHubTransport(self.config)
9a9926fd MT	240
	241	# Send our profile to the hub.
	242	self.send_builder_info()
	243
	244	while True:
	245	time_started = time.time()
	246
	247	# Send keepalive message.
	248	self.send_keepalive()
	249
	250	# Get runtime of this loop iteration.
	251	time_elapsed = time.time() - time_started
	252
	253	# Wait if the heartbeat time has not been reached, yet.
	254	if time_elapsed < heartbeat:
	255	time.sleep(heartbeat - time_elapsed)
	256
	257	def shutdown(self):
	258	"""
	259	Ends this process immediately.
	260	"""
	261	sys.exit(1)
	262
	263	# Signal handling.
	264
	265	def register_signal_handlers(self):
	266	signal.signal(signal.SIGCHLD, self.handle_SIGCHLD)
	267	signal.signal(signal.SIGINT, self.handle_SIGTERM)
	268	signal.signal(signal.SIGTERM, self.handle_SIGTERM)
	269
	270	def handle_SIGCHLD(self, signum, frame):
	271	"""
	272	Handle signal SIGCHLD.
	273	"""
	274	# Must be here so that SIGCHLD won't be propagated to
	275	# PakfireDaemon.
	276	pass
	277
	278	def handle_SIGTERM(self, signum, frame):
	279	"""
	280	Handle signal SIGTERM.
	281	"""
	282	# Just shut down.
	283	self.shutdown()
	284
aa14071d MT	285	# Talking to the hub.
	286
	287	def send_builder_info(self):
	288	log.info(_("Sending builder information to hub..."))
	289
	290	data = {
	291	# CPU info
	292	"cpu_model" : system.cpu_model,
	293	"cpu_count" : system.cpu_count,
	294	"cpu_arch" : system.native_arch,
	295	"cpu_bogomips" : system.cpu_bogomips,
	296
	297	# Memory + swap
	298	"mem_total" : system.memory,
	299	"swap_total" : system.swap_total,
	300
	301	# Pakfire + OS
	302	"pakfire_version" : PAKFIRE_VERSION,
	303	"host_key" : self.config.get("signatures", "host_key", None),
	304	"os_name" : system.distro.pretty_name,
	305
	306	# Supported arches
	307	"supported_arches" : ",".join(system.supported_arches),
	308	}
	309	self.transport.post("/builders/info", data=data)
	310
	311	def send_keepalive(self):
	312	log.debug("Sending keepalive message to hub...")
	313
	314	data = {
	315	# Load average.
	316	"loadavg1" : system.loadavg1,
	317	"loadavg5" : system.loadavg5,
	318	"loadavg15" : system.loadavg15,
	319
	320	# Memory
	321	"mem_total" : system.memory_total,
	322	"mem_free" : system.memory_free,
	323
	324	# Swap
	325	"swap_total" : system.swap_total,
	326	"swap_free" : system.swap_free,
	327
	328	# Disk space
	329	"space_free" : self.free_space,
	330	}
	331	self.transport.post("/builders/keepalive", data=data)
	332
	333	@property
	334	def free_space(self):
	335	mp = system.get_mountpoint(BUILD_ROOT)
	336
	337	return mp.space_left
	338
	339
	340	class PakfireWorker(multiprocessing.Process):
	341	def __init__(self, config, waiting=None):
	342	multiprocessing.Process.__init__(self)
	343
	344	# Save config.
	345	self.config = config
	346
	347	# Waiting event. Clear if this worker is running a build.
	348	self.waiting = multiprocessing.Event()
349	self.waiting.set()
350
351	# Indicates if this worker is running.
352	self.__running = True
353
354	def run(self):
355	# Register signal handlers.
356	self.register_signal_handlers()
357
358	# Create connection to the hub.
359	self.transport = transport.PakfireHubTransport(self.config)
aa14071d MT	360
	361	while self.__running:
	362	# Try to get a new build job.
	363	job = self.get_new_build_job()
	364	if not job:
	365	continue
	366
	367	# If we got a job, we are not waiting anymore.
	368	self.waiting.clear()
	369
	370	# Run the job and return.
	371	return self.execute_job(job)
	372
	373	def shutdown(self):
	374	self.__running = False
	375
	376	# When we are just waiting, we can edit right away.
	377	if self.waiting.is_set():
	378	log.debug("Exiting immediately")
	379	sys.exit(1)
	380
	381	# XXX figure out what to do, when a build is running.
	382
	383	# Signal handling.
	384
	385	def register_signal_handlers(self):
	386	signal.signal(signal.SIGCHLD, self.handle_SIGCHLD)
	387	signal.signal(signal.SIGINT, self.handle_SIGTERM)
	388	signal.signal(signal.SIGTERM, self.handle_SIGTERM)
	389
	390	def handle_SIGCHLD(self, signum, frame):
	391	"""
	392	Handle signal SIGCHLD.
	393	"""
	394	# Must be here so that SIGCHLD won't be propagated to
	395	# PakfireDaemon.
	396	pass
	397
	398	def handle_SIGTERM(self, signum, frame):
	399	"""
	400	Handle signal SIGTERM.
	401	"""
	402	self.shutdown()
	403
	404	def get_new_build_job(self, timeout=600):
	405	log.debug("Requesting new job...")
	406
	407	try:
	408	job = self.transport.get_json("/builders/jobs/queue",
	409	data={ "timeout" : timeout, }, timeout=timeout)
	410
	411	if job:
	412	return BuildJob(job)
	413
	414	# As this is a long poll request, it is okay to time out.
	415	except TransportMaxTriesExceededError:
	416	pass
	417
	418	def execute_job(self, job):
	419	log.debug("Executing job: %s" % job)
	420
	421	# Call the function that processes the build and try to catch general
	422	# exceptions and report them to the server.
	423	# If everything goes okay, we tell this the server, too.
424	try:
425	# Create a temporary file and a directory for the resulting files.
426	tmpdir = tempfile.mkdtemp()
427	tmpfile = os.path.join(tmpdir, os.path.basename(job.source_url))
428	logfile = os.path.join(tmpdir, "build.log")
429
430	# Create pakfire configuration instance.
431	config = pakfire.config.ConfigDaemon()
432	config.parse(job.config)
433
434	# Create pakfire instance.
435	p = None
436	try:
437	p = pakfire.base.PakfireBuilder(config=config, arch=job.arch)
438
439	# Download the source package.
440	grabber = pakfire.downloader.PackageDownloader(p)
441	grabber.urlgrab(job.source_url, filename=tmpfile)
442
443	# Check if the download checksum matches (if provided).
444	if job.source_hash_sha512:
445	h = hashlib.new("sha512")
446	f = open(tmpfile, "rb")
447	while True:
448	buf = f.read(BUFFER_SIZE)
449	if not buf:
450	break
451
452	h.update(buf)
453	f.close()
454
455	if not job.source_hash_sha512 == h.hexdigest():
456	raise DownloadError, "Hash check did not succeed."
457
458	# Create a new instance of a build environment.
459	build = pakfire.builder.BuildEnviron(p, tmpfile,
460	release_build=True, build_id=job.id, logfile=logfile)
461
462	try:
463	# Create the build environment.
464	build.start()
465
466	# Update the build status on the server.
467	self.upload_buildroot(job, build.installed_packages)
468	self.update_state(job, "running")
469
470	# Run the build (without install test).
471	build.build(install_test=False)
472
473	# Copy the created packages to the tempdir.
474	build.copy_result(tmpdir)
475
476	finally:
477	# Cleanup the build environment.
478	build.stop()
479
480	# Jippie, build is finished, we are going to upload the files.
481	self.update_state(job, "uploading")
482
483	# Walk through the result directory and upload all (binary) files.
484	# Skip that for test builds.
485	if not job.type == "test":
486	for dir, subdirs, files in os.walk(tmpdir):
487	for file in files:
488	file = os.path.join(dir, file)
489	if file in (logfile, tmpfile,):
490	continue
491
492	self.upload_file(job, file, "package")
493
494	except DependencyError, e:
495	message = "%s: %s" % (e.__class__.__name__, e)
496	self.update_state(job, "dependency_error", message)
497	raise
498
499	except DownloadError, e:
500	message = "%s: %s" % (e.__class__.__name__, e)
501	self.update_state(job, "download_error", message)
502	raise
503
504	finally:
505	if p:
506	p.destroy()
507
508	# Upload the logfile in any case and if it exists.
509	if os.path.exists(logfile):
510	self.upload_file(job, logfile, "log")
511
512	# Cleanup the files we created.
513	pakfire.util.rm(tmpdir)
514
515	except DependencyError:
516	# This has already been reported.
517	raise
518
519	except (DownloadError,):
520	# Do not take any further action for these exceptions.
521	pass
522
523	except (KeyboardInterrupt, SystemExit):
524	self.update_state(job, "aborted")
525
526	except Exception, e:
527	# Format the exception and send it to the server.
528	message = "%s: %s" % (e.__class__.__name__, e)
529
530	self.update_state(job, "failed", message)
531	raise
532
533	else:
534	self.update_state(job, "finished")
535
536	def update_state(self, job, state, message=None):
537	"""
538	Update state of the build job on the hub.
539	"""
540	data = {
541	"message" : message or "",
542	}
543
544	self.transport.post("/builders/jobs/%s/state/%s" % (job.id, state),
545	data=data)
546
547	def upload_file(self, job, filename, type):
548	assert os.path.exists(filename)
549	assert type in ("package", "log")
550
551	# First upload the file data and save the upload_id.
552	upload_id = self.transport.upload_file(filename)
553
554	data = {
555	"type" : type,
556	}
557
558	# Add the file to the build.
559	self.transport.post("/builders/jobs/%s/addfile/%s" % (job.id, upload_id),
560	data=data)
561
562	def upload_buildroot(self, job, installed_packages):
563	pkgs = []
564	for pkg in installed_packages:
565	pkgs.append((pkg.friendly_name, pkg.uuid))
566
567	data = { "buildroot" : json.dumps(pkgs) }
568
569	self.transport.post("/builders/jobs/%s/buildroot" % job.id, data=data)