diff --git a/docs/user/centralized_management.rst b/docs/user/centralized_management.rst index ba479473..0165ea9d 100644 --- a/docs/user/centralized_management.rst +++ b/docs/user/centralized_management.rst @@ -192,8 +192,14 @@ with a [d] appended to their usual agent_class name. If HostManager finds services in the docker-compose.yaml that don't seem to correspond to agent instances in site config, it will still -permit them to be "managed" (brought up and down). The agent_class, -in ocsbow or ocs-web, will show up as simply "[docker]". +track their state and make it visible in ocsbow / ocs-web. The +agent_class will show up as simply "[docker]". HostManager can be +told to bring the containers up or down, but this state will not be +recorded as a persistent target state. Other than during the request +and transition stages, the target state will be recorded as "passive". +This means such services can be brought up and down by other means +(i.e. docker compose on the command line) without any conflict with +HostManager. Advanced host config @@ -212,8 +218,8 @@ It is possible to mix host- and docker-based agents in a single host config block, and control them all with a single HostManager instance. Just make sure your docker-based agents are marked with ``'manage': 'docker'`` in site config, and have service name ``ocs-[instance-id]`` -as usual. Usually, docker-based agents have some command line -parameter overrides set in docker-compose.yaml (or in the site config +as usual. Sometimes docker-based agents will require some command line +parameter overrides in docker-compose.yaml (or in the site config block), because the crossbar address is different or weird from inside the container. If the hostname, in the docker container, is not the same as on the host system then specify the native host hostname with @@ -311,7 +317,7 @@ specific instance will show up with agent-class "?/[docker]" and an instance-id corresponding to the service name. For example:: [instance-id] [agent-class] [state] [target] - influxdb ?/[docker] up up + influxdb ?/[docker] up passive ``state`` and ``target`` diff --git a/ocs/agents/host_manager/agent.py b/ocs/agents/host_manager/agent.py index 853a9384..125ba994 100644 --- a/ocs/agents/host_manager/agent.py +++ b/ocs/agents/host_manager/agent.py @@ -6,7 +6,7 @@ import argparse from twisted.internet import reactor -from twisted.internet.defer import inlineCallbacks, returnValue +from twisted.internet.defer import inlineCallbacks, returnValue, DeferredList from autobahn.twisted.util import sleep as dsleep import os @@ -31,7 +31,11 @@ def __init__(self, agent, docker_composes=[], docker_service_prefix='ocs-'): self.agent = agent self.running = False - self.database = {} # key is instance_id (or docker service name). + # The database maps instance_id (or docker service name, if + # it's an unmanaged docker container) to a ManagedInstance. + self.database = {} + self.orphans = {} + self.new_tags = {} self.docker_composes = docker_composes self.docker_service_prefix = docker_service_prefix @@ -108,9 +112,12 @@ def _update_docker_services(self, session): """ # Read services from all docker-compose files. docker_services = {} + orphans = {} + new_tags = {} for compose in self.docker_composes: try: - services = yield hm_utils.parse_docker_state(compose) + services, _orphans = yield hm_utils.parse_docker_state(compose) + orphans.update(_orphans) this_ok = True this_msg = f'Successfully parsed {compose} and its service states.' except Exception as e: @@ -133,36 +140,52 @@ def _update_docker_services(self, session): # Update all docker things in the database. retirees = [] assigned_services = [] - for key, instance in self.database.items(): - if instance['management'] != 'docker': + for key, minst in self.database.items(): + if minst.management != 'docker': continue - service_name = instance['agent_script'] + service_name = minst.agent_script service_data = docker_services.get(service_name) assigned_services.append(service_name) - prot = instance.get('prot') + prot = minst.prot if prot is None: if service_data is not None: # Create a prot entry with the service info. - instance['prot'] = hm_utils.DockerContainerHelper(service_data) - instance['operable'] = True - if instance['agent_class'] != NONAGENT_DOCKER: - instance['agent_class'] = _clsname_tool(instance['agent_class'], '[d]') + minst.prot = hm_utils.DockerContainerHelper(service_data) + minst.operable = True + if minst.agent_class != NONAGENT_DOCKER: + minst.agent_class = _clsname_tool(minst.agent_class, '[d]') + # Update current_state to not trigger a "docker + # up" on a running container. Normally that would + # be a no-op, but not if image tag has changed. + if service_data['running']: + session.add_message(f'Docker-based instance {key} seems to be running.') + minst.next_action = 'up' + else: if service_data is not None: prot.update(service_data) else: # service_data is missing, but there used to be a # service there. Close it out. - instance['prot'] = None - instance['operable'] = False - if instance['agent_class'] == NONAGENT_DOCKER: + minst.prot = None + minst.operable = False + if minst.agent_class == NONAGENT_DOCKER: session.add_message(f'Deleting non-agent service {key}') retirees.append(key) else: session.add_message(f'Marking missing service for {key}') - instance['agent_class'] = _clsname_tool(instance['agent_class'], '[d?]') + minst.agent_class = _clsname_tool(minst.agent_class, '[d?]') + + restart_required = False + if minst.prot is not None and service_data is not None \ + and service_data.get('running') \ + and (service_data.get('running_image') != service_data.get('image_id')): + restart_required = True + if service_data is not None and service_data.get('image_id') == 'unknown': + new_tags[key] = service_data['image_tag'] + minst.restart_required = restart_required # If a non-agent [docker] service has disappeared, there's no # reason to show it in a list, and no persistent state / @@ -174,23 +197,35 @@ def _update_docker_services(self, session): unassigned_services = set(docker_services.keys()) \ .difference(assigned_services) for srv in unassigned_services: - instance = hm_utils.ManagedInstance.init( + session.add_message(f'Adding non-agent service "{srv}"') + minst = hm_utils.ManagedInstance( management='docker', instance_id=srv, agent_class=NONAGENT_DOCKER, - full_name=(f'[docker]:{srv}')) - instance.update({ - 'agent_script': srv, - 'operable': True, - }) + full_name=f'[docker]:{srv}', + agent_script=srv, + operable=True, + passive_tracking=True, + target_state='passive', + ) + service_data = docker_services[srv] - instance['prot'] = hm_utils.DockerContainerHelper(service_data) + minst.prot = hm_utils.DockerContainerHelper(service_data) - self.database[srv] = instance # If it's up, leave it up. if service_data['running']: - instance['target_state'] = 'up' - instance['next_action'] = 'up' + minst.next_action = 'up' + + self.database[srv] = minst + + # Update the list of orphans ... + orphans_gone = set(self.orphans.keys()).difference(orphans.keys()) + for k in orphans_gone: + del self.orphans[k] + self.orphans.update(orphans) + + # And new tags that need a docker pull ... + self.new_tags = new_tags returnValue(docker_services) @@ -212,12 +247,12 @@ def _reload_config(self, session): """ def retire(db_key): - instance = self.database.get(db_key, None) - if instance is None: + minst = self.database.get(db_key, None) + if minst is None: return - instance['management'] = 'retired' - instance['at'] = time.time() - instance['target_state'] = 'down' + minst.management = 'retired' + minst.at = time.time() + minst.target_state = 'down' def _full_name(cls, iid): if cls != NONAGENT_DOCKER: @@ -240,11 +275,11 @@ def same_base_class(a, b): # agent_dict should be immediately retired. That includes # things that are suddenly marked as manage=no. Ignore docker # non-agents. - for iid, instance in self.database.items(): - if instance['agent_class'] != NONAGENT_DOCKER and ( + for iid, minst in self.database.items(): + if minst.agent_class != NONAGENT_DOCKER and ( iid not in agent_dict or agent_dict[iid].get('manage') == 'ignore'): session.add_message( - f'Retiring {instance["full_name"]}, which has disappeared from ' + f'Retiring {minst.full_name}, which has disappeared from ' f'configuration file(s) or has manage:no.') retire(iid) @@ -263,12 +298,12 @@ def same_base_class(a, b): srv = self.docker_service_prefix + iid # See if we already tracking this agent. - instance = self.database.get(iid) + minst = self.database.get(iid) - if instance is not None: + if minst is not None: # Already tracking; just check for major config change. - _cls = instance['agent_class'] - _mgmt = instance['management'] + _cls = minst.agent_class + _mgmt = minst.management if not same_base_class(_cls, cls) or _mgmt != mgmt: session.add_message( f'Managed agent "{iid}" changed agent_class ' @@ -277,60 +312,65 @@ def same_base_class(a, b): # Bring down existing instance self._terminate_instance(iid) # Start a new one - instance = None + minst = None # Do we have an unmatched docker entry for this? - if instance is None and srv in self.database: + if minst is None and srv in self.database: + session.add_message( + f'Unmanaged docker service {srv} will now be managed as ' + f'instance_id={iid}.') # Re-register it under instance_id - instance = self.database.pop(srv) - self.database[iid] = instance - instance.update({ - 'instance_id': iid, - 'agent_class': _clsname_tool(cls, '[d]'), - 'full_name': _full_name(cls, iid), - }) - - if instance is None: - instance = hm_utils.ManagedInstance.init( + minst = self.database.pop(srv) + minst.instance_id = iid + minst.agent_class = _clsname_tool(cls, '[d]') + minst.full_name = _full_name(cls, iid) + minst.passive_tracking = False + if minst.target_state == 'passive': + minst.target_state = \ + minst.next_action if minst.next_action in ['up', 'down'] else 'down' + self.database[iid] = minst + + if minst is None: + minst = hm_utils.ManagedInstance( management=mgmt, instance_id=iid, agent_class=cls, full_name=_full_name(cls, iid), + target_state=start_state, ) - instance['target_state'] = start_state - self.database[iid] = instance + self.database[iid] = minst # Get agent class list from modern plugin system. agent_plugins = agent_cli.build_agent_list() # Assign plugins / scripts / whatever to any new instances. - for iid, instance in self.database.items(): - if instance['agent_script'] is not None: + for iid, minst in self.database.items(): + if minst.agent_script is not None: continue - if instance['management'] == 'host': - cls = instance['agent_class'] + if minst.management == 'host': + cls = minst.agent_class # Check for the agent class in the plugin system if cls in agent_plugins: session.add_message(f'Found plugin for "{cls}"') - instance['agent_script'] = '__plugin__' - instance['operable'] = True + minst.agent_script = '__plugin__' + minst.operable = True else: session.add_message('No plugin ' f'found for agent_class "{cls}"!') - elif instance['management'] == 'docker': - instance['agent_script'] = self.docker_service_prefix + iid + elif minst.management == 'docker': + minst.agent_script = self.docker_service_prefix + iid # Read the compose files; query container states; updater stuff. yield self._update_docker_services(session) returnValue(warnings) - def _launch_instance(self, instance): + def _launch_instance(self, minst): """Launch an Agent instance (whether 'host' or 'docker' managed) using hm_utils. For 'host' managed agents: hm_utils will use reactor.spawnProcess, and store the AgentProcessHelper (which - inherits from twisted ProcessProtocol) in instance['prot']. + inherits from twisted ProcessProtocol) in minst.prot. The site_file and instance_id are passed on the command line; this means that any weird config overrides passed to this HostManager are not propagated. One exception is working_dir, @@ -338,15 +378,15 @@ def _launch_instance(self, instance): sense. For 'docker' managed agents: hm_utils will try to start the - right service container, and instance['prot'] will hold a + right service container, and minst.prot will hold a DockerContainerHelper (which has some common interface with AgentProcessHelper). """ - if instance['management'] == 'docker': - prot = instance['prot'] + if minst.management == 'docker': + prot = minst.prot else: - iid = instance['instance_id'] + iid = minst.instance_id pyth = sys.executable cmd = [pyth, '-m', 'ocs.agent_cli', '--instance-id', iid, @@ -355,13 +395,13 @@ def _launch_instance(self, instance): '--working-dir', self.working_dir] prot = hm_utils.AgentProcessHelper(iid, cmd) prot.up() - instance['prot'] = prot + minst.prot = prot def _terminate_instance(self, key): """ Use the ProcessProtocol to request the Agent instance to exit. """ - prot = self.database[key]['prot'] # Get the ProcessProtocol. + prot = self.database[key].prot # Get the ProcessProtocol. if prot is None: return True, 'Instance was not running.' if prot.killed: @@ -377,20 +417,19 @@ def _process_target_states(self, session, requests=[]): """ # Special requests will target specific instance_id; make a map for that. - addressable = {k: v for k, v in self.database.items() - if v['management'] != 'retired'} - + addressable = {k: minst for k, minst in self.database.items() + if minst.management != 'retired'} for key, state in requests: if state not in VALID_TARGETS: session.add_message('Ignoring request for "%s" -> invalid state "%s".' % (key, state)) continue if key == 'all': - for v in addressable.values(): - v['target_state'] = state + for minst in addressable.values(): + minst.target_state = state else: if key in addressable: - addressable[key]['target_state'] = state + addressable[key].target_state = state else: session.add_message(f'Ignoring invalid target, {key}') @@ -431,11 +470,21 @@ def manager(self, session, params): attempt will be made to terminate them before the Process exits. - The session.data is a dict, and entry 'child_states' - contains a list with the managed Agent statuses. For - example:: - - {'child_states': [ + The session.data is a dict, with entries: + - 'child_states' + - 'config_parse_status' - indicates how recently the various + input files havebeen parsed. + - 'orphans' - lists any orphaned (in the sense of docker + compose) containers. + - 'new_tags' - dict mapping instance_id to new docker image + tag, if that tag is not known to docker system. Only + populated for tracked instances where the tag is not + known. + + The 'child_states' entry is a list of managed Agent status; + for example:: + + [ {'next_action': 'up', 'target_state': 'up', 'stability': 1.0, @@ -451,8 +500,7 @@ def manager(self, session, params): 'stability': 1.0, 'agent_class': 'FakeDataAgent[d]', 'instance_id': 'faker6'}, - ], - } + ] If you are looking for the "current state", it's called "next_action" here. @@ -463,11 +511,40 @@ def manager(self, session, params): HostManager cannot actually identify the docker-compose service associated with the agent description in the SCF.) + The 'config_parse_status' is a dict where the key is a + docker compose filename, or "[SCF]" for the site config + file, and the value is a tuple (success, timestamp, + message). + + The 'orphans' entry is as a dict mapping docker container ID + to some information about the container. E.g.:: + + { + "30027f37e0ef4b...": { + "compose_file": "/home/ocs/config/docker-compose.yml", + "service": "ocs-faker3", + "container_id": "30027f37e0ef4b...", + "running": true, + "exit_code": 0, + "container_found": true, + "running_image": "sha256:7eaa6d6f6..." + } + } + + The 'new_tags' entry looks like this:: + + { + "faker1": "simonsobs/ocs:v0.11.3", + "faker2": "simonsobs/ocs:v0.11.3" + } + """ self.config_parse_status = {} session.data = { 'child_states': [], 'config_parse_status': self.config_parse_status, + 'orphans': self.orphans, + 'new_tags': self.new_tags, } self.running = True @@ -489,43 +566,52 @@ def manager(self, session, params): sleep_times = [1.] any_jobs = False - for key, db in self.database.items(): + for key, minst in self.database.items(): # If Process exit is requested, force all targets to down. - if not self.running: - db['target_state'] = 'down' + if not self.running and not minst.passive_tracking: + minst.target_state = 'down' - actions = hm_utils.resolve_child_state(db) + actions = hm_utils.resolve_child_state(minst) for msg in actions['messages']: session.add_message(msg) if actions['terminate']: self._terminate_instance(key) if actions['launch']: - reactor.callFromThread(self._launch_instance, db) + reactor.callFromThread(self._launch_instance, minst) if actions['sleep']: sleep_times.append(actions['sleep']) - any_jobs = (any_jobs or (db['next_action'] != 'down')) + + if minst.passive_tracking: + this_job = minst.next_action != 'passive' + else: + this_job = minst.next_action != 'down' + any_jobs = (any_jobs or this_job) # Criteria for stability: - db['fail_times'], db['stability'] = hm_utils.stability_factor( - db['fail_times']) + minst.fail_times, minst.stability = hm_utils.stability_factor( + minst.fail_times) # Clean up retired items. self.database = { - k: v for k, v in self.database.items() - if v['management'] != 'retired' or v['next_action'] not in ['down', '?']} + k: minst for k, minst in self.database.items() + if minst.management != 'retired' or minst.next_action not in ['down', '?']} # Update session info. child_states = [] - for state in self.database.values(): - child_states.append({_k: state[_k] for _k in + for minst in self.database.values(): + child_states.append({_k: getattr(minst, _k) for _k in ['next_action', 'target_state', 'stability', 'agent_class', - 'instance_id']}) + 'instance_id', + 'operable', + 'restart_required', + ]}) session.data['child_states'] = child_states + session.data['new_tags'] = self.new_tags yield dsleep(max(min(sleep_times), .001)) return True, 'Exited.' @@ -598,7 +684,73 @@ def update(self, session, params): return True, 'Update requested.' @inlineCallbacks + def remove_orphans(self, session, params): + """remove_orphans(stop_time=10.) + + **Task** - Use docker stop and docker rm to remove orphaned + containers associated with managed docker compose files. + + This does not really do any error checking. + """ + containers = list(self.orphans.values()) + session.add_message(f'Attempting stop and remove of {len(containers)} containers.') + + defs = [] + for cont in containers: + print(f'Stopping {cont["container_id"][:16]} ...') + d = hm_utils._run_docker(['stop', cont['container_id']]) + defs.append(d) + + yield DeferredList(defs) + + defs = [] + for cont in containers: + print(f'Removing {cont["container_id"][:16]} ...') + d = hm_utils._run_docker(['rm', cont['container_id']]) + defs.append(d) + + yield DeferredList(defs) + + return True, 'Done.' + + @inlineCallbacks + def docker_pull(self, session, params): + """docker_pull() + + **Task** - Use docker compose to pull any (new) images for the + managed docker compose files. + + """ + for compose in self.docker_composes: + session.add_message(f'Running pull for {compose} ...') + yield hm_utils._run_docker(['compose', '-f', compose, 'pull']) + + return True, 'Done.' + + @inlineCallbacks + @ocs_agent.param('disown_dockers', default=False, type=bool) def die(self, session, params): + """die(disown_dockers=False) + + **Task** - trigger a shutdown of the manage process and then + stop the reactor, causing the HostManager to exit. + + Args: + disown_dockers (bool): If True, then all tracked docker + services will be put in "passive tracking" mode, meaning + that they will not be stopped and removed during this + shutdown process. This can be used to restart HostManager + without needing to also restart all (docker-based) agents + on the system. + + """ + if params['disown_dockers']: + for minst in self.database.values(): + if minst.management == 'docker': + minst.passive_tracking = True + if minst.target_state in ['up', 'down']: + minst.target_state = 'passive' + if not self.running: session.add_message('Manager process is not running.') else: @@ -684,6 +836,8 @@ def main(args=None): blocking=False, startup=startup_params) agent.register_task('update', host_manager.update, blocking=False) + agent.register_task('remove_orphans', host_manager.remove_orphans, blocking=False) + agent.register_task('docker_pull', host_manager.docker_pull, blocking=False) agent.register_task('die', host_manager.die, blocking=False) reactor.addSystemEventTrigger('before', 'shutdown', agent._stop_all_running_sessions) diff --git a/ocs/agents/host_manager/drivers.py b/ocs/agents/host_manager/drivers.py index 24f931b4..6fe8cdc1 100644 --- a/ocs/agents/host_manager/drivers.py +++ b/ocs/agents/host_manager/drivers.py @@ -5,64 +5,95 @@ from twisted.internet import reactor, utils, protocol from twisted.internet.defer import inlineCallbacks +from dataclasses import dataclass, field +from typing import List -class ManagedInstance(dict): - """Track properties of a managed Agent-instance. This is just a dict - with a schema docstring and an "init" function to set defaults. - - Properties that must be set explicitly by user: - - - 'management' (str): Either 'host', 'docker', or 'retired'. - - 'agent_class' (str): The agent class name, which may include a - suffix ([d] or [d?]) if the agent is managed through Docker. - For instances corresponding to docker services that do not have - a corresponding SCF entry, the value here will be '[docker]'. - - 'instance_id' (str): The agent instance-id, or the docker - service name if the instance is an unmatched docker service. - - 'full_name' (str): agent_class:instance_id - - Properties that are given a default value by init function: - - - 'operable' (bool): indicates whether the instance can be - manipulated (whether calls to up/down should be expected to - work). - - 'agent_script' (str): The docker service_name, if docker-managed. - Otherwise, the string ``__plugin__`` to indicate it is host managed. - - 'prot': The twisted ProcessProtocol object (if host system - managed), or the DockerContainerHelper (if a docker container). - - 'target_state' (state): The state we're trying to achieve (up or - down). - - 'next_action' (state): The thing HostManager needs to do next; - this will sometimes indicate the "current state" (up or down), - but sometimes it will carry a transitional state, such as - "wait_start". - - 'at' (float): a unix timestamp for transitional states - (e.g. used to set how long to wait for something). - - 'fail_times' (list of floats): unix timestamps when the instance - process has stopped unexpectedly (used to identify "unstable" - agents). + +WAIT_DEAD_TIME = 11 +WAIT_START_TIME_INIT = 15 +WAIT_START_TIME_FOLLOWUP = 5 + + +@dataclass +class ManagedInstance: + """Tracks the properties of a managed Agent-instance, including + how to launch it, the current run state, target state, etc. """ - @classmethod - def init(cls, **kwargs): - # Note some core things are not included. - self = cls({ - 'agent_script': None, - 'operable': False, - 'prot': None, - 'next_action': 'down', - 'target_state': 'down', - 'fail_times': [], - 'at': 0, - }) - self.update(kwargs) - return self - - -def resolve_child_state(db): + + #: How host is managed; either "host", "docker", or "retired". + management: str + + #: Agent class name (which may include suffix "[d]" or "[d?]" for + #: docker-managed instances; or simply "[docker]" for services + #: that do not seem to be registered in the SCF. + agent_class: str + + #: The agent instance's instance_id, or else the docker service + #: name associated with entry in the SCF. + instance_id: str + + #: Indentier constructed as agent_class:instance_id. + full_name: str + + #: Indicates whether the instance can be manipulated (whether + #: calls to up/down should be expected to work). + operable: bool = False + + #: Indicates if instance is retired and can be removed from + #: tracking. + retired: bool = False + + #: Indicates if instance should be "passively" managed, e.g. not + #: be enforced other than ephemerally to attempt a start / stop. + #: This is expected to only be used for docker-based instances. + passive_tracking: bool = False + + #: The docker service name, if docker-managed; otherwisre the + #: string ``__plugin__`` to indicate it is host managed. + agent_script: str = None + + #: The Twisted ProcessProtocol object, if host system managed; or + #: else the DockerContainerHelper if docker-based. + prot: object = None + + #: Indicates a restart is in order, due to change of docker tag or + #: other new software version. + restart_required: bool = False + + #: The run state HostManager is trying to enforce (up, down, passive). + target_state: str = 'down' + + #: The thing HostManager plans to do next; this will sometimes + #: mirror the current state (up or down) and will sometimes carry a + #: transitional state, such as "wait_start". + next_action: str = 'down' + + #: Unix timestamp, used by transitional states to indicate time at + #: which some subsequent action should be taken. + at: float = 0 + + #: List of unix timestamps for recent events where an instance + #: stopped unexpectedly; used to identify "unstable" agents. + fail_times: List = field(default_factory=list) + + @property + def is_running(self): + if self.prot is None: + return False + return self.prot.is_running + + @property + def exit_code(self): + if self.prot is None: + return None + return self.prot.status[0] + + +def resolve_child_state(minst): """Args: - db (ManagedInstance): the instance state information. This will + minst (ManagedInstance): the instance state information. This will be modified in place. Returns: @@ -85,116 +116,117 @@ def resolve_child_state(db): messages = [] sleeps = [] - # State machine. - prot = db['prot'] - # If the entry is not "operable", send next_action to '?' and # don't try to do anything else. - - if not db['operable']: - db['next_action'] = '?' + if not minst.operable: + minst.next_action = '?' # The uninterruptible transition state(s) are most easily handled # in the same way regardless of target state. # Transitional: wait_start, which bridges from start -> up. - elif db['next_action'] == 'wait_start': - if prot is not None: - messages.append('Launched {full_name}'.format(**db)) - db['next_action'] = 'up' - else: - if time.time() >= db['at']: + elif minst.next_action == 'wait_start': + if minst.is_running: + messages.append('Launched {0.full_name}'.format(minst)) + minst.next_action = 'up' + if minst.passive_tracking: + minst.target_state = 'passive' + elif time.time() >= minst.at: + if minst.passive_tracking: + messages.append( + 'Launch not detected for {0.full_name}! ' + 'Passive tracking so will not try again.' + .format(minst)) + minst.target_state = 'passive' + minst.next_action = 'down' + else: messages.append('Launch not detected for ' - '{full_name}! Will retry.'.format(**db)) - db['next_action'] = 'start_at' - db['at'] = time.time() + 5. + '{0.full_name}! Will retry.'.format(minst)) + minst.next_action = 'start_at' + minst.at = time.time() + WAIT_START_TIME_FOLLOWUP # Transitional: wait_dead, which bridges from kill -> idle. - elif db['next_action'] == 'wait_dead': - if prot is None: - stat, t = 0, None - else: - stat, t = prot.status - if stat is not None: - db['next_action'] = 'down' - elif time.time() >= db['at']: - if stat is None: - messages.append('Agent instance {full_name} ' - 'refused to die.'.format(**db)) - db['next_action'] = 'down' + elif minst.next_action == 'wait_dead': + if not minst.is_running: + minst.next_action = 'down' + if minst.passive_tracking: + minst.target_state = 'passive' + messages.append('Agent instance {0.full_name} has exited' + .format(minst)) + elif time.time() >= minst.at: + messages.append('Agent instance {0.full_name} ' + 'refused to die.'.format(minst)) + minst.next_action = 'down' else: - sleeps.append(db['at'] - time.time()) + sleeps.append(minst.at - time.time()) # State handling when target is to be 'up'. - elif db['target_state'] == 'up': - if db['next_action'] == 'start_at': - if time.time() >= db['at']: - db['next_action'] = 'start' + elif minst.target_state == 'up': + if minst.next_action == 'start_at': + if time.time() >= minst.at: + minst.next_action = 'start' else: - sleeps.append(db['at'] - time.time()) - elif db['next_action'] == 'start': + sleeps.append(minst.at - time.time()) + elif minst.next_action == 'start': messages.append( - 'Requested launch for {full_name}'.format(**db)) + 'Requested launch for {0.full_name}'.format(minst)) actions['launch'] = True - db['next_action'] = 'wait_start' + minst.next_action = 'wait_start' now = time.time() - db['at'] = now + 1. - elif db['next_action'] == 'up': - if prot is None: - stat, t = 0, None - else: - stat, t = prot.status - if stat is not None: - messages.append('Detected exit of {full_name} ' - 'with code {stat}.'.format(stat=stat, **db)) - if hasattr(prot, 'lines'): + minst.at = now + WAIT_START_TIME_INIT + elif minst.next_action == 'up': + if not minst.is_running: + messages.append('Detected exit of {0.full_name} ' + 'with code {0.exit_code}.'.format(minst)) + if hasattr(minst.prot, 'lines'): note = '' - lines = prot.lines['stderr'] + lines = minst.prot.lines['stderr'] if len(lines) > 50: note = ' (trimmed)' lines = lines[-20:] - messages.append('stderr output from {full_name}{note}: {}' - .format('\n'.join(lines), note=note, **db)) - db['next_action'] = 'start_at' - db['at'] = time.time() + 3 - db['fail_times'].append(time.time()) + messages.append('stderr output from {minst.full_name}{note}: {}' + .format('\n'.join(lines), note=note, minst=minst)) + minst.next_action = 'start_at' + minst.at = time.time() + 3 + minst.fail_times.append(time.time()) else: # 'down' - db['next_action'] = 'start' + minst.next_action = 'start' # State handling when target is to be 'down'. - elif db['target_state'] == 'down': - if db['next_action'] == 'down': - # The lines below will prevent HostManager from killing - # Agents that suddenly seem to be alive. With these - # lines commented out, someone running "up" on a managed - # docker-compose.yaml will see their Agents immediately - # be brought down by HostManager. - # if prot is not None and prot.status[0] is None: - # messages.append('Detected unexpected session for {full_name} ' - # '(probably docker); changing target state to "up".'.format(**db)) - # db['target_state'] = 'up' - + elif minst.target_state == 'down': + if minst.next_action == 'down': # In fully managed mode, force a termination. - if prot is not None and prot.status[0] is None: - messages.append('Detected unexpected session for {full_name} ' - '(probably docker); it will be shut down.'.format(**db)) - db['next_action'] = 'up' - elif db['next_action'] == 'up': + if minst.is_running: + messages.append('Detected unexpected session for {0.full_name} ' + '(probably docker); it will be shut down.'.format(minst)) + minst.next_action = 'up' + elif minst.next_action == 'up': messages.append('Requesting termination of ' - '{full_name}'.format(**db)) + '{0.full_name}'.format(minst)) actions['terminate'] = True - db['next_action'] = 'wait_dead' - db['at'] = time.time() + 5 + minst.next_action = 'wait_dead' + minst.at = time.time() + WAIT_DEAD_TIME else: # 'start_at', 'start' - messages.append('Modifying state of {full_name} from ' - '{next_action} to idle'.format(**db)) - db['next_action'] = 'down' + messages.append('Modifying state of {0.full_name} from ' + '{0.next_action} to idle'.format(minst)) + minst.next_action = 'down' + + elif minst.passive_tracking: + # For passive tracking, next_action always reflects the + # current running state. + if minst.is_running and minst.next_action != 'up': + messages.append(f'Passively tracked {minst.full_name} is now up.') + minst.next_action = 'up' + elif not minst.is_running and minst.next_action != 'down': + messages.append(f'Passively tracked {minst.full_name} is now down.') + minst.next_action = 'down' + minst.target_state = 'passive' # Should not get here. else: messages.append( - 'State machine failure: state={next_action}, target_state' - '={target_state}'.format(**db)) + 'State machine failure: state={0.next_action}, target_state' + '={0.target_state}'.format(minst)) actions['messages'] = messages if len(sleeps): @@ -243,6 +275,10 @@ def down(self): if self.status[0] is None: reactor.callFromThread(self.transport.signalProcess, 'INT') + @property + def is_running(self): + return self.status[0] is None + # See https://twistedmatrix.com/documents/current/core/howto/process.html # # These notes, and the useless prototypes below them, are to get @@ -283,9 +319,24 @@ def errReceived(self, data): self.lines['stderr'] = self.lines['stderr'][-100:] -def _run_docker(args): - return utils.getProcessOutputAndValue( +def _decode(args): + out, err, code = args + return (out.decode('utf8'), err.decode('utf8'), code) + + +def _deyaml(args): + out, err, code = args + return (yaml.safe_load(out), err, code) + + +def _run_docker(args, decode=False, deyaml=False): + d = utils.getProcessOutputAndValue( 'docker', args, env=os.environ) + if decode or deyaml: + d = d.addCallback(_decode) + if deyaml: + d = d.addCallback(_deyaml) + return d class DockerContainerHelper: @@ -298,7 +349,8 @@ class DockerContainerHelper: def __init__(self, service, docker_bin=None): self.service = {} - self.status = -1, time.time() + self.is_running = False + self.status = None, time.time() self.killed = False self.instance_id = service['service'] self.d = None @@ -311,15 +363,17 @@ def update(self, service): """ self.service.update(service) if service['running']: + self.is_running = True self.status = None, time.time() else: + self.is_running = False self.status = service['exit_code'], time.time() self.killed = False def up(self): self.d = _run_docker( ['compose', '-f', self.service['compose_file'], - 'up', '-d', self.service['service']]) + 'up', '--remove-orphans', '-d', self.service['service']]) self.status = None, time.time() def down(self): @@ -336,41 +390,75 @@ def parse_docker_state(docker_compose_file): service is running or not. Returns: - A dict where the key is the service name and each value is a - dict with the following entries: - - - 'compose_file': the path to the docker compose file - - 'service': service name - - 'container_found': bool, indicates whether a container for - this service was found (whether or not it was running). - - 'running': bool, indicating that a container for this service - is currently in state "Running". - - 'exit_code': int, which is either extracted from the docker - inspect output or is set to 127. (This should never be None.) + services: + A dict where the key is the service name and each value is a + dict with the following entries: + + - 'compose_file': the path to the docker compose file + - 'service': service name + - 'image_tag': the tag listed for the image in the compose + file (this may differ from the running image). + - 'image_id': the docker image ID corresponding to + 'image_tag'; will be "unknown" if, e.g., listed tag is not + yet pulled to the running system. + - 'container_found': bool, indicates whether a container for + this service was found (whether or not it was running). + - 'container_id': the docker ID of the container (if found). + - 'running': bool, indicating that the found container is + in state "Running". + - 'running_image': the ID of the image for the container (if + found; e.g. "sha:0f..."). + - 'exit_code': int, which is either extracted from the docker + inspect output or is set to 127. (This should never be None.) + + orphans: + A dict (by container id) of dicts describing running + containers that are associated with this compose file but have + apparently been removed from the service list. Key is the + service name. """ summary = {} + orphans = {} + compose, err, code = yield _run_docker(['compose', '-f', docker_compose_file, 'config'], + deyaml=True) - compose = yaml.safe_load(open(docker_compose_file, 'r')) - for key, cfg in compose.get('services', []).items(): + for key, cfg in compose.get('services', {}).items(): summary[key] = { + 'compose_file': docker_compose_file, 'service': key, + 'image_tag': cfg['image'], + 'image_id': 'unknown', + 'container_found': False, + 'container_id': None, 'running': False, + 'running_image': None, 'exit_code': 127, - 'container_found': False, - 'compose_file': docker_compose_file, } + # Look up each tag; create map from tag to image_id. + to_inspect = list(set([cfg['image_tag'] for cfg in summary.values()])) + image_ids = {} + if len(to_inspect): + # Output from inspect is not neccessarily one-to-one with + # items on command line, if image of a tag is not yet known. + out, err, code = yield _run_docker(['inspect'] + to_inspect, deyaml=True) + for image in out: + image_ids.update({k: image['Id'] for k in image['RepoTags']}) + + for cfg in summary.values(): + cfg['image_id'] = image_ids.get(cfg['image_tag'], 'unknown') + # Query docker compose for container ids... out, err, code = yield _run_docker( - ['compose', '-f', docker_compose_file, 'ps', '-q']) + ['compose', '-f', docker_compose_file, 'ps', '-q'], decode=True) if code != 0: raise RuntimeError("Could not run docker compose or could not parse " "compose.yaml file; exit code %i, error text: %s" % (code, err)) - cont_ids = [line.strip() for line in out.decode('utf8').split('\n') + cont_ids = [line.strip() for line in out.split('\n') if line.strip() != ''] # Run docker inspect. @@ -385,30 +473,34 @@ def parse_docker_state(docker_compose_file): service = info.pop('service') if service not in summary: - raise RuntimeError("Consistency problem: image does not self-report " - "as a listed service? (%s)" % (service)) - summary[service].update(info) + orphans[cont_id] = { + 'compose_file': docker_compose_file, + 'service': service, + 'container_id': cont_id, + } | info + else: + summary[service].update(info) - return summary + return summary, orphans @inlineCallbacks def _inspectContainer(cont_id, docker_compose_file): """Run docker inspect on cont_id, return dict with the results.""" - out, err, code = yield _run_docker( - ['inspect', cont_id]) + info, err, code = yield _run_docker( + ['inspect', cont_id], deyaml=True) if code != 0 and 'No such object' in err.decode('utf8'): # This is likely due to a race condition where some # container was brought down since we ran docker compose. # Return None to indicate this -- caller should just ignore for now. print(f'(_inspectContainer: warning, no such object: {cont_id}') return None - elif code != 0: + elif code != 0 or len(info) != 1: raise RuntimeError( f'Trouble running "docker inspect {cont_id}".\n' - f'stdout: {out}\n stderr {err}') + f'stdout: {info}\n stderr {err}') # Reconcile config against docker compose ... - info = yaml.safe_load(out)[0] + info = info[0] config = info['Config']['Labels'] _dc_file = os.path.join(config['com.docker.compose.project.working_dir'], config['com.docker.compose.project.config_files']) @@ -416,9 +508,12 @@ def _inspectContainer(cont_id, docker_compose_file): raise RuntimeError("Consistency problem: container started from " "some other compose file?\n%s\n%s" % (docker_compose_file, _dc_file)) service = config['com.docker.compose.service'] + # Note returned dict is merged into summary output for + # parse_docker_state, so use keys documented there. return { 'service': service, 'running': info['State']['Running'], 'exit_code': info['State'].get('ExitCode', 127), 'container_found': True, + 'running_image': info['Image'], }