1294 lines
49 KiB
Python
Executable File
1294 lines
49 KiB
Python
Executable File
#!/usr/bin/env python
|
|
# Copyright 2015 Hewlett-Packard Development Company, L.P.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import shlex
|
|
import uuid
|
|
|
|
from nodepool import config as nodepool_config
|
|
from nodepool import exceptions
|
|
from nodepool import provider_manager
|
|
from nodepool import stats
|
|
from nodepool import zk
|
|
|
|
|
|
MINS = 60
|
|
HOURS = 60 * MINS
|
|
# How long to wait for an image save
|
|
IMAGE_TIMEOUT = 6 * HOURS
|
|
|
|
# How long to wait between checks for ZooKeeper connectivity if it disappears.
|
|
SUSPEND_WAIT_TIME = 30
|
|
|
|
# HP Cloud requires qemu compat with 0.10. That version works elsewhere,
|
|
# so just hardcode it for all qcow2 building
|
|
DEFAULT_QEMU_IMAGE_COMPAT_OPTIONS = "--qemu-img-options 'compat=0.10'"
|
|
|
|
|
|
class DibImageFile(object):
|
|
'''
|
|
Class used as an API to finding locally built DIB image files, and
|
|
also used to represent the found files. Image files are named using
|
|
a unique ID, but can be available in multiple formats (with different
|
|
extensions).
|
|
'''
|
|
def __init__(self, image_id, extension=None):
|
|
self.image_id = image_id
|
|
self.extension = extension
|
|
self.md5 = None
|
|
self.md5_file = None
|
|
self.sha256 = None
|
|
self.sha256_file = None
|
|
|
|
@staticmethod
|
|
def from_path(path):
|
|
image_file = os.path.basename(path)
|
|
image_id, extension = image_file.rsplit('.', 1)
|
|
return DibImageFile(image_id, extension)
|
|
|
|
@staticmethod
|
|
def from_image_id(images_dir, image_id):
|
|
images = []
|
|
for image_filename in os.listdir(images_dir):
|
|
if os.path.isfile(os.path.join(images_dir, image_filename)):
|
|
image = DibImageFile.from_path(image_filename)
|
|
if image.image_id == image_id:
|
|
images.append(image)
|
|
return images
|
|
|
|
@staticmethod
|
|
def from_images_dir(images_dir):
|
|
return [DibImageFile.from_path(x) for x in os.listdir(images_dir)]
|
|
|
|
def to_path(self, images_dir, with_extension=True):
|
|
my_path = os.path.join(images_dir, self.image_id)
|
|
if with_extension:
|
|
if self.extension is None:
|
|
raise exceptions.BuilderError(
|
|
'Cannot specify image extension of None'
|
|
)
|
|
my_path += '.' + self.extension
|
|
|
|
md5_path = '%s.%s' % (my_path, 'md5')
|
|
md5 = self._checksum(md5_path)
|
|
if md5:
|
|
self.md5_file = md5_path
|
|
self.md5 = md5[0:32]
|
|
|
|
sha256_path = '%s.%s' % (my_path, 'sha256')
|
|
sha256 = self._checksum(sha256_path)
|
|
if sha256:
|
|
self.sha256_file = sha256_path
|
|
self.sha256 = sha256[0:64]
|
|
|
|
return my_path
|
|
|
|
def _checksum(self, filename):
|
|
if not os.path.isfile(filename):
|
|
return None
|
|
with open(filename, 'r') as f:
|
|
data = f.read()
|
|
return data
|
|
|
|
|
|
class BaseWorker(threading.Thread):
|
|
def __init__(self, builder_id, config_path, secure_path, interval, zk):
|
|
super(BaseWorker, self).__init__()
|
|
self.log = logging.getLogger("nodepool.builder.BaseWorker")
|
|
self.daemon = True
|
|
self._running = False
|
|
self._stop_event = threading.Event()
|
|
self._config = None
|
|
self._config_path = config_path
|
|
self._secure_path = secure_path
|
|
self._zk = zk
|
|
self._hostname = socket.gethostname()
|
|
self._statsd = stats.get_client()
|
|
self._interval = interval
|
|
self._builder_id = builder_id
|
|
|
|
def _checkForZooKeeperChanges(self, new_config):
|
|
'''
|
|
Check config for ZooKeeper cluster changes.
|
|
|
|
If the defined set of ZooKeeper servers changes, the connection
|
|
will use the new server set.
|
|
'''
|
|
if self._config.zookeeper_servers != new_config.zookeeper_servers:
|
|
self.log.debug("Detected ZooKeeper server changes")
|
|
self._zk.resetHosts(list(new_config.zookeeper_servers.values()))
|
|
|
|
@property
|
|
def running(self):
|
|
return self._running
|
|
|
|
def shutdown(self):
|
|
self._running = False
|
|
self._stop_event.set()
|
|
|
|
|
|
class CleanupWorker(BaseWorker):
|
|
'''
|
|
The janitor of nodepool-builder that will remove images from providers
|
|
and any local DIB builds.
|
|
'''
|
|
|
|
def __init__(self, name, builder_id, config_path, secure_path,
|
|
interval, zk):
|
|
super(CleanupWorker, self).__init__(builder_id, config_path,
|
|
secure_path, interval, zk)
|
|
self.log = logging.getLogger(
|
|
"nodepool.builder.CleanupWorker.%s" % name)
|
|
self.name = 'CleanupWorker.%s' % name
|
|
|
|
def _buildUploadRecencyTable(self):
|
|
'''
|
|
Builds a table for each image of the most recent uploads to each
|
|
provider.
|
|
|
|
Example)
|
|
|
|
image1:
|
|
providerA: [ (build_id, upload_id, upload_time), ... ]
|
|
providerB: [ (build_id, upload_id, upload_time), ... ]
|
|
image2:
|
|
providerC: [ (build_id, upload_id, upload_time), ... ]
|
|
'''
|
|
self._rtable = {}
|
|
for image in self._zk.getImageNames():
|
|
self._rtable[image] = {}
|
|
for build in self._zk.getBuilds(image, zk.READY):
|
|
for provider in self._zk.getBuildProviders(image, build.id):
|
|
if provider not in self._rtable[image]:
|
|
self._rtable[image][provider] = []
|
|
uploads = self._zk.getMostRecentBuildImageUploads(
|
|
2, image, build.id, provider, zk.READY)
|
|
for upload in uploads:
|
|
self._rtable[image][provider].append(
|
|
(build.id, upload.id, upload.state_time)
|
|
)
|
|
|
|
# Sort uploads by state_time (upload time) and keep the 2 most recent
|
|
for i in list(self._rtable.keys()):
|
|
for p in self._rtable[i].keys():
|
|
self._rtable[i][p].sort(key=lambda x: x[2], reverse=True)
|
|
self._rtable[i][p] = self._rtable[i][p][:2]
|
|
|
|
self.log.debug("Upload recency table: %s", self._rtable)
|
|
|
|
def _isRecentUpload(self, image, provider, build_id, upload_id):
|
|
'''
|
|
Search for an upload for a build within the recency table.
|
|
'''
|
|
provider = self._rtable[image].get(provider)
|
|
if not provider:
|
|
return False
|
|
|
|
for b_id, u_id, u_time in provider:
|
|
if build_id == b_id and upload_id == u_id:
|
|
return True
|
|
return False
|
|
|
|
def _inProgressUpload(self, upload):
|
|
'''
|
|
Determine if an upload is in progress.
|
|
'''
|
|
if upload.state != zk.UPLOADING:
|
|
return False
|
|
|
|
try:
|
|
with self._zk.imageUploadLock(upload.image_name, upload.build_id,
|
|
upload.provider_name,
|
|
blocking=False):
|
|
pass
|
|
except exceptions.ZKLockException:
|
|
return True
|
|
return False
|
|
|
|
def _removeDibItem(self, filename):
|
|
if filename is None:
|
|
return
|
|
try:
|
|
os.remove(filename)
|
|
self.log.info("Removed DIB file %s" % filename)
|
|
except OSError as e:
|
|
if e.errno != 2: # No such file or directory
|
|
raise e
|
|
|
|
def _deleteLocalBuild(self, image, build):
|
|
'''
|
|
Remove expired image build from local disk.
|
|
|
|
:param str image: Name of the image whose build we are deleting.
|
|
:param ImageBuild build: The build we want to delete.
|
|
|
|
:returns: True if files were deleted, False if none were found.
|
|
'''
|
|
base = "-".join([image, build.id])
|
|
files = DibImageFile.from_image_id(self._config.imagesdir, base)
|
|
if not files:
|
|
# NOTE(pabelanger): It is possible we don't have any files because
|
|
# diskimage-builder failed. So, check to see if we have the correct
|
|
# builder so we can removed the data from zookeeper.
|
|
|
|
# To maintain backward compatibility with builders that didn't
|
|
# use unique builder IDs before, but do now, always compare to
|
|
# hostname as well since some ZK data may still reference that.
|
|
if (build.builder_id == self._builder_id or
|
|
build.builder == self._hostname
|
|
):
|
|
return True
|
|
return False
|
|
|
|
self.log.info("Doing cleanup for %s:%s" % (image, build.id))
|
|
|
|
manifest_dir = None
|
|
|
|
for f in files:
|
|
filename = f.to_path(self._config.imagesdir, True)
|
|
if not manifest_dir:
|
|
path, ext = filename.rsplit('.', 1)
|
|
manifest_dir = path + ".d"
|
|
items = [filename, f.md5_file, f.sha256_file]
|
|
list(map(self._removeDibItem, items))
|
|
|
|
try:
|
|
shutil.rmtree(manifest_dir)
|
|
self.log.info("Removed DIB manifest %s" % manifest_dir)
|
|
except OSError as e:
|
|
if e.errno != 2: # No such file or directory
|
|
raise e
|
|
|
|
return True
|
|
|
|
def _cleanupProvider(self, provider, image, build_id):
|
|
all_uploads = self._zk.getUploads(image, build_id, provider.name)
|
|
|
|
for upload in all_uploads:
|
|
if self._isRecentUpload(image, provider.name, build_id, upload.id):
|
|
continue
|
|
self._deleteUpload(upload)
|
|
|
|
def _cleanupObsoleteProviderUploads(self, provider, image, build_id):
|
|
if image in provider.diskimages:
|
|
# This image is in use for this provider
|
|
return
|
|
|
|
all_uploads = self._zk.getUploads(image, build_id, provider.name)
|
|
for upload in all_uploads:
|
|
self._deleteUpload(upload)
|
|
|
|
def _deleteUpload(self, upload):
|
|
deleted = False
|
|
|
|
if upload.state != zk.DELETING:
|
|
if not self._inProgressUpload(upload):
|
|
data = zk.ImageUpload()
|
|
data.state = zk.DELETING
|
|
self._zk.storeImageUpload(upload.image_name, upload.build_id,
|
|
upload.provider_name, data,
|
|
upload.id)
|
|
deleted = True
|
|
|
|
if upload.state == zk.DELETING or deleted:
|
|
manager = self._config.provider_managers[upload.provider_name]
|
|
try:
|
|
# It is possible we got this far, but don't actually have an
|
|
# external_name. This could mean that zookeeper and cloud
|
|
# provider are some how out of sync.
|
|
if upload.external_name:
|
|
base = "-".join([upload.image_name, upload.build_id])
|
|
self.log.info("Deleting image build %s from %s" %
|
|
(base, upload.provider_name))
|
|
manager.deleteImage(upload.external_name)
|
|
except Exception:
|
|
self.log.exception(
|
|
"Unable to delete image %s from %s:",
|
|
upload.external_name, upload.provider_name)
|
|
else:
|
|
self.log.debug("Deleting image upload: %s", upload)
|
|
self._zk.deleteUpload(upload.image_name, upload.build_id,
|
|
upload.provider_name, upload.id)
|
|
|
|
def _inProgressBuild(self, build, image):
|
|
'''
|
|
Determine if a DIB build is in progress.
|
|
'''
|
|
if build.state != zk.BUILDING:
|
|
return False
|
|
|
|
try:
|
|
with self._zk.imageBuildLock(image, blocking=False):
|
|
# An additional state check is needed to make sure it hasn't
|
|
# changed on us. If it has, then let's pretend a build is
|
|
# still in progress so that it is checked again later with
|
|
# its new build state.
|
|
b = self._zk.getBuild(image, build.id)
|
|
if b.state != zk.BUILDING:
|
|
return True
|
|
pass
|
|
except exceptions.ZKLockException:
|
|
return True
|
|
return False
|
|
|
|
def _cleanup(self):
|
|
'''
|
|
Clean up builds on disk and in providers.
|
|
'''
|
|
known_providers = self._config.providers.values()
|
|
image_names = self._zk.getImageNames()
|
|
|
|
self._buildUploadRecencyTable()
|
|
|
|
for image in image_names:
|
|
try:
|
|
self._cleanupImage(known_providers, image)
|
|
except Exception:
|
|
self.log.exception("Exception cleaning up image %s:", image)
|
|
|
|
def _filterLocalBuilds(self, image, builds):
|
|
'''Return the subset of builds that are local'''
|
|
ret = []
|
|
for build in builds:
|
|
base = "-".join([image, build.id])
|
|
files = DibImageFile.from_image_id(self._config.imagesdir, base)
|
|
if files:
|
|
ret.append(build)
|
|
return ret
|
|
|
|
def _cleanupCurrentProviderUploads(self, provider, image, build_id):
|
|
'''
|
|
Remove cruft from a current build.
|
|
|
|
Current builds (the ones we want to keep) are treated special since
|
|
we want to remove any ZK nodes for uploads that failed exceptionally
|
|
hard (i.e., we could not set the state to FAILED and they remain as
|
|
UPLOADING), and we also want to remove any uploads that have been
|
|
marked for deleting.
|
|
'''
|
|
cruft = self._zk.getUploads(image, build_id, provider,
|
|
states=[zk.UPLOADING,
|
|
zk.DELETING,
|
|
zk.FAILED])
|
|
for upload in cruft:
|
|
if (upload.state == zk.UPLOADING and
|
|
not self._inProgressUpload(upload)
|
|
):
|
|
# Since we cache the uploads above, we need to verify the
|
|
# state hasn't changed on us (e.g., it could have gone from
|
|
# an in progress upload to a successfully completed upload
|
|
# between the getUploads() and the _inProgressUpload() check.
|
|
u = self._zk.getImageUpload(image, build_id, provider,
|
|
upload.id)
|
|
if upload.state != u.state:
|
|
continue
|
|
self.log.debug("Removing failed upload record: %s" % upload)
|
|
self._zk.deleteUpload(image, build_id, provider, upload.id)
|
|
elif upload.state == zk.DELETING:
|
|
self.log.debug(
|
|
"Removing deleted upload and record: %s" % upload)
|
|
self._deleteUpload(upload)
|
|
elif upload.state == zk.FAILED:
|
|
self.log.debug(
|
|
"Removing failed upload and record: %s" % upload)
|
|
self._deleteUpload(upload)
|
|
|
|
def _cleanupImage(self, known_providers, image):
|
|
'''
|
|
Clean up one image.
|
|
'''
|
|
# Get the list of all builds, then work from that so that we
|
|
# have a consistent view of the data.
|
|
all_builds = self._zk.getBuilds(image)
|
|
builds_to_keep = set([b for b in sorted(all_builds, reverse=True,
|
|
key=lambda y: y.state_time)
|
|
if b.state == zk.READY][:2])
|
|
local_builds = set(self._filterLocalBuilds(image, all_builds))
|
|
diskimage = self._config.diskimages.get(image)
|
|
if not diskimage and not local_builds:
|
|
# This builder is and was not responsible for this image,
|
|
# so ignore it.
|
|
return
|
|
# Remove any local builds that are not in use.
|
|
if not diskimage or (diskimage and not diskimage.image_types):
|
|
builds_to_keep -= local_builds
|
|
# TODO(jeblair): When all builds for an image which is not
|
|
# in use are deleted, the image znode should be deleted as
|
|
# well.
|
|
|
|
for build in all_builds:
|
|
# Start by deleting any uploads that are no longer needed
|
|
# because this image has been removed from a provider
|
|
# (since this should be done regardless of the build
|
|
# state).
|
|
for provider in known_providers:
|
|
if not provider.manage_images:
|
|
# This provider doesn't manage images
|
|
continue
|
|
try:
|
|
self._cleanupObsoleteProviderUploads(provider, image,
|
|
build.id)
|
|
if build in builds_to_keep:
|
|
self._cleanupCurrentProviderUploads(provider.name,
|
|
image,
|
|
build.id)
|
|
except Exception:
|
|
self.log.exception("Exception cleaning up uploads "
|
|
"of build %s of image %s in "
|
|
"provider %s:",
|
|
build, image, provider)
|
|
|
|
# If the build is in the delete state, we will try to
|
|
# delete the entire thing regardless.
|
|
if build.state != zk.DELETING:
|
|
# If it is in any other state, we will only delete it
|
|
# if it is older than the most recent two ready
|
|
# builds, or is in the building state but not actually
|
|
# building.
|
|
if build in builds_to_keep:
|
|
continue
|
|
elif self._inProgressBuild(build, image):
|
|
continue
|
|
|
|
for provider in known_providers:
|
|
if not provider.manage_images:
|
|
# This provider doesn't manage images
|
|
continue
|
|
try:
|
|
self._cleanupProvider(provider, image, build.id)
|
|
except Exception:
|
|
self.log.exception("Exception cleaning up build %s "
|
|
"of image %s in provider %s:",
|
|
build, image, provider)
|
|
|
|
uploads_exist = False
|
|
for p in self._zk.getBuildProviders(image, build.id):
|
|
if self._zk.getImageUploadNumbers(image, build.id, p):
|
|
uploads_exist = True
|
|
break
|
|
|
|
if not uploads_exist:
|
|
if build.state != zk.DELETING:
|
|
with self._zk.imageBuildNumberLock(
|
|
image, build.id, blocking=False
|
|
):
|
|
build.state = zk.DELETING
|
|
self._zk.storeBuild(image, build, build.id)
|
|
|
|
# Release the lock here so we can delete the build znode
|
|
if self._deleteLocalBuild(image, build):
|
|
if not self._zk.deleteBuild(image, build.id):
|
|
self.log.error("Unable to delete build %s because"
|
|
" uploads still remain.", build)
|
|
|
|
def run(self):
|
|
'''
|
|
Start point for the CleanupWorker thread.
|
|
'''
|
|
self._running = True
|
|
while self._running:
|
|
# Don't do work if we've lost communication with the ZK cluster
|
|
did_suspend = False
|
|
while self._zk and (self._zk.suspended or self._zk.lost):
|
|
did_suspend = True
|
|
self.log.info("ZooKeeper suspended. Waiting")
|
|
time.sleep(SUSPEND_WAIT_TIME)
|
|
if did_suspend:
|
|
self.log.info("ZooKeeper available. Resuming")
|
|
|
|
try:
|
|
self._run()
|
|
except Exception:
|
|
self.log.exception("Exception in CleanupWorker:")
|
|
time.sleep(10)
|
|
|
|
self._stop_event.wait(self._interval)
|
|
|
|
provider_manager.ProviderManager.stopProviders(self._config)
|
|
|
|
def _run(self):
|
|
'''
|
|
Body of run method for exception handling purposes.
|
|
'''
|
|
new_config = nodepool_config.loadConfig(self._config_path)
|
|
if self._secure_path:
|
|
nodepool_config.loadSecureConfig(new_config, self._secure_path)
|
|
if not self._config:
|
|
self._config = new_config
|
|
|
|
self._checkForZooKeeperChanges(new_config)
|
|
provider_manager.ProviderManager.reconfigure(self._config, new_config,
|
|
self._zk,
|
|
use_taskmanager=False,
|
|
only_image_manager=True)
|
|
self._config = new_config
|
|
|
|
self._cleanup()
|
|
|
|
|
|
class BuildWorker(BaseWorker):
|
|
def __init__(self, name, builder_id, config_path, secure_path,
|
|
interval, zk, dib_cmd):
|
|
super(BuildWorker, self).__init__(builder_id, config_path, secure_path,
|
|
interval, zk)
|
|
self.log = logging.getLogger("nodepool.builder.BuildWorker.%s" % name)
|
|
self.name = 'BuildWorker.%s' % name
|
|
self.dib_cmd = dib_cmd
|
|
|
|
def _getBuildLogRoot(self, name):
|
|
log_dir = self._config.build_log_dir
|
|
if not log_dir:
|
|
log_dir = '/var/log/nodepool/builds'
|
|
if not os.path.exists(log_dir):
|
|
os.makedirs(log_dir)
|
|
return log_dir
|
|
|
|
def _pruneBuildLogs(self, name):
|
|
log_dir = self._getBuildLogRoot(name)
|
|
keep = max(self._config.build_log_retention, 1)
|
|
existing = sorted(os.listdir(log_dir))
|
|
existing = [f for f in existing if f.startswith(name)]
|
|
delete = existing[:0 - keep]
|
|
for f in delete:
|
|
fp = os.path.join(log_dir, f)
|
|
self.log.info("Deleting old build log %s" % (fp,))
|
|
os.unlink(fp)
|
|
|
|
def _getBuildLog(self, name, build_id):
|
|
log_dir = self._getBuildLogRoot(name)
|
|
return os.path.join(log_dir, '%s-%s.log' % (name, build_id))
|
|
|
|
def _checkForScheduledImageUpdates(self):
|
|
'''
|
|
Check every DIB image to see if it has aged out and needs rebuilt.
|
|
'''
|
|
for diskimage in self._config.diskimages.values():
|
|
# Check if we've been told to shutdown
|
|
# or if ZK connection is suspended
|
|
if not self._running or self._zk.suspended or self._zk.lost:
|
|
return
|
|
try:
|
|
self._checkImageForScheduledImageUpdates(diskimage)
|
|
except Exception:
|
|
self.log.exception("Exception checking for scheduled "
|
|
"update of diskimage %s",
|
|
diskimage.name)
|
|
|
|
def _checkImageForScheduledImageUpdates(self, diskimage):
|
|
'''
|
|
Check one DIB image to see if it needs to be rebuilt.
|
|
|
|
.. note:: It's important to lock the image build before we check
|
|
the state time and then build to eliminate any race condition.
|
|
'''
|
|
# Check if diskimage builds are paused.
|
|
if diskimage.pause:
|
|
return
|
|
|
|
if not diskimage.image_types:
|
|
# We don't know what formats to build.
|
|
return
|
|
|
|
now = int(time.time())
|
|
builds = self._zk.getMostRecentBuilds(1, diskimage.name, zk.READY)
|
|
|
|
# If there is no build for this image, or it has aged out
|
|
# or if the current build is missing an image type from
|
|
# the config file, start a new build.
|
|
if (not builds
|
|
or (now - builds[0].state_time) >= diskimage.rebuild_age
|
|
or not set(builds[0].formats).issuperset(diskimage.image_types)
|
|
):
|
|
|
|
try:
|
|
with self._zk.imageBuildLock(diskimage.name, blocking=False):
|
|
# To avoid locking each image repeatedly, we have an
|
|
# second, redundant check here to verify that a new
|
|
# build didn't appear between the first check and the
|
|
# lock acquisition. If it's not the same build as
|
|
# identified in the first check above, assume another
|
|
# BuildWorker created the build for us and continue.
|
|
builds2 = self._zk.getMostRecentBuilds(
|
|
1, diskimage.name, zk.READY)
|
|
if builds2 and builds[0].id != builds2[0].id:
|
|
return
|
|
|
|
self.log.info("Building image %s" % diskimage.name)
|
|
|
|
data = zk.ImageBuild()
|
|
data.state = zk.BUILDING
|
|
data.builder_id = self._builder_id
|
|
data.builder = self._hostname
|
|
data.formats = list(diskimage.image_types)
|
|
|
|
bnum = self._zk.storeBuild(diskimage.name, data)
|
|
data = self._buildImage(bnum, diskimage)
|
|
self._zk.storeBuild(diskimage.name, data, bnum)
|
|
except exceptions.ZKLockException:
|
|
# Lock is already held. Skip it.
|
|
pass
|
|
|
|
def _checkForManualBuildRequest(self):
|
|
'''
|
|
Query ZooKeeper for any manual image build requests.
|
|
'''
|
|
for diskimage in self._config.diskimages.values():
|
|
# Check if we've been told to shutdown
|
|
# or if ZK connection is suspended
|
|
if not self._running or self._zk.suspended or self._zk.lost:
|
|
return
|
|
try:
|
|
self._checkImageForManualBuildRequest(diskimage)
|
|
except Exception:
|
|
self.log.exception("Exception checking for manual "
|
|
"update of diskimage %s",
|
|
diskimage)
|
|
|
|
def _checkImageForManualBuildRequest(self, diskimage):
|
|
'''
|
|
Query ZooKeeper for a manual image build request for one image.
|
|
'''
|
|
# Check if diskimage builds are paused.
|
|
if diskimage.pause:
|
|
return
|
|
|
|
# Reduce use of locks by adding an initial check here and
|
|
# a redundant check after lock acquisition.
|
|
if not self._zk.hasBuildRequest(diskimage.name):
|
|
return
|
|
|
|
try:
|
|
with self._zk.imageBuildLock(diskimage.name, blocking=False):
|
|
# Redundant check
|
|
if not self._zk.hasBuildRequest(diskimage.name):
|
|
return
|
|
|
|
self.log.info(
|
|
"Manual build request for image %s" % diskimage.name)
|
|
|
|
data = zk.ImageBuild()
|
|
data.state = zk.BUILDING
|
|
data.builder_id = self._builder_id
|
|
data.builder = self._hostname
|
|
data.formats = list(diskimage.image_types)
|
|
|
|
bnum = self._zk.storeBuild(diskimage.name, data)
|
|
data = self._buildImage(bnum, diskimage)
|
|
self._zk.storeBuild(diskimage.name, data, bnum)
|
|
|
|
# Remove request on a successful build
|
|
if data.state == zk.READY:
|
|
self._zk.removeBuildRequest(diskimage.name)
|
|
|
|
except exceptions.ZKLockException:
|
|
# Lock is already held. Skip it.
|
|
pass
|
|
|
|
def _buildImage(self, build_id, diskimage):
|
|
'''
|
|
Run the external command to build the diskimage.
|
|
|
|
:param str build_id: The ID for the build (used in image filename).
|
|
:param diskimage: The diskimage as retrieved from our config file.
|
|
|
|
:returns: An ImageBuild object of build-related data.
|
|
|
|
:raises: BuilderError if we failed to execute the build command.
|
|
'''
|
|
base = "-".join([diskimage.name, build_id])
|
|
image_file = DibImageFile(base)
|
|
filename = image_file.to_path(self._config.imagesdir, False)
|
|
|
|
env = os.environ.copy()
|
|
env['DIB_RELEASE'] = diskimage.release
|
|
env['DIB_IMAGE_NAME'] = diskimage.name
|
|
env['DIB_IMAGE_FILENAME'] = filename
|
|
|
|
# Note we use a reference to the nodepool config here so
|
|
# that whenever the config is updated we get up to date
|
|
# values in this thread.
|
|
if self._config.elementsdir:
|
|
env['ELEMENTS_PATH'] = self._config.elementsdir
|
|
|
|
# send additional env vars if needed
|
|
for k, v in diskimage.env_vars.items():
|
|
env[k] = v
|
|
|
|
img_elements = diskimage.elements
|
|
img_types = ",".join(diskimage.image_types)
|
|
|
|
qemu_img_options = ''
|
|
if 'qcow2' in img_types:
|
|
qemu_img_options = DEFAULT_QEMU_IMAGE_COMPAT_OPTIONS
|
|
|
|
cmd = ('%s -x -t %s --checksum --no-tmpfs %s -o %s %s' %
|
|
(self.dib_cmd, img_types, qemu_img_options, filename,
|
|
img_elements))
|
|
|
|
self._pruneBuildLogs(diskimage.name)
|
|
log_fn = self._getBuildLog(diskimage.name, build_id)
|
|
|
|
self.log.info('Running %s' % (cmd,))
|
|
self.log.info('Logging to %s' % (log_fn,))
|
|
|
|
start_time = time.monotonic()
|
|
try:
|
|
p = subprocess.Popen(
|
|
shlex.split(cmd),
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
env=env)
|
|
except OSError as e:
|
|
raise exceptions.BuilderError(
|
|
"Failed to exec '%s'. Error: '%s'" % (cmd, e.strerror)
|
|
)
|
|
|
|
with open(log_fn, 'wb') as log:
|
|
while True:
|
|
ln = p.stdout.readline()
|
|
log.write(ln)
|
|
log.flush()
|
|
if not ln:
|
|
break
|
|
|
|
rc = p.wait()
|
|
m = "Exit code: %s\n" % rc
|
|
log.write(m.encode('utf8'))
|
|
|
|
# It's possible the connection to the ZK cluster could have been
|
|
# interrupted during the build. If so, wait for it to return.
|
|
# It could transition directly from SUSPENDED to CONNECTED, or go
|
|
# through the LOST state before CONNECTED.
|
|
did_suspend = False
|
|
while self._zk.suspended or self._zk.lost:
|
|
did_suspend = True
|
|
self.log.info("ZooKeeper suspended during build. Waiting")
|
|
time.sleep(SUSPEND_WAIT_TIME)
|
|
if did_suspend:
|
|
self.log.info("ZooKeeper available. Resuming")
|
|
|
|
build_time = time.monotonic() - start_time
|
|
|
|
build_data = zk.ImageBuild()
|
|
build_data.builder_id = self._builder_id
|
|
build_data.builder = self._hostname
|
|
build_data.username = diskimage.username
|
|
|
|
if self._zk.didLoseConnection:
|
|
self.log.info("ZooKeeper lost while building %s" % diskimage.name)
|
|
self._zk.resetLostFlag()
|
|
build_data.state = zk.FAILED
|
|
elif p.returncode:
|
|
self.log.info(
|
|
"DIB failed creating %s (%s)" % (diskimage.name, p.returncode))
|
|
build_data.state = zk.FAILED
|
|
else:
|
|
self.log.info("DIB image %s is built" % diskimage.name)
|
|
build_data.state = zk.READY
|
|
build_data.formats = list(diskimage.image_types)
|
|
|
|
if self._statsd:
|
|
# record stats on the size of each image we create
|
|
for ext in img_types.split(','):
|
|
key = 'nodepool.dib_image_build.%s.%s.size' % (
|
|
diskimage.name, ext)
|
|
# A bit tricky because these image files may be sparse
|
|
# files; we only want the true size of the file for
|
|
# purposes of watching if we've added too much stuff
|
|
# into the image. Note that st_blocks is defined as
|
|
# 512-byte blocks by stat(2)
|
|
size = os.stat("%s.%s" % (filename, ext)).st_blocks * 512
|
|
self.log.debug("%s created image %s.%s (size: %d) " %
|
|
(diskimage.name, filename, ext, size))
|
|
self._statsd.gauge(key, size)
|
|
|
|
if self._statsd:
|
|
# report result to statsd
|
|
for ext in img_types.split(','):
|
|
key_base = 'nodepool.dib_image_build.%s.%s' % (
|
|
diskimage.name, ext)
|
|
self._statsd.gauge(key_base + '.rc', rc)
|
|
self._statsd.timing(key_base + '.duration',
|
|
int(build_time * 1000))
|
|
|
|
return build_data
|
|
|
|
def run(self):
|
|
'''
|
|
Start point for the BuildWorker thread.
|
|
'''
|
|
self._running = True
|
|
while self._running:
|
|
# Don't do work if we've lost communication with the ZK cluster
|
|
did_suspend = False
|
|
while self._zk and (self._zk.suspended or self._zk.lost):
|
|
did_suspend = True
|
|
self.log.info("ZooKeeper suspended. Waiting")
|
|
time.sleep(SUSPEND_WAIT_TIME)
|
|
if did_suspend:
|
|
self.log.info("ZooKeeper available. Resuming")
|
|
|
|
try:
|
|
self._run()
|
|
except Exception:
|
|
self.log.exception("Exception in BuildWorker:")
|
|
time.sleep(10)
|
|
|
|
self._stop_event.wait(self._interval)
|
|
|
|
def _run(self):
|
|
'''
|
|
Body of run method for exception handling purposes.
|
|
'''
|
|
# NOTE: For the first iteration, we expect self._config to be None
|
|
new_config = nodepool_config.loadConfig(self._config_path)
|
|
if self._secure_path:
|
|
nodepool_config.loadSecureConfig(new_config, self._secure_path)
|
|
if not self._config:
|
|
self._config = new_config
|
|
|
|
self._checkForZooKeeperChanges(new_config)
|
|
self._config = new_config
|
|
|
|
self._checkForScheduledImageUpdates()
|
|
self._checkForManualBuildRequest()
|
|
|
|
|
|
class UploadWorker(BaseWorker):
|
|
def __init__(self, name, builder_id, config_path, secure_path,
|
|
interval, zk):
|
|
super(UploadWorker, self).__init__(builder_id, config_path,
|
|
secure_path, interval, zk)
|
|
self.log = logging.getLogger("nodepool.builder.UploadWorker.%s" % name)
|
|
self.name = 'UploadWorker.%s' % name
|
|
|
|
def _reloadConfig(self):
|
|
'''
|
|
Reload the nodepool configuration file.
|
|
'''
|
|
new_config = nodepool_config.loadConfig(self._config_path)
|
|
if self._secure_path:
|
|
nodepool_config.loadSecureConfig(new_config, self._secure_path)
|
|
if not self._config:
|
|
self._config = new_config
|
|
|
|
self._checkForZooKeeperChanges(new_config)
|
|
provider_manager.ProviderManager.reconfigure(self._config, new_config,
|
|
self._zk,
|
|
use_taskmanager=False,
|
|
only_image_manager=True)
|
|
self._config = new_config
|
|
|
|
def _uploadImage(self, build_id, upload_id, image_name, images, provider,
|
|
username):
|
|
'''
|
|
Upload a local DIB image build to a provider.
|
|
|
|
:param str build_id: Unique ID of the image build to upload.
|
|
:param str upload_id: Unique ID of the upload.
|
|
:param str image_name: Name of the diskimage.
|
|
:param list images: A list of DibImageFile objects from this build
|
|
that available for uploading.
|
|
:param provider: The provider from the parsed config file.
|
|
:param username:
|
|
'''
|
|
start_time = time.time()
|
|
timestamp = int(start_time)
|
|
|
|
image = None
|
|
for i in images:
|
|
if provider.image_type == i.extension:
|
|
image = i
|
|
break
|
|
|
|
if not image:
|
|
raise exceptions.BuilderInvalidCommandError(
|
|
"Unable to find image file of type %s for id %s to upload" %
|
|
(provider.image_type, build_id)
|
|
)
|
|
|
|
self.log.debug("Found image file of type %s for image id: %s" %
|
|
(image.extension, image.image_id))
|
|
|
|
filename = image.to_path(self._config.imagesdir, with_extension=True)
|
|
|
|
ext_image_name = provider.image_name_format.format(
|
|
image_name=image_name, timestamp=str(timestamp)
|
|
)
|
|
|
|
self.log.info("Uploading DIB image build %s from %s to %s" %
|
|
(build_id, filename, provider.name))
|
|
|
|
manager = self._config.provider_managers[provider.name]
|
|
provider_image = provider.diskimages.get(image_name)
|
|
if provider_image is None:
|
|
raise exceptions.BuilderInvalidCommandError(
|
|
"Could not find matching provider image for %s" % image_name
|
|
)
|
|
|
|
meta = provider_image.meta.copy()
|
|
meta['nodepool_build_id'] = build_id
|
|
meta['nodepool_upload_id'] = upload_id
|
|
|
|
try:
|
|
external_id = manager.uploadImage(
|
|
ext_image_name, filename,
|
|
image_type=image.extension,
|
|
meta=meta,
|
|
md5=image.md5,
|
|
sha256=image.sha256,
|
|
)
|
|
except Exception:
|
|
self.log.exception("Failed to upload image %s to provider %s" %
|
|
(image_name, provider.name))
|
|
data = zk.ImageUpload()
|
|
data.state = zk.FAILED
|
|
return data
|
|
|
|
if self._statsd:
|
|
dt = int((time.time() - start_time) * 1000)
|
|
key = 'nodepool.image_update.%s.%s' % (image_name,
|
|
provider.name)
|
|
self._statsd.timing(key, dt)
|
|
self._statsd.incr(key)
|
|
|
|
base = "-".join([image_name, build_id])
|
|
self.log.info("Image build %s in %s is ready" %
|
|
(base, provider.name))
|
|
|
|
data = zk.ImageUpload()
|
|
data.state = zk.READY
|
|
data.external_id = external_id
|
|
data.external_name = ext_image_name
|
|
data.format = image.extension
|
|
data.username = username
|
|
|
|
return data
|
|
|
|
def _checkForProviderUploads(self):
|
|
'''
|
|
Check for any image builds that need to be uploaded to providers.
|
|
|
|
If we find any builds in the 'ready' state that haven't been uploaded
|
|
to providers, do the upload if they are available on the local disk.
|
|
'''
|
|
for provider in self._config.providers.values():
|
|
if not provider.manage_images:
|
|
continue
|
|
for image in provider.diskimages.values():
|
|
uploaded = False
|
|
|
|
# Check if we've been told to shutdown
|
|
# or if ZK connection is suspended
|
|
if not self._running or self._zk.suspended or self._zk.lost:
|
|
return
|
|
try:
|
|
uploaded = self._checkProviderImageUpload(provider, image)
|
|
except Exception:
|
|
self.log.exception("Error uploading image %s "
|
|
"to provider %s:",
|
|
image.name, provider.name)
|
|
|
|
# NOTE: Due to the configuration file disagreement issue
|
|
# (the copy we have may not be current), if we took the time
|
|
# to attempt to upload an image, let's short-circuit this loop
|
|
# to give us a chance to reload the configuration file.
|
|
if uploaded:
|
|
return
|
|
|
|
def _checkProviderImageUpload(self, provider, image):
|
|
'''
|
|
The main body of _checkForProviderUploads. This encapsulates
|
|
checking whether an image for a provider should be uploaded
|
|
and performing the upload. It is a separate function so that
|
|
exception handling can treat all provider-image uploads
|
|
indepedently.
|
|
|
|
:returns: True if an upload was attempted, False otherwise.
|
|
'''
|
|
# Check if image uploads are paused.
|
|
if provider.diskimages.get(image.name).pause:
|
|
return False
|
|
|
|
# Search for the most recent 'ready' image build
|
|
builds = self._zk.getMostRecentBuilds(1, image.name,
|
|
zk.READY)
|
|
if not builds:
|
|
return False
|
|
|
|
build = builds[0]
|
|
|
|
# Search for locally built images. The image name and build
|
|
# sequence ID is used to name the image.
|
|
local_images = DibImageFile.from_image_id(
|
|
self._config.imagesdir, "-".join([image.name, build.id]))
|
|
if not local_images:
|
|
return False
|
|
|
|
# See if this image has already been uploaded
|
|
upload = self._zk.getMostRecentBuildImageUploads(
|
|
1, image.name, build.id, provider.name, zk.READY)
|
|
if upload:
|
|
return False
|
|
|
|
# See if this provider supports the available image formats
|
|
if provider.image_type not in build.formats:
|
|
return False
|
|
|
|
try:
|
|
with self._zk.imageUploadLock(
|
|
image.name, build.id, provider.name,
|
|
blocking=False
|
|
):
|
|
# Verify once more that it hasn't been uploaded since the
|
|
# last check.
|
|
upload = self._zk.getMostRecentBuildImageUploads(
|
|
1, image.name, build.id, provider.name, zk.READY)
|
|
if upload:
|
|
return False
|
|
|
|
# NOTE: Due to the configuration file disagreement issue
|
|
# (the copy we have may not be current), we try to verify
|
|
# that another thread isn't trying to delete this build just
|
|
# before we upload.
|
|
b = self._zk.getBuild(image.name, build.id)
|
|
if b.state == zk.DELETING:
|
|
return False
|
|
|
|
# New upload number with initial state 'uploading'
|
|
data = zk.ImageUpload()
|
|
data.state = zk.UPLOADING
|
|
data.username = build.username
|
|
|
|
upnum = self._zk.storeImageUpload(
|
|
image.name, build.id, provider.name, data)
|
|
|
|
data = self._uploadImage(build.id, upnum, image.name,
|
|
local_images, provider,
|
|
build.username)
|
|
|
|
# Set final state
|
|
self._zk.storeImageUpload(image.name, build.id,
|
|
provider.name, data, upnum)
|
|
return True
|
|
except exceptions.ZKLockException:
|
|
# Lock is already held. Skip it.
|
|
return False
|
|
|
|
def run(self):
|
|
|
|
'''
|
|
Start point for the UploadWorker thread.
|
|
'''
|
|
self._running = True
|
|
while self._running:
|
|
# Don't do work if we've lost communication with the ZK cluster
|
|
did_suspend = False
|
|
while self._zk and (self._zk.suspended or self._zk.lost):
|
|
did_suspend = True
|
|
self.log.info("ZooKeeper suspended. Waiting")
|
|
time.sleep(SUSPEND_WAIT_TIME)
|
|
if did_suspend:
|
|
self.log.info("ZooKeeper available. Resuming")
|
|
|
|
try:
|
|
self._reloadConfig()
|
|
self._checkForProviderUploads()
|
|
except Exception:
|
|
self.log.exception("Exception in UploadWorker:")
|
|
time.sleep(10)
|
|
|
|
self._stop_event.wait(self._interval)
|
|
|
|
provider_manager.ProviderManager.stopProviders(self._config)
|
|
|
|
|
|
class NodePoolBuilder(object):
|
|
'''
|
|
Main class for the Nodepool Builder.
|
|
|
|
The builder has the responsibility to:
|
|
|
|
* Start and maintain the working state of each worker thread.
|
|
'''
|
|
log = logging.getLogger("nodepool.builder.NodePoolBuilder")
|
|
|
|
def __init__(self, config_path, secure_path=None,
|
|
num_builders=1, num_uploaders=4, fake=False):
|
|
'''
|
|
Initialize the NodePoolBuilder object.
|
|
|
|
:param str config_path: Path to configuration file.
|
|
:param str secure_path: Path to secure configuration file.
|
|
:param int num_builders: Number of build workers to start.
|
|
:param int num_uploaders: Number of upload workers to start.
|
|
:param bool fake: Whether to fake the image builds.
|
|
'''
|
|
self._config_path = config_path
|
|
self._secure_path = secure_path
|
|
self._config = None
|
|
self._num_builders = num_builders
|
|
self._build_workers = []
|
|
self._num_uploaders = num_uploaders
|
|
self._upload_workers = []
|
|
self._janitor = None
|
|
self._running = False
|
|
self.cleanup_interval = 60
|
|
self.build_interval = 10
|
|
self.upload_interval = 10
|
|
if fake:
|
|
self.dib_cmd = os.path.join(os.path.dirname(__file__), '..',
|
|
'nodepool/tests/fake-image-create')
|
|
else:
|
|
self.dib_cmd = 'disk-image-create'
|
|
self.zk = None
|
|
|
|
# This lock is needed because the run() method is started in a
|
|
# separate thread of control, which can return before the scheduler
|
|
# has completed startup. We need to avoid shutting down before the
|
|
# startup process has completed.
|
|
self._start_lock = threading.Lock()
|
|
|
|
# ======================================================================
|
|
# Private methods
|
|
# ======================================================================
|
|
|
|
def _getBuilderID(self, id_file):
|
|
if not os.path.exists(id_file):
|
|
with open(id_file, "w") as f:
|
|
builder_id = str(uuid.uuid4())
|
|
f.write(builder_id)
|
|
return builder_id
|
|
|
|
with open(id_file, "r") as f:
|
|
builder_id = f.read()
|
|
return builder_id
|
|
|
|
def _getAndValidateConfig(self):
|
|
config = nodepool_config.loadConfig(self._config_path)
|
|
if self._secure_path:
|
|
nodepool_config.loadSecureConfig(config, self._secure_path)
|
|
if not config.zookeeper_servers.values():
|
|
raise RuntimeError('No ZooKeeper servers specified in config.')
|
|
if not config.imagesdir:
|
|
raise RuntimeError('No images-dir specified in config.')
|
|
return config
|
|
|
|
# ======================================================================
|
|
# Public methods
|
|
# ======================================================================
|
|
|
|
def start(self):
|
|
'''
|
|
Start the builder.
|
|
|
|
The builder functionality is encapsulated within threads run
|
|
by the NodePoolBuilder. This starts the needed sub-threads
|
|
which will run forever until we tell them to stop.
|
|
'''
|
|
with self._start_lock:
|
|
if self._running:
|
|
raise exceptions.BuilderError('Cannot start, already running.')
|
|
|
|
self._config = self._getAndValidateConfig()
|
|
self._running = True
|
|
|
|
builder_id_file = os.path.join(self._config.imagesdir,
|
|
"builder_id.txt")
|
|
builder_id = self._getBuilderID(builder_id_file)
|
|
|
|
# All worker threads share a single ZooKeeper instance/connection.
|
|
self.zk = zk.ZooKeeper(enable_cache=False)
|
|
self.zk.connect(list(self._config.zookeeper_servers.values()))
|
|
|
|
self.log.debug('Starting listener for build jobs')
|
|
|
|
# Create build and upload worker objects
|
|
for i in range(self._num_builders):
|
|
w = BuildWorker(i, builder_id,
|
|
self._config_path, self._secure_path,
|
|
self.build_interval, self.zk, self.dib_cmd)
|
|
w.start()
|
|
self._build_workers.append(w)
|
|
|
|
for i in range(self._num_uploaders):
|
|
w = UploadWorker(i, builder_id,
|
|
self._config_path, self._secure_path,
|
|
self.upload_interval, self.zk)
|
|
w.start()
|
|
self._upload_workers.append(w)
|
|
|
|
if self.cleanup_interval > 0:
|
|
self._janitor = CleanupWorker(
|
|
0, builder_id,
|
|
self._config_path, self._secure_path,
|
|
self.cleanup_interval, self.zk)
|
|
self._janitor.start()
|
|
|
|
# Wait until all threads are running. Otherwise, we have a race
|
|
# on the worker _running attribute if shutdown() is called before
|
|
# run() actually begins.
|
|
workers = self._build_workers + self._upload_workers
|
|
if self._janitor:
|
|
workers += [self._janitor]
|
|
while not all([
|
|
x.running for x in (workers)]):
|
|
time.sleep(0)
|
|
|
|
def stop(self):
|
|
'''
|
|
Stop the builder.
|
|
|
|
Signal the sub threads to begin the shutdown process. We don't
|
|
want this method to return until the scheduler has successfully
|
|
stopped all of its own threads.
|
|
'''
|
|
with self._start_lock:
|
|
self.log.debug("Stopping. NodePoolBuilder shutting down workers")
|
|
# Note we do not add the upload workers to this list intentionally.
|
|
# The reason for this is that uploads can take many hours and there
|
|
# is no good way to stop the blocking writes performed by the
|
|
# uploads in order to join() below on a reasonable amount of time.
|
|
# Killing the process will stop the upload then both the record
|
|
# in zk and in the cloud will be deleted by any other running
|
|
# builders or when this builder starts again.
|
|
workers = self._build_workers
|
|
if self._janitor:
|
|
workers += [self._janitor]
|
|
for worker in (workers):
|
|
worker.shutdown()
|
|
|
|
self._running = False
|
|
|
|
self.log.debug('Waiting for jobs to complete')
|
|
|
|
# Do not exit until all of our owned threads exit.
|
|
for worker in (workers):
|
|
worker.join()
|
|
|
|
self.log.debug('Stopping providers')
|
|
provider_manager.ProviderManager.stopProviders(self._config)
|
|
|
|
self.log.debug('Terminating ZooKeeper connection')
|
|
self.zk.disconnect()
|
|
|
|
self.log.debug('Finished stopping')
|