Manage a pool of nodes for a distributed test infrastructure
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

builder.py 49KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293
  1. #!/usr/bin/env python
  2. # Copyright 2015 Hewlett-Packard Development Company, L.P.
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  5. # not use this file except in compliance with the License. You may obtain
  6. # a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13. # License for the specific language governing permissions and limitations
  14. # under the License.
  15. import logging
  16. import os
  17. import shutil
  18. import socket
  19. import subprocess
  20. import threading
  21. import time
  22. import shlex
  23. import uuid
  24. from nodepool import config as nodepool_config
  25. from nodepool import exceptions
  26. from nodepool import provider_manager
  27. from nodepool import stats
  28. from nodepool import zk
  29. MINS = 60
  30. HOURS = 60 * MINS
  31. # How long to wait for an image save
  32. IMAGE_TIMEOUT = 6 * HOURS
  33. # How long to wait between checks for ZooKeeper connectivity if it disappears.
  34. SUSPEND_WAIT_TIME = 30
  35. # HP Cloud requires qemu compat with 0.10. That version works elsewhere,
  36. # so just hardcode it for all qcow2 building
  37. DEFAULT_QEMU_IMAGE_COMPAT_OPTIONS = "--qemu-img-options 'compat=0.10'"
  38. class DibImageFile(object):
  39. '''
  40. Class used as an API to finding locally built DIB image files, and
  41. also used to represent the found files. Image files are named using
  42. a unique ID, but can be available in multiple formats (with different
  43. extensions).
  44. '''
  45. def __init__(self, image_id, extension=None):
  46. self.image_id = image_id
  47. self.extension = extension
  48. self.md5 = None
  49. self.md5_file = None
  50. self.sha256 = None
  51. self.sha256_file = None
  52. @staticmethod
  53. def from_path(path):
  54. image_file = os.path.basename(path)
  55. image_id, extension = image_file.rsplit('.', 1)
  56. return DibImageFile(image_id, extension)
  57. @staticmethod
  58. def from_image_id(images_dir, image_id):
  59. images = []
  60. for image_filename in os.listdir(images_dir):
  61. if os.path.isfile(os.path.join(images_dir, image_filename)):
  62. image = DibImageFile.from_path(image_filename)
  63. if image.image_id == image_id:
  64. images.append(image)
  65. return images
  66. @staticmethod
  67. def from_images_dir(images_dir):
  68. return [DibImageFile.from_path(x) for x in os.listdir(images_dir)]
  69. def to_path(self, images_dir, with_extension=True):
  70. my_path = os.path.join(images_dir, self.image_id)
  71. if with_extension:
  72. if self.extension is None:
  73. raise exceptions.BuilderError(
  74. 'Cannot specify image extension of None'
  75. )
  76. my_path += '.' + self.extension
  77. md5_path = '%s.%s' % (my_path, 'md5')
  78. md5 = self._checksum(md5_path)
  79. if md5:
  80. self.md5_file = md5_path
  81. self.md5 = md5[0:32]
  82. sha256_path = '%s.%s' % (my_path, 'sha256')
  83. sha256 = self._checksum(sha256_path)
  84. if sha256:
  85. self.sha256_file = sha256_path
  86. self.sha256 = sha256[0:64]
  87. return my_path
  88. def _checksum(self, filename):
  89. if not os.path.isfile(filename):
  90. return None
  91. with open(filename, 'r') as f:
  92. data = f.read()
  93. return data
  94. class BaseWorker(threading.Thread):
  95. def __init__(self, builder_id, config_path, secure_path, interval, zk):
  96. super(BaseWorker, self).__init__()
  97. self.log = logging.getLogger("nodepool.builder.BaseWorker")
  98. self.daemon = True
  99. self._running = False
  100. self._stop_event = threading.Event()
  101. self._config = None
  102. self._config_path = config_path
  103. self._secure_path = secure_path
  104. self._zk = zk
  105. self._hostname = socket.gethostname()
  106. self._statsd = stats.get_client()
  107. self._interval = interval
  108. self._builder_id = builder_id
  109. def _checkForZooKeeperChanges(self, new_config):
  110. '''
  111. Check config for ZooKeeper cluster changes.
  112. If the defined set of ZooKeeper servers changes, the connection
  113. will use the new server set.
  114. '''
  115. if self._config.zookeeper_servers != new_config.zookeeper_servers:
  116. self.log.debug("Detected ZooKeeper server changes")
  117. self._zk.resetHosts(list(new_config.zookeeper_servers.values()))
  118. @property
  119. def running(self):
  120. return self._running
  121. def shutdown(self):
  122. self._running = False
  123. self._stop_event.set()
  124. class CleanupWorker(BaseWorker):
  125. '''
  126. The janitor of nodepool-builder that will remove images from providers
  127. and any local DIB builds.
  128. '''
  129. def __init__(self, name, builder_id, config_path, secure_path,
  130. interval, zk):
  131. super(CleanupWorker, self).__init__(builder_id, config_path,
  132. secure_path, interval, zk)
  133. self.log = logging.getLogger(
  134. "nodepool.builder.CleanupWorker.%s" % name)
  135. self.name = 'CleanupWorker.%s' % name
  136. def _buildUploadRecencyTable(self):
  137. '''
  138. Builds a table for each image of the most recent uploads to each
  139. provider.
  140. Example)
  141. image1:
  142. providerA: [ (build_id, upload_id, upload_time), ... ]
  143. providerB: [ (build_id, upload_id, upload_time), ... ]
  144. image2:
  145. providerC: [ (build_id, upload_id, upload_time), ... ]
  146. '''
  147. self._rtable = {}
  148. for image in self._zk.getImageNames():
  149. self._rtable[image] = {}
  150. for build in self._zk.getBuilds(image, zk.READY):
  151. for provider in self._zk.getBuildProviders(image, build.id):
  152. if provider not in self._rtable[image]:
  153. self._rtable[image][provider] = []
  154. uploads = self._zk.getMostRecentBuildImageUploads(
  155. 2, image, build.id, provider, zk.READY)
  156. for upload in uploads:
  157. self._rtable[image][provider].append(
  158. (build.id, upload.id, upload.state_time)
  159. )
  160. # Sort uploads by state_time (upload time) and keep the 2 most recent
  161. for i in list(self._rtable.keys()):
  162. for p in self._rtable[i].keys():
  163. self._rtable[i][p].sort(key=lambda x: x[2], reverse=True)
  164. self._rtable[i][p] = self._rtable[i][p][:2]
  165. self.log.debug("Upload recency table: %s", self._rtable)
  166. def _isRecentUpload(self, image, provider, build_id, upload_id):
  167. '''
  168. Search for an upload for a build within the recency table.
  169. '''
  170. provider = self._rtable[image].get(provider)
  171. if not provider:
  172. return False
  173. for b_id, u_id, u_time in provider:
  174. if build_id == b_id and upload_id == u_id:
  175. return True
  176. return False
  177. def _inProgressUpload(self, upload):
  178. '''
  179. Determine if an upload is in progress.
  180. '''
  181. if upload.state != zk.UPLOADING:
  182. return False
  183. try:
  184. with self._zk.imageUploadLock(upload.image_name, upload.build_id,
  185. upload.provider_name,
  186. blocking=False):
  187. pass
  188. except exceptions.ZKLockException:
  189. return True
  190. return False
  191. def _removeDibItem(self, filename):
  192. if filename is None:
  193. return
  194. try:
  195. os.remove(filename)
  196. self.log.info("Removed DIB file %s" % filename)
  197. except OSError as e:
  198. if e.errno != 2: # No such file or directory
  199. raise e
  200. def _deleteLocalBuild(self, image, build):
  201. '''
  202. Remove expired image build from local disk.
  203. :param str image: Name of the image whose build we are deleting.
  204. :param ImageBuild build: The build we want to delete.
  205. :returns: True if files were deleted, False if none were found.
  206. '''
  207. base = "-".join([image, build.id])
  208. files = DibImageFile.from_image_id(self._config.imagesdir, base)
  209. if not files:
  210. # NOTE(pabelanger): It is possible we don't have any files because
  211. # diskimage-builder failed. So, check to see if we have the correct
  212. # builder so we can removed the data from zookeeper.
  213. # To maintain backward compatibility with builders that didn't
  214. # use unique builder IDs before, but do now, always compare to
  215. # hostname as well since some ZK data may still reference that.
  216. if (build.builder_id == self._builder_id or
  217. build.builder == self._hostname
  218. ):
  219. return True
  220. return False
  221. self.log.info("Doing cleanup for %s:%s" % (image, build.id))
  222. manifest_dir = None
  223. for f in files:
  224. filename = f.to_path(self._config.imagesdir, True)
  225. if not manifest_dir:
  226. path, ext = filename.rsplit('.', 1)
  227. manifest_dir = path + ".d"
  228. items = [filename, f.md5_file, f.sha256_file]
  229. list(map(self._removeDibItem, items))
  230. try:
  231. shutil.rmtree(manifest_dir)
  232. self.log.info("Removed DIB manifest %s" % manifest_dir)
  233. except OSError as e:
  234. if e.errno != 2: # No such file or directory
  235. raise e
  236. return True
  237. def _cleanupProvider(self, provider, image, build_id):
  238. all_uploads = self._zk.getUploads(image, build_id, provider.name)
  239. for upload in all_uploads:
  240. if self._isRecentUpload(image, provider.name, build_id, upload.id):
  241. continue
  242. self._deleteUpload(upload)
  243. def _cleanupObsoleteProviderUploads(self, provider, image, build_id):
  244. if image in provider.diskimages:
  245. # This image is in use for this provider
  246. return
  247. all_uploads = self._zk.getUploads(image, build_id, provider.name)
  248. for upload in all_uploads:
  249. self._deleteUpload(upload)
  250. def _deleteUpload(self, upload):
  251. deleted = False
  252. if upload.state != zk.DELETING:
  253. if not self._inProgressUpload(upload):
  254. data = zk.ImageUpload()
  255. data.state = zk.DELETING
  256. self._zk.storeImageUpload(upload.image_name, upload.build_id,
  257. upload.provider_name, data,
  258. upload.id)
  259. deleted = True
  260. if upload.state == zk.DELETING or deleted:
  261. manager = self._config.provider_managers[upload.provider_name]
  262. try:
  263. # It is possible we got this far, but don't actually have an
  264. # external_name. This could mean that zookeeper and cloud
  265. # provider are some how out of sync.
  266. if upload.external_name:
  267. base = "-".join([upload.image_name, upload.build_id])
  268. self.log.info("Deleting image build %s from %s" %
  269. (base, upload.provider_name))
  270. manager.deleteImage(upload.external_name)
  271. except Exception:
  272. self.log.exception(
  273. "Unable to delete image %s from %s:",
  274. upload.external_name, upload.provider_name)
  275. else:
  276. self.log.debug("Deleting image upload: %s", upload)
  277. self._zk.deleteUpload(upload.image_name, upload.build_id,
  278. upload.provider_name, upload.id)
  279. def _inProgressBuild(self, build, image):
  280. '''
  281. Determine if a DIB build is in progress.
  282. '''
  283. if build.state != zk.BUILDING:
  284. return False
  285. try:
  286. with self._zk.imageBuildLock(image, blocking=False):
  287. # An additional state check is needed to make sure it hasn't
  288. # changed on us. If it has, then let's pretend a build is
  289. # still in progress so that it is checked again later with
  290. # its new build state.
  291. b = self._zk.getBuild(image, build.id)
  292. if b.state != zk.BUILDING:
  293. return True
  294. pass
  295. except exceptions.ZKLockException:
  296. return True
  297. return False
  298. def _cleanup(self):
  299. '''
  300. Clean up builds on disk and in providers.
  301. '''
  302. known_providers = self._config.providers.values()
  303. image_names = self._zk.getImageNames()
  304. self._buildUploadRecencyTable()
  305. for image in image_names:
  306. try:
  307. self._cleanupImage(known_providers, image)
  308. except Exception:
  309. self.log.exception("Exception cleaning up image %s:", image)
  310. def _filterLocalBuilds(self, image, builds):
  311. '''Return the subset of builds that are local'''
  312. ret = []
  313. for build in builds:
  314. base = "-".join([image, build.id])
  315. files = DibImageFile.from_image_id(self._config.imagesdir, base)
  316. if files:
  317. ret.append(build)
  318. return ret
  319. def _cleanupCurrentProviderUploads(self, provider, image, build_id):
  320. '''
  321. Remove cruft from a current build.
  322. Current builds (the ones we want to keep) are treated special since
  323. we want to remove any ZK nodes for uploads that failed exceptionally
  324. hard (i.e., we could not set the state to FAILED and they remain as
  325. UPLOADING), and we also want to remove any uploads that have been
  326. marked for deleting.
  327. '''
  328. cruft = self._zk.getUploads(image, build_id, provider,
  329. states=[zk.UPLOADING,
  330. zk.DELETING,
  331. zk.FAILED])
  332. for upload in cruft:
  333. if (upload.state == zk.UPLOADING and
  334. not self._inProgressUpload(upload)
  335. ):
  336. # Since we cache the uploads above, we need to verify the
  337. # state hasn't changed on us (e.g., it could have gone from
  338. # an in progress upload to a successfully completed upload
  339. # between the getUploads() and the _inProgressUpload() check.
  340. u = self._zk.getImageUpload(image, build_id, provider,
  341. upload.id)
  342. if upload.state != u.state:
  343. continue
  344. self.log.debug("Removing failed upload record: %s" % upload)
  345. self._zk.deleteUpload(image, build_id, provider, upload.id)
  346. elif upload.state == zk.DELETING:
  347. self.log.debug(
  348. "Removing deleted upload and record: %s" % upload)
  349. self._deleteUpload(upload)
  350. elif upload.state == zk.FAILED:
  351. self.log.debug(
  352. "Removing failed upload and record: %s" % upload)
  353. self._deleteUpload(upload)
  354. def _cleanupImage(self, known_providers, image):
  355. '''
  356. Clean up one image.
  357. '''
  358. # Get the list of all builds, then work from that so that we
  359. # have a consistent view of the data.
  360. all_builds = self._zk.getBuilds(image)
  361. builds_to_keep = set([b for b in sorted(all_builds, reverse=True,
  362. key=lambda y: y.state_time)
  363. if b.state == zk.READY][:2])
  364. local_builds = set(self._filterLocalBuilds(image, all_builds))
  365. diskimage = self._config.diskimages.get(image)
  366. if not diskimage and not local_builds:
  367. # This builder is and was not responsible for this image,
  368. # so ignore it.
  369. return
  370. # Remove any local builds that are not in use.
  371. if not diskimage or (diskimage and not diskimage.image_types):
  372. builds_to_keep -= local_builds
  373. # TODO(jeblair): When all builds for an image which is not
  374. # in use are deleted, the image znode should be deleted as
  375. # well.
  376. for build in all_builds:
  377. # Start by deleting any uploads that are no longer needed
  378. # because this image has been removed from a provider
  379. # (since this should be done regardless of the build
  380. # state).
  381. for provider in known_providers:
  382. if not provider.manage_images:
  383. # This provider doesn't manage images
  384. continue
  385. try:
  386. self._cleanupObsoleteProviderUploads(provider, image,
  387. build.id)
  388. if build in builds_to_keep:
  389. self._cleanupCurrentProviderUploads(provider.name,
  390. image,
  391. build.id)
  392. except Exception:
  393. self.log.exception("Exception cleaning up uploads "
  394. "of build %s of image %s in "
  395. "provider %s:",
  396. build, image, provider)
  397. # If the build is in the delete state, we will try to
  398. # delete the entire thing regardless.
  399. if build.state != zk.DELETING:
  400. # If it is in any other state, we will only delete it
  401. # if it is older than the most recent two ready
  402. # builds, or is in the building state but not actually
  403. # building.
  404. if build in builds_to_keep:
  405. continue
  406. elif self._inProgressBuild(build, image):
  407. continue
  408. for provider in known_providers:
  409. if not provider.manage_images:
  410. # This provider doesn't manage images
  411. continue
  412. try:
  413. self._cleanupProvider(provider, image, build.id)
  414. except Exception:
  415. self.log.exception("Exception cleaning up build %s "
  416. "of image %s in provider %s:",
  417. build, image, provider)
  418. uploads_exist = False
  419. for p in self._zk.getBuildProviders(image, build.id):
  420. if self._zk.getImageUploadNumbers(image, build.id, p):
  421. uploads_exist = True
  422. break
  423. if not uploads_exist:
  424. if build.state != zk.DELETING:
  425. with self._zk.imageBuildNumberLock(
  426. image, build.id, blocking=False
  427. ):
  428. build.state = zk.DELETING
  429. self._zk.storeBuild(image, build, build.id)
  430. # Release the lock here so we can delete the build znode
  431. if self._deleteLocalBuild(image, build):
  432. if not self._zk.deleteBuild(image, build.id):
  433. self.log.error("Unable to delete build %s because"
  434. " uploads still remain.", build)
  435. def run(self):
  436. '''
  437. Start point for the CleanupWorker thread.
  438. '''
  439. self._running = True
  440. while self._running:
  441. # Don't do work if we've lost communication with the ZK cluster
  442. did_suspend = False
  443. while self._zk and (self._zk.suspended or self._zk.lost):
  444. did_suspend = True
  445. self.log.info("ZooKeeper suspended. Waiting")
  446. time.sleep(SUSPEND_WAIT_TIME)
  447. if did_suspend:
  448. self.log.info("ZooKeeper available. Resuming")
  449. try:
  450. self._run()
  451. except Exception:
  452. self.log.exception("Exception in CleanupWorker:")
  453. time.sleep(10)
  454. self._stop_event.wait(self._interval)
  455. provider_manager.ProviderManager.stopProviders(self._config)
  456. def _run(self):
  457. '''
  458. Body of run method for exception handling purposes.
  459. '''
  460. new_config = nodepool_config.loadConfig(self._config_path)
  461. if self._secure_path:
  462. nodepool_config.loadSecureConfig(new_config, self._secure_path)
  463. if not self._config:
  464. self._config = new_config
  465. self._checkForZooKeeperChanges(new_config)
  466. provider_manager.ProviderManager.reconfigure(self._config, new_config,
  467. self._zk,
  468. use_taskmanager=False,
  469. only_image_manager=True)
  470. self._config = new_config
  471. self._cleanup()
  472. class BuildWorker(BaseWorker):
  473. def __init__(self, name, builder_id, config_path, secure_path,
  474. interval, zk, dib_cmd):
  475. super(BuildWorker, self).__init__(builder_id, config_path, secure_path,
  476. interval, zk)
  477. self.log = logging.getLogger("nodepool.builder.BuildWorker.%s" % name)
  478. self.name = 'BuildWorker.%s' % name
  479. self.dib_cmd = dib_cmd
  480. def _getBuildLogRoot(self, name):
  481. log_dir = self._config.build_log_dir
  482. if not log_dir:
  483. log_dir = '/var/log/nodepool/builds'
  484. if not os.path.exists(log_dir):
  485. os.makedirs(log_dir)
  486. return log_dir
  487. def _pruneBuildLogs(self, name):
  488. log_dir = self._getBuildLogRoot(name)
  489. keep = max(self._config.build_log_retention, 1)
  490. existing = sorted(os.listdir(log_dir))
  491. existing = [f for f in existing if f.startswith(name)]
  492. delete = existing[:0 - keep]
  493. for f in delete:
  494. fp = os.path.join(log_dir, f)
  495. self.log.info("Deleting old build log %s" % (fp,))
  496. os.unlink(fp)
  497. def _getBuildLog(self, name, build_id):
  498. log_dir = self._getBuildLogRoot(name)
  499. return os.path.join(log_dir, '%s-%s.log' % (name, build_id))
  500. def _checkForScheduledImageUpdates(self):
  501. '''
  502. Check every DIB image to see if it has aged out and needs rebuilt.
  503. '''
  504. for diskimage in self._config.diskimages.values():
  505. # Check if we've been told to shutdown
  506. # or if ZK connection is suspended
  507. if not self._running or self._zk.suspended or self._zk.lost:
  508. return
  509. try:
  510. self._checkImageForScheduledImageUpdates(diskimage)
  511. except Exception:
  512. self.log.exception("Exception checking for scheduled "
  513. "update of diskimage %s",
  514. diskimage.name)
  515. def _checkImageForScheduledImageUpdates(self, diskimage):
  516. '''
  517. Check one DIB image to see if it needs to be rebuilt.
  518. .. note:: It's important to lock the image build before we check
  519. the state time and then build to eliminate any race condition.
  520. '''
  521. # Check if diskimage builds are paused.
  522. if diskimage.pause:
  523. return
  524. if not diskimage.image_types:
  525. # We don't know what formats to build.
  526. return
  527. now = int(time.time())
  528. builds = self._zk.getMostRecentBuilds(1, diskimage.name, zk.READY)
  529. # If there is no build for this image, or it has aged out
  530. # or if the current build is missing an image type from
  531. # the config file, start a new build.
  532. if (not builds
  533. or (now - builds[0].state_time) >= diskimage.rebuild_age
  534. or not set(builds[0].formats).issuperset(diskimage.image_types)
  535. ):
  536. try:
  537. with self._zk.imageBuildLock(diskimage.name, blocking=False):
  538. # To avoid locking each image repeatedly, we have an
  539. # second, redundant check here to verify that a new
  540. # build didn't appear between the first check and the
  541. # lock acquisition. If it's not the same build as
  542. # identified in the first check above, assume another
  543. # BuildWorker created the build for us and continue.
  544. builds2 = self._zk.getMostRecentBuilds(
  545. 1, diskimage.name, zk.READY)
  546. if builds2 and builds[0].id != builds2[0].id:
  547. return
  548. self.log.info("Building image %s" % diskimage.name)
  549. data = zk.ImageBuild()
  550. data.state = zk.BUILDING
  551. data.builder_id = self._builder_id
  552. data.builder = self._hostname
  553. data.formats = list(diskimage.image_types)
  554. bnum = self._zk.storeBuild(diskimage.name, data)
  555. data = self._buildImage(bnum, diskimage)
  556. self._zk.storeBuild(diskimage.name, data, bnum)
  557. except exceptions.ZKLockException:
  558. # Lock is already held. Skip it.
  559. pass
  560. def _checkForManualBuildRequest(self):
  561. '''
  562. Query ZooKeeper for any manual image build requests.
  563. '''
  564. for diskimage in self._config.diskimages.values():
  565. # Check if we've been told to shutdown
  566. # or if ZK connection is suspended
  567. if not self._running or self._zk.suspended or self._zk.lost:
  568. return
  569. try:
  570. self._checkImageForManualBuildRequest(diskimage)
  571. except Exception:
  572. self.log.exception("Exception checking for manual "
  573. "update of diskimage %s",
  574. diskimage)
  575. def _checkImageForManualBuildRequest(self, diskimage):
  576. '''
  577. Query ZooKeeper for a manual image build request for one image.
  578. '''
  579. # Check if diskimage builds are paused.
  580. if diskimage.pause:
  581. return
  582. # Reduce use of locks by adding an initial check here and
  583. # a redundant check after lock acquisition.
  584. if not self._zk.hasBuildRequest(diskimage.name):
  585. return
  586. try:
  587. with self._zk.imageBuildLock(diskimage.name, blocking=False):
  588. # Redundant check
  589. if not self._zk.hasBuildRequest(diskimage.name):
  590. return
  591. self.log.info(
  592. "Manual build request for image %s" % diskimage.name)
  593. data = zk.ImageBuild()
  594. data.state = zk.BUILDING
  595. data.builder_id = self._builder_id
  596. data.builder = self._hostname
  597. data.formats = list(diskimage.image_types)
  598. bnum = self._zk.storeBuild(diskimage.name, data)
  599. data = self._buildImage(bnum, diskimage)
  600. self._zk.storeBuild(diskimage.name, data, bnum)
  601. # Remove request on a successful build
  602. if data.state == zk.READY:
  603. self._zk.removeBuildRequest(diskimage.name)
  604. except exceptions.ZKLockException:
  605. # Lock is already held. Skip it.
  606. pass
  607. def _buildImage(self, build_id, diskimage):
  608. '''
  609. Run the external command to build the diskimage.
  610. :param str build_id: The ID for the build (used in image filename).
  611. :param diskimage: The diskimage as retrieved from our config file.
  612. :returns: An ImageBuild object of build-related data.
  613. :raises: BuilderError if we failed to execute the build command.
  614. '''
  615. base = "-".join([diskimage.name, build_id])
  616. image_file = DibImageFile(base)
  617. filename = image_file.to_path(self._config.imagesdir, False)
  618. env = os.environ.copy()
  619. env['DIB_RELEASE'] = diskimage.release
  620. env['DIB_IMAGE_NAME'] = diskimage.name
  621. env['DIB_IMAGE_FILENAME'] = filename
  622. # Note we use a reference to the nodepool config here so
  623. # that whenever the config is updated we get up to date
  624. # values in this thread.
  625. if self._config.elementsdir:
  626. env['ELEMENTS_PATH'] = self._config.elementsdir
  627. # send additional env vars if needed
  628. for k, v in diskimage.env_vars.items():
  629. env[k] = v
  630. img_elements = diskimage.elements
  631. img_types = ",".join(diskimage.image_types)
  632. qemu_img_options = ''
  633. if 'qcow2' in img_types:
  634. qemu_img_options = DEFAULT_QEMU_IMAGE_COMPAT_OPTIONS
  635. cmd = ('%s -x -t %s --checksum --no-tmpfs %s -o %s %s' %
  636. (self.dib_cmd, img_types, qemu_img_options, filename,
  637. img_elements))
  638. self._pruneBuildLogs(diskimage.name)
  639. log_fn = self._getBuildLog(diskimage.name, build_id)
  640. self.log.info('Running %s' % (cmd,))
  641. self.log.info('Logging to %s' % (log_fn,))
  642. start_time = time.monotonic()
  643. try:
  644. p = subprocess.Popen(
  645. shlex.split(cmd),
  646. stdout=subprocess.PIPE,
  647. stderr=subprocess.STDOUT,
  648. env=env)
  649. except OSError as e:
  650. raise exceptions.BuilderError(
  651. "Failed to exec '%s'. Error: '%s'" % (cmd, e.strerror)
  652. )
  653. with open(log_fn, 'wb') as log:
  654. while True:
  655. ln = p.stdout.readline()
  656. log.write(ln)
  657. log.flush()
  658. if not ln:
  659. break
  660. rc = p.wait()
  661. m = "Exit code: %s\n" % rc
  662. log.write(m.encode('utf8'))
  663. # It's possible the connection to the ZK cluster could have been
  664. # interrupted during the build. If so, wait for it to return.
  665. # It could transition directly from SUSPENDED to CONNECTED, or go
  666. # through the LOST state before CONNECTED.
  667. did_suspend = False
  668. while self._zk.suspended or self._zk.lost:
  669. did_suspend = True
  670. self.log.info("ZooKeeper suspended during build. Waiting")
  671. time.sleep(SUSPEND_WAIT_TIME)
  672. if did_suspend:
  673. self.log.info("ZooKeeper available. Resuming")
  674. build_time = time.monotonic() - start_time
  675. build_data = zk.ImageBuild()
  676. build_data.builder_id = self._builder_id
  677. build_data.builder = self._hostname
  678. build_data.username = diskimage.username
  679. if self._zk.didLoseConnection:
  680. self.log.info("ZooKeeper lost while building %s" % diskimage.name)
  681. self._zk.resetLostFlag()
  682. build_data.state = zk.FAILED
  683. elif p.returncode:
  684. self.log.info(
  685. "DIB failed creating %s (%s)" % (diskimage.name, p.returncode))
  686. build_data.state = zk.FAILED
  687. else:
  688. self.log.info("DIB image %s is built" % diskimage.name)
  689. build_data.state = zk.READY
  690. build_data.formats = list(diskimage.image_types)
  691. if self._statsd:
  692. # record stats on the size of each image we create
  693. for ext in img_types.split(','):
  694. key = 'nodepool.dib_image_build.%s.%s.size' % (
  695. diskimage.name, ext)
  696. # A bit tricky because these image files may be sparse
  697. # files; we only want the true size of the file for
  698. # purposes of watching if we've added too much stuff
  699. # into the image. Note that st_blocks is defined as
  700. # 512-byte blocks by stat(2)
  701. size = os.stat("%s.%s" % (filename, ext)).st_blocks * 512
  702. self.log.debug("%s created image %s.%s (size: %d) " %
  703. (diskimage.name, filename, ext, size))
  704. self._statsd.gauge(key, size)
  705. if self._statsd:
  706. # report result to statsd
  707. for ext in img_types.split(','):
  708. key_base = 'nodepool.dib_image_build.%s.%s' % (
  709. diskimage.name, ext)
  710. self._statsd.gauge(key_base + '.rc', rc)
  711. self._statsd.timing(key_base + '.duration',
  712. int(build_time * 1000))
  713. return build_data
  714. def run(self):
  715. '''
  716. Start point for the BuildWorker thread.
  717. '''
  718. self._running = True
  719. while self._running:
  720. # Don't do work if we've lost communication with the ZK cluster
  721. did_suspend = False
  722. while self._zk and (self._zk.suspended or self._zk.lost):
  723. did_suspend = True
  724. self.log.info("ZooKeeper suspended. Waiting")
  725. time.sleep(SUSPEND_WAIT_TIME)
  726. if did_suspend:
  727. self.log.info("ZooKeeper available. Resuming")
  728. try:
  729. self._run()
  730. except Exception:
  731. self.log.exception("Exception in BuildWorker:")
  732. time.sleep(10)
  733. self._stop_event.wait(self._interval)
  734. def _run(self):
  735. '''
  736. Body of run method for exception handling purposes.
  737. '''
  738. # NOTE: For the first iteration, we expect self._config to be None
  739. new_config = nodepool_config.loadConfig(self._config_path)
  740. if self._secure_path:
  741. nodepool_config.loadSecureConfig(new_config, self._secure_path)
  742. if not self._config:
  743. self._config = new_config
  744. self._checkForZooKeeperChanges(new_config)
  745. self._config = new_config
  746. self._checkForScheduledImageUpdates()
  747. self._checkForManualBuildRequest()
  748. class UploadWorker(BaseWorker):
  749. def __init__(self, name, builder_id, config_path, secure_path,
  750. interval, zk):
  751. super(UploadWorker, self).__init__(builder_id, config_path,
  752. secure_path, interval, zk)
  753. self.log = logging.getLogger("nodepool.builder.UploadWorker.%s" % name)
  754. self.name = 'UploadWorker.%s' % name
  755. def _reloadConfig(self):
  756. '''
  757. Reload the nodepool configuration file.
  758. '''
  759. new_config = nodepool_config.loadConfig(self._config_path)
  760. if self._secure_path:
  761. nodepool_config.loadSecureConfig(new_config, self._secure_path)
  762. if not self._config:
  763. self._config = new_config
  764. self._checkForZooKeeperChanges(new_config)
  765. provider_manager.ProviderManager.reconfigure(self._config, new_config,
  766. self._zk,
  767. use_taskmanager=False,
  768. only_image_manager=True)
  769. self._config = new_config
  770. def _uploadImage(self, build_id, upload_id, image_name, images, provider,
  771. username):
  772. '''
  773. Upload a local DIB image build to a provider.
  774. :param str build_id: Unique ID of the image build to upload.
  775. :param str upload_id: Unique ID of the upload.
  776. :param str image_name: Name of the diskimage.
  777. :param list images: A list of DibImageFile objects from this build
  778. that available for uploading.
  779. :param provider: The provider from the parsed config file.
  780. :param username:
  781. '''
  782. start_time = time.time()
  783. timestamp = int(start_time)
  784. image = None
  785. for i in images:
  786. if provider.image_type == i.extension:
  787. image = i
  788. break
  789. if not image:
  790. raise exceptions.BuilderInvalidCommandError(
  791. "Unable to find image file of type %s for id %s to upload" %
  792. (provider.image_type, build_id)
  793. )
  794. self.log.debug("Found image file of type %s for image id: %s" %
  795. (image.extension, image.image_id))
  796. filename = image.to_path(self._config.imagesdir, with_extension=True)
  797. ext_image_name = provider.image_name_format.format(
  798. image_name=image_name, timestamp=str(timestamp)
  799. )
  800. self.log.info("Uploading DIB image build %s from %s to %s" %
  801. (build_id, filename, provider.name))
  802. manager = self._config.provider_managers[provider.name]
  803. provider_image = provider.diskimages.get(image_name)
  804. if provider_image is None:
  805. raise exceptions.BuilderInvalidCommandError(
  806. "Could not find matching provider image for %s" % image_name
  807. )
  808. meta = provider_image.meta.copy()
  809. meta['nodepool_build_id'] = build_id
  810. meta['nodepool_upload_id'] = upload_id
  811. try:
  812. external_id = manager.uploadImage(
  813. ext_image_name, filename,
  814. image_type=image.extension,
  815. meta=meta,
  816. md5=image.md5,
  817. sha256=image.sha256,
  818. )
  819. except Exception:
  820. self.log.exception("Failed to upload image %s to provider %s" %
  821. (image_name, provider.name))
  822. data = zk.ImageUpload()
  823. data.state = zk.FAILED
  824. return data
  825. if self._statsd:
  826. dt = int((time.time() - start_time) * 1000)
  827. key = 'nodepool.image_update.%s.%s' % (image_name,
  828. provider.name)
  829. self._statsd.timing(key, dt)
  830. self._statsd.incr(key)
  831. base = "-".join([image_name, build_id])
  832. self.log.info("Image build %s in %s is ready" %
  833. (base, provider.name))
  834. data = zk.ImageUpload()
  835. data.state = zk.READY
  836. data.external_id = external_id
  837. data.external_name = ext_image_name
  838. data.format = image.extension
  839. data.username = username
  840. return data
  841. def _checkForProviderUploads(self):
  842. '''
  843. Check for any image builds that need to be uploaded to providers.
  844. If we find any builds in the 'ready' state that haven't been uploaded
  845. to providers, do the upload if they are available on the local disk.
  846. '''
  847. for provider in self._config.providers.values():
  848. if not provider.manage_images:
  849. continue
  850. for image in provider.diskimages.values():
  851. uploaded = False
  852. # Check if we've been told to shutdown
  853. # or if ZK connection is suspended
  854. if not self._running or self._zk.suspended or self._zk.lost:
  855. return
  856. try:
  857. uploaded = self._checkProviderImageUpload(provider, image)
  858. except Exception:
  859. self.log.exception("Error uploading image %s "
  860. "to provider %s:",
  861. image.name, provider.name)
  862. # NOTE: Due to the configuration file disagreement issue
  863. # (the copy we have may not be current), if we took the time
  864. # to attempt to upload an image, let's short-circuit this loop
  865. # to give us a chance to reload the configuration file.
  866. if uploaded:
  867. return
  868. def _checkProviderImageUpload(self, provider, image):
  869. '''
  870. The main body of _checkForProviderUploads. This encapsulates
  871. checking whether an image for a provider should be uploaded
  872. and performing the upload. It is a separate function so that
  873. exception handling can treat all provider-image uploads
  874. indepedently.
  875. :returns: True if an upload was attempted, False otherwise.
  876. '''
  877. # Check if image uploads are paused.
  878. if provider.diskimages.get(image.name).pause:
  879. return False
  880. # Search for the most recent 'ready' image build
  881. builds = self._zk.getMostRecentBuilds(1, image.name,
  882. zk.READY)
  883. if not builds:
  884. return False
  885. build = builds[0]
  886. # Search for locally built images. The image name and build
  887. # sequence ID is used to name the image.
  888. local_images = DibImageFile.from_image_id(
  889. self._config.imagesdir, "-".join([image.name, build.id]))
  890. if not local_images:
  891. return False
  892. # See if this image has already been uploaded
  893. upload = self._zk.getMostRecentBuildImageUploads(
  894. 1, image.name, build.id, provider.name, zk.READY)
  895. if upload:
  896. return False
  897. # See if this provider supports the available image formats
  898. if provider.image_type not in build.formats:
  899. return False
  900. try:
  901. with self._zk.imageUploadLock(
  902. image.name, build.id, provider.name,
  903. blocking=False
  904. ):
  905. # Verify once more that it hasn't been uploaded since the
  906. # last check.
  907. upload = self._zk.getMostRecentBuildImageUploads(
  908. 1, image.name, build.id, provider.name, zk.READY)
  909. if upload:
  910. return False
  911. # NOTE: Due to the configuration file disagreement issue
  912. # (the copy we have may not be current), we try to verify
  913. # that another thread isn't trying to delete this build just
  914. # before we upload.
  915. b = self._zk.getBuild(image.name, build.id)
  916. if b.state == zk.DELETING:
  917. return False
  918. # New upload number with initial state 'uploading'
  919. data = zk.ImageUpload()
  920. data.state = zk.UPLOADING
  921. data.username = build.username
  922. upnum = self._zk.storeImageUpload(
  923. image.name, build.id, provider.name, data)
  924. data = self._uploadImage(build.id, upnum, image.name,
  925. local_images, provider,
  926. build.username)
  927. # Set final state
  928. self._zk.storeImageUpload(image.name, build.id,
  929. provider.name, data, upnum)
  930. return True
  931. except exceptions.ZKLockException:
  932. # Lock is already held. Skip it.
  933. return False
  934. def run(self):
  935. '''
  936. Start point for the UploadWorker thread.
  937. '''
  938. self._running = True
  939. while self._running:
  940. # Don't do work if we've lost communication with the ZK cluster
  941. did_suspend = False
  942. while self._zk and (self._zk.suspended or self._zk.lost):
  943. did_suspend = True
  944. self.log.info("ZooKeeper suspended. Waiting")
  945. time.sleep(SUSPEND_WAIT_TIME)
  946. if did_suspend:
  947. self.log.info("ZooKeeper available. Resuming")
  948. try:
  949. self._reloadConfig()
  950. self._checkForProviderUploads()
  951. except Exception:
  952. self.log.exception("Exception in UploadWorker:")
  953. time.sleep(10)
  954. self._stop_event.wait(self._interval)
  955. provider_manager.ProviderManager.stopProviders(self._config)
  956. class NodePoolBuilder(object):
  957. '''
  958. Main class for the Nodepool Builder.
  959. The builder has the responsibility to:
  960. * Start and maintain the working state of each worker thread.
  961. '''
  962. log = logging.getLogger("nodepool.builder.NodePoolBuilder")
  963. def __init__(self, config_path, secure_path=None,
  964. num_builders=1, num_uploaders=4, fake=False):
  965. '''
  966. Initialize the NodePoolBuilder object.
  967. :param str config_path: Path to configuration file.
  968. :param str secure_path: Path to secure configuration file.
  969. :param int num_builders: Number of build workers to start.
  970. :param int num_uploaders: Number of upload workers to start.
  971. :param bool fake: Whether to fake the image builds.
  972. '''
  973. self._config_path = config_path
  974. self._secure_path = secure_path
  975. self._config = None
  976. self._num_builders = num_builders
  977. self._build_workers = []
  978. self._num_uploaders = num_uploaders
  979. self._upload_workers = []
  980. self._janitor = None
  981. self._running = False
  982. self.cleanup_interval = 60
  983. self.build_interval = 10
  984. self.upload_interval = 10
  985. if fake:
  986. self.dib_cmd = os.path.join(os.path.dirname(__file__), '..',
  987. 'nodepool/tests/fake-image-create')
  988. else:
  989. self.dib_cmd = 'disk-image-create'
  990. self.zk = None
  991. # This lock is needed because the run() method is started in a
  992. # separate thread of control, which can return before the scheduler
  993. # has completed startup. We need to avoid shutting down before the
  994. # startup process has completed.
  995. self._start_lock = threading.Lock()
  996. # ======================================================================
  997. # Private methods
  998. # ======================================================================
  999. def _getBuilderID(self, id_file):
  1000. if not os.path.exists(id_file):
  1001. with open(id_file, "w") as f:
  1002. builder_id = str(uuid.uuid4())
  1003. f.write(builder_id)
  1004. return builder_id
  1005. with open(id_file, "r") as f:
  1006. builder_id = f.read()
  1007. return builder_id
  1008. def _getAndValidateConfig(self):
  1009. config = nodepool_config.loadConfig(self._config_path)
  1010. if self._secure_path:
  1011. nodepool_config.loadSecureConfig(config, self._secure_path)
  1012. if not config.zookeeper_servers.values():
  1013. raise RuntimeError('No ZooKeeper servers specified in config.')
  1014. if not config.imagesdir:
  1015. raise RuntimeError('No images-dir specified in config.')
  1016. return config
  1017. # ======================================================================
  1018. # Public methods
  1019. # ======================================================================
  1020. def start(self):
  1021. '''
  1022. Start the builder.
  1023. The builder functionality is encapsulated within threads run
  1024. by the NodePoolBuilder. This starts the needed sub-threads
  1025. which will run forever until we tell them to stop.
  1026. '''
  1027. with self._start_lock:
  1028. if self._running:
  1029. raise exceptions.BuilderError('Cannot start, already running.')
  1030. self._config = self._getAndValidateConfig()
  1031. self._running = True
  1032. builder_id_file = os.path.join(self._config.imagesdir,
  1033. "builder_id.txt")
  1034. builder_id = self._getBuilderID(builder_id_file)
  1035. # All worker threads share a single ZooKeeper instance/connection.
  1036. self.zk = zk.ZooKeeper(enable_cache=False)
  1037. self.zk.connect(list(self._config.zookeeper_servers.values()))
  1038. self.log.debug('Starting listener for build jobs')
  1039. # Create build and upload worker objects
  1040. for i in range(self._num_builders):
  1041. w = BuildWorker(i, builder_id,
  1042. self._config_path, self._secure_path,
  1043. self.build_interval, self.zk, self.dib_cmd)
  1044. w.start()
  1045. self._build_workers.append(w)
  1046. for i in range(self._num_uploaders):
  1047. w = UploadWorker(i, builder_id,
  1048. self._config_path, self._secure_path,
  1049. self.upload_interval, self.zk)
  1050. w.start()
  1051. self._upload_workers.append(w)
  1052. if self.cleanup_interval > 0:
  1053. self._janitor = CleanupWorker(
  1054. 0, builder_id,
  1055. self._config_path, self._secure_path,
  1056. self.cleanup_interval, self.zk)
  1057. self._janitor.start()
  1058. # Wait until all threads are running. Otherwise, we have a race
  1059. # on the worker _running attribute if shutdown() is called before
  1060. # run() actually begins.
  1061. workers = self._build_workers + self._upload_workers
  1062. if self._janitor:
  1063. workers += [self._janitor]
  1064. while not all([
  1065. x.running for x in (workers)]):
  1066. time.sleep(0)
  1067. def stop(self):
  1068. '''
  1069. Stop the builder.
  1070. Signal the sub threads to begin the shutdown process. We don't
  1071. want this method to return until the scheduler has successfully
  1072. stopped all of its own threads.
  1073. '''
  1074. with self._start_lock:
  1075. self.log.debug("Stopping. NodePoolBuilder shutting down workers")
  1076. # Note we do not add the upload workers to this list intentionally.
  1077. # The reason for this is that uploads can take many hours and there
  1078. # is no good way to stop the blocking writes performed by the
  1079. # uploads in order to join() below on a reasonable amount of time.
  1080. # Killing the process will stop the upload then both the record
  1081. # in zk and in the cloud will be deleted by any other running
  1082. # builders or when this builder starts again.
  1083. workers = self._build_workers
  1084. if self._janitor:
  1085. workers += [self._janitor]
  1086. for worker in (workers):
  1087. worker.shutdown()
  1088. self._running = False
  1089. self.log.debug('Waiting for jobs to complete')
  1090. # Do not exit until all of our owned threads exit.
  1091. for worker in (workers):
  1092. worker.join()
  1093. self.log.debug('Stopping providers')
  1094. provider_manager.ProviderManager.stopProviders(self._config)
  1095. self.log.debug('Terminating ZooKeeper connection')
  1096. self.zk.disconnect()
  1097. self.log.debug('Finished stopping')