mirror of
https://github.com/docker/compose.git
synced 2026-07-03 14:53:58 +00:00
28f8b8549d
Starting with Docker 20.10, the docker daemon has support for "dual logging", which allows reading back logs, irregardless of the logging-driver that is configured (except for "none" as logging driver). This patch removes the local check, which used a hard-coded list of logging drivers that are expected to support reading logs. When using an older version of Docker, the API should return an error that reading logs is not supported, so no local check should be needed. Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
322 lines
9.1 KiB
Python
322 lines
9.1 KiB
Python
from functools import reduce
|
|
|
|
from docker.errors import ImageNotFound
|
|
|
|
from .const import LABEL_CONTAINER_NUMBER
|
|
from .const import LABEL_ONE_OFF
|
|
from .const import LABEL_PROJECT
|
|
from .const import LABEL_SERVICE
|
|
from .const import LABEL_SLUG
|
|
from .const import LABEL_VERSION
|
|
from .utils import truncate_id
|
|
from .version import ComposeVersion
|
|
|
|
|
|
class Container:
|
|
"""
|
|
Represents a Docker container, constructed from the output of
|
|
GET /containers/:id:/json.
|
|
"""
|
|
def __init__(self, client, dictionary, has_been_inspected=False):
|
|
self.client = client
|
|
self.dictionary = dictionary
|
|
self.has_been_inspected = has_been_inspected
|
|
self.log_stream = None
|
|
|
|
@classmethod
|
|
def from_ps(cls, client, dictionary, **kwargs):
|
|
"""
|
|
Construct a container object from the output of GET /containers/json.
|
|
"""
|
|
name = get_container_name(dictionary)
|
|
if name is None:
|
|
return None
|
|
|
|
new_dictionary = {
|
|
'Id': dictionary['Id'],
|
|
'Image': dictionary['Image'],
|
|
'Name': '/' + name,
|
|
}
|
|
return cls(client, new_dictionary, **kwargs)
|
|
|
|
@classmethod
|
|
def from_id(cls, client, id):
|
|
return cls(client, client.inspect_container(id), has_been_inspected=True)
|
|
|
|
@classmethod
|
|
def create(cls, client, **options):
|
|
response = client.create_container(**options)
|
|
return cls.from_id(client, response['Id'])
|
|
|
|
@property
|
|
def id(self):
|
|
return self.dictionary['Id']
|
|
|
|
@property
|
|
def image(self):
|
|
return self.dictionary['Image']
|
|
|
|
@property
|
|
def image_config(self):
|
|
return self.client.inspect_image(self.image)
|
|
|
|
@property
|
|
def short_id(self):
|
|
return self.id[:12]
|
|
|
|
@property
|
|
def name(self):
|
|
return self.dictionary['Name'][1:]
|
|
|
|
@property
|
|
def project(self):
|
|
return self.labels.get(LABEL_PROJECT)
|
|
|
|
@property
|
|
def service(self):
|
|
return self.labels.get(LABEL_SERVICE)
|
|
|
|
@property
|
|
def name_without_project(self):
|
|
if self.name.startswith('{}_{}'.format(self.project, self.service)):
|
|
return '{}_{}'.format(self.service, self.number if self.number is not None else self.slug)
|
|
else:
|
|
return self.name
|
|
|
|
@property
|
|
def number(self):
|
|
if self.one_off:
|
|
# One-off containers are no longer assigned numbers and use slugs instead.
|
|
return None
|
|
|
|
number = self.labels.get(LABEL_CONTAINER_NUMBER)
|
|
if not number:
|
|
raise ValueError("Container {} does not have a {} label".format(
|
|
self.short_id, LABEL_CONTAINER_NUMBER))
|
|
return int(number)
|
|
|
|
@property
|
|
def slug(self):
|
|
if not self.full_slug:
|
|
return None
|
|
return truncate_id(self.full_slug)
|
|
|
|
@property
|
|
def full_slug(self):
|
|
return self.labels.get(LABEL_SLUG)
|
|
|
|
@property
|
|
def one_off(self):
|
|
return self.labels.get(LABEL_ONE_OFF) == 'True'
|
|
|
|
@property
|
|
def ports(self):
|
|
self.inspect_if_not_inspected()
|
|
return self.get('NetworkSettings.Ports') or {}
|
|
|
|
@property
|
|
def human_readable_ports(self):
|
|
def format_port(private, public):
|
|
if not public:
|
|
return [private]
|
|
return [
|
|
'{HostIp}:{HostPort}->{private}'.format(private=private, **pub)
|
|
for pub in public
|
|
]
|
|
|
|
return ', '.join(
|
|
','.join(format_port(*item))
|
|
for item in sorted(self.ports.items())
|
|
)
|
|
|
|
@property
|
|
def labels(self):
|
|
return self.get('Config.Labels') or {}
|
|
|
|
@property
|
|
def stop_signal(self):
|
|
return self.get('Config.StopSignal')
|
|
|
|
@property
|
|
def log_config(self):
|
|
return self.get('HostConfig.LogConfig') or None
|
|
|
|
@property
|
|
def human_readable_state(self):
|
|
if self.is_paused:
|
|
return 'Paused'
|
|
if self.is_restarting:
|
|
return 'Restarting'
|
|
if self.is_running:
|
|
return 'Ghost' if self.get('State.Ghost') else self.human_readable_health_status
|
|
else:
|
|
return 'Exit %s' % self.get('State.ExitCode')
|
|
|
|
@property
|
|
def human_readable_command(self):
|
|
entrypoint = self.get('Config.Entrypoint') or []
|
|
cmd = self.get('Config.Cmd') or []
|
|
return ' '.join(entrypoint + cmd)
|
|
|
|
@property
|
|
def environment(self):
|
|
def parse_env(var):
|
|
if '=' in var:
|
|
return var.split("=", 1)
|
|
return var, None
|
|
return dict(parse_env(var) for var in self.get('Config.Env') or [])
|
|
|
|
@property
|
|
def exit_code(self):
|
|
return self.get('State.ExitCode')
|
|
|
|
@property
|
|
def is_running(self):
|
|
return self.get('State.Running')
|
|
|
|
@property
|
|
def is_restarting(self):
|
|
return self.get('State.Restarting')
|
|
|
|
@property
|
|
def is_paused(self):
|
|
return self.get('State.Paused')
|
|
|
|
@property
|
|
def log_driver(self):
|
|
return self.get('HostConfig.LogConfig.Type')
|
|
|
|
@property
|
|
def human_readable_health_status(self):
|
|
""" Generate UP status string with up time and health
|
|
"""
|
|
status_string = 'Up'
|
|
container_status = self.get('State.Health.Status')
|
|
if container_status == 'starting':
|
|
status_string += ' (health: starting)'
|
|
elif container_status is not None:
|
|
status_string += ' (%s)' % container_status
|
|
return status_string
|
|
|
|
def attach_log_stream(self):
|
|
self.log_stream = self.attach(stdout=True, stderr=True, stream=True)
|
|
|
|
def get(self, key):
|
|
"""Return a value from the container or None if the value is not set.
|
|
|
|
:param key: a string using dotted notation for nested dictionary
|
|
lookups
|
|
"""
|
|
self.inspect_if_not_inspected()
|
|
|
|
def get_value(dictionary, key):
|
|
return (dictionary or {}).get(key)
|
|
|
|
return reduce(get_value, key.split('.'), self.dictionary)
|
|
|
|
def get_local_port(self, port, protocol='tcp'):
|
|
port = self.ports.get("{}/{}".format(port, protocol))
|
|
return "{HostIp}:{HostPort}".format(**port[0]) if port else None
|
|
|
|
def get_mount(self, mount_dest):
|
|
for mount in self.get('Mounts'):
|
|
if mount['Destination'] == mount_dest:
|
|
return mount
|
|
return None
|
|
|
|
def start(self, **options):
|
|
return self.client.start(self.id, **options)
|
|
|
|
def stop(self, **options):
|
|
return self.client.stop(self.id, **options)
|
|
|
|
def pause(self, **options):
|
|
return self.client.pause(self.id, **options)
|
|
|
|
def unpause(self, **options):
|
|
return self.client.unpause(self.id, **options)
|
|
|
|
def kill(self, **options):
|
|
return self.client.kill(self.id, **options)
|
|
|
|
def restart(self, **options):
|
|
return self.client.restart(self.id, **options)
|
|
|
|
def remove(self, **options):
|
|
return self.client.remove_container(self.id, **options)
|
|
|
|
def create_exec(self, command, **options):
|
|
return self.client.exec_create(self.id, command, **options)
|
|
|
|
def start_exec(self, exec_id, **options):
|
|
return self.client.exec_start(exec_id, **options)
|
|
|
|
def rename_to_tmp_name(self):
|
|
"""Rename the container to a hopefully unique temporary container name
|
|
by prepending the short id.
|
|
"""
|
|
if not self.name.startswith(self.short_id):
|
|
self.client.rename(
|
|
self.id, '{}_{}'.format(self.short_id, self.name)
|
|
)
|
|
|
|
def inspect_if_not_inspected(self):
|
|
if not self.has_been_inspected:
|
|
self.inspect()
|
|
|
|
def wait(self):
|
|
return self.client.wait(self.id).get('StatusCode', 127)
|
|
|
|
def logs(self, *args, **kwargs):
|
|
return self.client.logs(self.id, *args, **kwargs)
|
|
|
|
def inspect(self):
|
|
self.dictionary = self.client.inspect_container(self.id)
|
|
self.has_been_inspected = True
|
|
return self.dictionary
|
|
|
|
def image_exists(self):
|
|
try:
|
|
self.client.inspect_image(self.image)
|
|
except ImageNotFound:
|
|
return False
|
|
|
|
return True
|
|
|
|
def reset_image(self, img_id):
|
|
""" If this container's image has been removed, temporarily replace the old image ID
|
|
with `img_id`.
|
|
"""
|
|
if not self.image_exists():
|
|
self.dictionary['Image'] = img_id
|
|
|
|
def attach(self, *args, **kwargs):
|
|
return self.client.attach(self.id, *args, **kwargs)
|
|
|
|
def has_legacy_proj_name(self, project_name):
|
|
return (
|
|
ComposeVersion(self.labels.get(LABEL_VERSION)) < ComposeVersion('1.21.0') and
|
|
self.project != project_name
|
|
)
|
|
|
|
def __repr__(self):
|
|
return '<Container: {} ({})>'.format(self.name, self.id[:6])
|
|
|
|
def __eq__(self, other):
|
|
if type(self) != type(other):
|
|
return False
|
|
return self.id == other.id
|
|
|
|
def __hash__(self):
|
|
return self.id.__hash__()
|
|
|
|
|
|
def get_container_name(container):
|
|
if not container.get('Name') and not container.get('Names'):
|
|
return None
|
|
# inspect
|
|
if 'Name' in container:
|
|
return container['Name']
|
|
# ps
|
|
shortest_name = min(container['Names'], key=lambda n: len(n.split('/')))
|
|
return shortest_name.split('/')[-1]
|