Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _create_tagged_image(base_image_tag, new_image_tag, app_or_lib_name):
docker_client = get_docker_client()
command = _get_test_image_setup_command(app_or_lib_name)
split_volumes = _get_split_volumes(get_volume_mounts(app_or_lib_name, get_expanded_libs_specs(), test=True))
create_container_volumes = _get_create_container_volumes(split_volumes)
create_container_binds = _get_create_container_binds(split_volumes)
container = docker_client.create_container(image=base_image_tag,
command=command,
volumes=create_container_volumes,
host_config=docker.utils.create_host_config(binds=create_container_binds))
docker_client.start(container=container['Id'])
log_to_client('Running commands to create new image:')
for line in docker_client.logs(container['Id'], stdout=True, stderr=True, stream=True):
log_to_client(line.strip())
exit_code = docker_client.wait(container['Id'])
if exit_code:
raise ImageCreationError(exit_code)
new_image = docker_client.commit(container=container['Id'])
try:
docker_client.remove_image(image=new_image_tag)
except:
log_to_client('Not able to remove image {}'.format(new_image_tag))
docker_client.tag(image=new_image['Id'], repository=new_image_tag, force=True)
docker_client.remove_container(container=container['Id'], v=True)
def _spawn_container(self, image, options):
host_config = create_host_config(network_mode="bridge",
binds=options.get_volume_binds(),
mem_limit=options.get_memory_limit())
cont = self.cli.create_container(image=image,
environment=options.get_environment(),
network_disabled=False,
host_config=host_config,
detach=True,
volumes=options.get_volumes(),
command=options.get_command())
self.cli.start(container=cont.get('Id'))
docker_info = self.cli.inspect_container(container=cont.get('Id'))
info = {
"docker_id": cont.get("Id"),
"docker_ip": docker_info["NetworkSettings"]["IPAddress"]
}
return info
"bind": "/proc_host",
"ro": False
},
log_dir:
{
"bind": "/var/log/calico",
"ro": False
},
"/run/docker/plugins":
{
"bind": "/usr/share/docker/plugins",
"ro": False
}
}
host_config = docker.utils.create_host_config(
privileged=True,
restart_policy={"Name": "Always"},
network_mode="host",
binds=binds)
_find_or_pull_node_image(node_image, docker_client)
container = docker_client.create_container(
node_image,
name="calico-node",
detach=True,
environment=environment,
host_config=host_config,
volumes=["/proc_host",
"/var/log/calico",
"/usr/share/docker/plugins"])
cid = container["Id"]
self.cur_time))
port_bindings = {
DEFAULT_APPLICATION_PORT: self.port,
self.internal_admin_port: self.admin_port,
}
if self.extra_ports:
port_bindings.update(self.extra_ports)
# The host_config specifies port bindings and volume bindings.
# /storage is bound to the storage_path. Internally, the
# devappserver writes all the db files to /storage. The mapping
# thus allows these files to appear on the host machine. As for
# port mappings, we only want to expose the application (via the
# proxy), and the admin panel.
devappserver_hconf = docker.utils.create_host_config(
port_bindings=port_bindings,
binds={
self.storage_path: {'bind': '/storage'},
}
)
self.devappserver_container = container.Container(self.dclient)
self.devappserver_container.create(
name=devappserver_container_name,
image=devappserver_image,
ports=port_bindings.keys(),
volumes=['/storage'],
host_config=devappserver_hconf,
environment=das_env)
self.devappserver_container.start()
def start_redis_container(docker_client, repository, tag):
"""
Start a Redis Docker container.
"""
logging.debug('Starting redis container')
redis_port = port_for.select_random()
host_config = docker.utils.create_host_config(port_bindings={
DEFAULT_REDIS_PORT: redis_port,
})
redis_image = '{}:{}'.format(repository, tag)
container_id = docker_client.create_container(
redis_image,
host_config=host_config
)['Id']
docker_client.start(container_id)
logging.info('Waiting for Redis to start...')
success = False
for _ in xrange(60):
def create_host_config(self):
"""
Create HostConfig object
"""
params = self.get_start_params()
return docker.utils.create_host_config(**params)
image=image,
command=command,
hostname=hostname,
user=user,
detach=detach,
stdin_open=stdin_open,
tty=tty,
ports=ports,
environment=environment,
dns=dns,
volumes=volumes,
volumes_from=volumes_from,
name=name,
cpu_shares=cpu_shares,
cpuset=cpuset,
host_config=docker.utils.create_host_config(binds=binds,
mem_limit=mem_limit)
)
else:
container_info = client.create_container(
image=image,
command=command,
hostname=hostname,
user=user,
detach=detach,
stdin_open=stdin_open,
tty=tty,
mem_limit=mem_limit,
ports=ports,
environment=environment,
dns=dns,
volumes=volumes,
def _make_host_config(self):
return create_host_config(
binds=self.volume_binds,
port_bindings=self.port_bindings,
network_mode=self.network_mode,
extra_hosts=self.extra_hosts,
# TODO: Support all of these.
lxc_conf=None,
publish_all_ports=False,
links=None,
privileged=False,
dns=None,
dns_search=None,
volumes_from=None,
restart_policy=None,
cap_add=None,
cap_drop=None,
devices=None,
# specified.
app_image = self.image_name or self.build_app_image()
app_container_name = self.make_timestamped_name('test_app',
self.cur_time)
# If devappserver is running, hook up the app to it.
if self.run_devappserver:
network_mode = ('container:%s' %
self.devappserver_container.get_id())
ports = port_bindings = None
else:
port_bindings = {DEFAULT_APPLICATION_PORT: self.port}
ports = [DEFAULT_APPLICATION_PORT]
network_mode = None
app_hconf = docker.utils.create_host_config(
port_bindings=port_bindings,
binds={
self.log_path: {'bind': '/var/log/app_engine'}
},
)
self.app_container = container.ApplicationContainer(
self.application_configuration,
self.dclient)
self.app_container.create(
name=app_container_name,
image=app_image,
ports=ports,
volumes=['/var/log/app_engine'],
host_config=app_hconf,
# Try to find an existing container
alias = alias or repo
image_name = '{}:{}'.format(repo, tag)
old_containers = get_containers_with_alias(network_name, alias)
env = []
if old_containers:
env = c.inspect_container(old_containers[0])['Config']['Env']
old_img = c.inspect_container(old_containers[0])['Image']
new_img = c.inspect_image(image_name)['Id']
if old_img == new_img:
# Same image, just let the old one run
return
try:
volumes = (c.inspect_image(image_name)['Config']['Volumes'] or {}).keys()
host_config = docker.utils.create_host_config(binds=[
'{}-{}-{}-{}:{}'.format(environ['DOMAIN_NAME'], alias, tag,
vol.replace('/', '_'), vol)
for vol in volumes
])
new_container = c.create_container('{}:{}'.format(repo, tag),
environment=env,
volumes=volumes,
host_config=host_config)
except docker.errors.NotFound as e:
logging.error(e)
# Just abort for now
return
c.connect_container_to_network(new_container, network_name,
aliases=[alias])
c.start(new_container)