1 # Copyright 2015 gRPC authors. 2 # 3 # Licensed under the Apache License, Version 2.0 (the "License"); 4 # you may not use this file except in compliance with the License. 5 # You may obtain a copy of the License at 6 # 7 # http://www.apache.org/licenses/LICENSE-2.0 8 # 9 # Unless required by applicable law or agreed to in writing, software 10 # distributed under the License is distributed on an "AS IS" BASIS, 11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 # See the License for the specific language governing permissions and 13 # limitations under the License. 14 """Helpers to run docker instances as jobs.""" 15 16 from __future__ import print_function 17 18 import tempfile 19 import time 20 import uuid 21 import os 22 import subprocess 23 24 import jobset 25 26 _DEVNULL = open(os.devnull, 'w') 27 28 29 def random_name(base_name): 30 """Randomizes given base name.""" 31 return '%s_%s' % (base_name, uuid.uuid4()) 32 33 34 def docker_kill(cid): 35 """Kills a docker container. Returns True if successful.""" 36 return subprocess.call( 37 ['docker', 'kill', str(cid)], 38 stdin=subprocess.PIPE, 39 stdout=_DEVNULL, 40 stderr=subprocess.STDOUT) == 0 41 42 43 def docker_mapped_port(cid, port, timeout_seconds=15): 44 """Get port mapped to internal given internal port for given container.""" 45 started = time.time() 46 while time.time() - started < timeout_seconds: 47 try: 48 output = subprocess.check_output( 49 'docker port %s %s' % (cid, port), stderr=_DEVNULL, shell=True) 50 return int(output.split(':', 2)[1]) 51 except subprocess.CalledProcessError as e: 52 pass 53 raise Exception('Failed to get exposed port %s for container %s.' % (port, 54 cid)) 55 56 57 def wait_for_healthy(cid, shortname, timeout_seconds): 58 """Wait timeout_seconds for the container to become healthy""" 59 started = time.time() 60 while time.time() - started < timeout_seconds: 61 try: 62 output = subprocess.check_output( 63 [ 64 'docker', 'inspect', '--format="{{.State.Health.Status}}"', 65 cid 66 ], 67 stderr=_DEVNULL) 68 if output.strip('\n') == 'healthy': 69 return 70 except subprocess.CalledProcessError as e: 71 pass 72 time.sleep(1) 73 raise Exception('Timed out waiting for %s (%s) to pass health check' % 74 (shortname, cid)) 75 76 77 def finish_jobs(jobs): 78 """Kills given docker containers and waits for corresponding jobs to finish""" 79 for job in jobs: 80 job.kill(suppress_failure=True) 81 82 while any(job.is_running() for job in jobs): 83 time.sleep(1) 84 85 86 def image_exists(image): 87 """Returns True if given docker image exists.""" 88 return subprocess.call( 89 ['docker', 'inspect', image], 90 stdin=subprocess.PIPE, 91 stdout=_DEVNULL, 92 stderr=subprocess.STDOUT) == 0 93 94 95 def remove_image(image, skip_nonexistent=False, max_retries=10): 96 """Attempts to remove docker image with retries.""" 97 if skip_nonexistent and not image_exists(image): 98 return True 99 for attempt in range(0, max_retries): 100 if subprocess.call( 101 ['docker', 'rmi', '-f', image], 102 stdin=subprocess.PIPE, 103 stdout=_DEVNULL, 104 stderr=subprocess.STDOUT) == 0: 105 return True 106 time.sleep(2) 107 print('Failed to remove docker image %s' % image) 108 return False 109 110 111 class DockerJob: 112 """Encapsulates a job""" 113 114 def __init__(self, spec): 115 self._spec = spec 116 self._job = jobset.Job( 117 spec, newline_on_success=True, travis=True, add_env={}) 118 self._container_name = spec.container_name 119 120 def mapped_port(self, port): 121 return docker_mapped_port(self._container_name, port) 122 123 def wait_for_healthy(self, timeout_seconds): 124 wait_for_healthy(self._container_name, self._spec.shortname, 125 timeout_seconds) 126 127 def kill(self, suppress_failure=False): 128 """Sends kill signal to the container.""" 129 if suppress_failure: 130 self._job.suppress_failure_message() 131 return docker_kill(self._container_name) 132 133 def is_running(self): 134 """Polls a job and returns True if given job is still running.""" 135 return self._job.state() == jobset._RUNNING 136
