+++ /dev/null
-# This file was part of Buildbot. Buildbot is free software: you can
-# redistribute it and/or modify it under the terms of the GNU General Public
-# License as published by the Free Software Foundation, version 2.
-#
-# This program is distributed in the hope that it will be useful, but WITHOUT
-# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
-# details.
-#
-# You should have received a copy of the GNU General Public License along with
-# this program; if not, write to the Free Software Foundation, Inc., 51
-# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
-#
-# Portions Copyright Buildbot Team Members
-# Portions Copyright 2010 Isotoma Limited
-
-
-import os
-
-from twisted.internet import defer
-from twisted.internet import threads
-from twisted.internet import utils
-from twisted.python import failure
-from twisted.python import log
-
-from buildbot import config
-from buildbot.util.eventual import eventually
-from buildbot.worker import AbstractLatentWorker
-
-try:
- import libvirt
-except ImportError:
- libvirt = None
-
-import random
-import string
-
-def random_string_generator():
- chars = string.ascii_letters
- return ''.join(random.choice(chars) for x in range(6))
-
-class WorkQueue:
-
- """
- I am a class that turns parallel access into serial access.
-
- I exist because we want to run libvirt access in threads as we don't
- trust calls not to block, but under load libvirt doesn't seem to like
- this kind of threaded use.
- """
-
- def __init__(self):
- self.queue = []
-
- def _process(self):
- log.msg("Looking to start a piece of work now...")
-
- # Is there anything to do?
- if not self.queue:
- log.msg("_process called when there is no work")
- return
-
- # Peek at the top of the stack - get a function to call and
- # a deferred to fire when its all over
- d, next_operation, args, kwargs = self.queue[0]
-
- # Start doing some work - expects a deferred
- try:
- d2 = next_operation(*args, **kwargs)
- except Exception:
- d2 = defer.fail()
-
- # Whenever a piece of work is done, whether it worked or not
- # call this to schedule the next piece of work
- @d2.addBoth
- def _work_done(res):
- log.msg("Completed a piece of work")
- self.queue.pop(0)
- if self.queue:
- log.msg("Preparing next piece of work")
- eventually(self._process)
- return res
-
- # When the work is done, trigger d
- d2.chainDeferred(d)
-
- def execute(self, cb, *args, **kwargs):
- kickstart_processing = not self.queue
- d = defer.Deferred()
- self.queue.append((d, cb, args, kwargs))
- if kickstart_processing:
- self._process()
- return d
-
- def executeInThread(self, cb, *args, **kwargs):
- return self.execute(threads.deferToThread, cb, *args, **kwargs)
-
-
-# A module is effectively a singleton class, so this is OK
-queue = WorkQueue()
-
-
-class Domain:
-
- """
- I am a wrapper around a libvirt Domain object
- """
-
- def __init__(self, connection, domain):
- self.connection = connection
- self.domain = domain
-
- def name(self):
- return queue.executeInThread(self.domain.name)
-
- def create(self):
- return queue.executeInThread(self.domain.create)
-
- def shutdown(self):
- return queue.executeInThread(self.domain.shutdown)
-
- def destroy(self):
- return queue.executeInThread(self.domain.destroy)
-
-class Volume:
- def __init__(self, connection, volume):
- self.connection = connection
- self.volume = volume
-
- @defer.inlineCallbacks
- def destroy(self):
- yield queue.executeInThread(self.volume.wipe)
- yield queue.executeInThread(self.volume.delete)
-
-class Pool:
- VolumeClass = Volume
- def __init__(self, connection, pool):
- self.connection = connection
- self.pool = pool
-
- @defer.inlineCallbacks
- def create_volume(self, xml):
- res = yield queue.executeInThread(self.pool.createXML, xml)
- return self.VolumeClass(self.connection, res)
-
-class Connection:
-
- """
- I am a wrapper around a libvirt Connection object.
- """
-
- DomainClass = Domain
- PoolClass = Pool
-
- def __init__(self, uri):
- self.uri = uri
- self._connection = None
-
- @property
- def connection(self):
- if self._connection is not None:
- try:
- if not self._connection.isAlive():
- self._connection = None
- except:
- self._connection = None
- if self._connection is None:
- self._connection = libvirt.open(self.uri)
- return self._connection
-
- @defer.inlineCallbacks
- def create(self, xml):
- """ I take libvirt XML and start a new VM """
- res = yield queue.executeInThread(self.connection.createXML, xml, 0)
- return self.DomainClass(self, res)
-
- @defer.inlineCallbacks
- def lookup_pool(self, name):
- res = yield queue.executeInThread(self.connection.storagePoolLookupByName, name)
- return self.PoolClass(self, res)
-
-class LibVirtWorker(AbstractLatentWorker):
-
- def __init__(self, name, password, connection, master_url, base_image=None, **kwargs):
- super().__init__(name, password, **kwargs)
- if not libvirt:
- config.error(
- "The python module 'libvirt' is needed to use a LibVirtWorker")
-
- self.master_url = master_url
- self.random_name = random_string_generator()
- self.connection = connection
- self.base_image = base_image
-
- self.domain = None
- self.domain_name = "buildbot-" + self.workername + "-" + self.random_name
- self.volume = None
- self.volume_name = "buildbot-" + self.workername + "-" + self.random_name
- self.pool_name = "buildbot-disks"
-
- def reconfigService(self, *args, **kwargs):
- if 'build_wait_timeout' not in kwargs:
- kwargs['build_wait_timeout'] = 0
- return super().reconfigService(*args, **kwargs)
-
- def canStartBuild(self):
- if self.domain and not self.isConnected():
- log.msg(
- "Not accepting builds as existing domain but worker not connected")
- return False
-
- return super().canStartBuild()
-
- @defer.inlineCallbacks
- def _prepare_image(self):
- log.msg("Creating temporary image {}".format(self.volume_name))
- pool = yield self.connection.lookup_pool(self.pool_name)
- vol_xml = """
- <volume type='file'>
- <name>{vol_name}</name>
- <capacity unit='G'>10</capacity>
- <target>
- <format type='qcow2'/>
- <permissions>
- <mode>0600</mode>
- <owner>0</owner>
- <group>0</group>
- </permissions>
- </target>
- <backingStore>
- <path>/etc/libvirtd/base-images/buildbot.qcow2</path>
- <format type='qcow2'/>
- </backingStore>
- </volume>
- """.format(vol_name = self.volume_name)
- self.volume = yield pool.create_volume(vol_xml)
-
- @defer.inlineCallbacks
- def start_instance(self, build):
- """
- I start a new instance of a VM.
-
- If a base_image is specified, I will make a clone of that otherwise i will
- use image directly.
-
- If i'm not given libvirt domain definition XML, I will look for my name
- in the list of defined virtual machines and start that.
- """
- domain_xml = """
- <domain type="kvm">
- <name>{domain_name}</name>
- <memory unit="GiB">2</memory>
- <vcpu>1</vcpu>
- <sysinfo type='smbios'>
- <oemStrings>
- <entry>buildbot_master_url={master_url}</entry>
- <entry>buildbot_worker_name={worker_name}</entry>
- </oemStrings>
- </sysinfo>
- <os>
- <type arch="x86_64">hvm</type>
- <smbios mode='sysinfo'/>
- </os>
- <devices>
- <emulator>/run/current-system/sw/bin/qemu-system-x86_64</emulator>
- <disk type="volume" device="disk">
- <driver name='qemu' type='qcow2' />
- <source type="volume" pool="{pool_name}" volume="{volume_name}" />
- <backingStore type='volume'>
- <format type='qcow2'/>
- <source type="volume" pool="niximages" volume="buildbot.qcow2" />
- </backingStore>
- <target dev="vda" bus="virtio"/>
- </disk>
- <input type="keyboard" bus="usb"/>
- <graphics type="vnc" port="-1" autoport="yes"/>
- <interface type="network">
- <source network="immae" />
- </interface>
- </devices>
- </domain>
- """.format(volume_name = self.volume_name, master_url = self.master_url, pool_name =
- self.pool_name, domain_name = self.domain_name, worker_name = self.workername)
-
- yield self._prepare_image()
-
- try:
- self.domain = yield self.connection.create(domain_xml)
- except Exception:
- log.err(failure.Failure(),
- ("Cannot start a VM ({}), failing gracefully and triggering"
- "a new build check").format(self.workername))
- self.domain = None
- return False
-
- return [self.domain_name]
-
- def stop_instance(self, fast=False):
- """
- I attempt to stop a running VM.
- I make sure any connection to the worker is removed.
- If the VM was using a cloned image, I remove the clone
- When everything is tidied up, I ask that bbot looks for work to do
- """
-
- log.msg("Attempting to stop '{}'".format(self.workername))
- if self.domain is None:
- log.msg("I don't think that domain is even running, aborting")
- return defer.succeed(None)
-
- domain = self.domain
- self.domain = None
-
- d = domain.destroy()
- if self.volume is not None:
- self.volume.destroy()
-
- return d
projects / perso / Immae / Config / Nix.git / blobdiff