<commit_before>"""
The runner knows about starting a container from a docker image.

It is responsible for also starting any dependency containers that are included
via ``volumes.share_with`` or ``link``.

It will also ensure containers are killed and removed after use.

Finally, the Runner is also responsible for starting and cleaning up intervention
containers.
"""

from harpoon.errors import BadOption, BadImage, BadResult, UserQuit, AlreadyBoundPorts
from harpoon.option_spec.harpoon_specs import HarpoonSpec
from harpoon import helpers as hp
from harpoon.helpers import until

from docker.errors import APIError as DockerAPIError
from input_algorithms.spec_base import NotSpecified
from input_algorithms.meta import Meta
from contextlib import contextmanager
from harpoon import dockerpty
from six.moves import input
import docker.errors
import logging
import socket
import uuid
import time
import six
import os

log = logging.getLogger("harpoon.ship.runner")

class Runner(object):
    """Knows how to run containers given Image objects"""

    ########################
    ###   USAGE
    ########################

    def run_container(self, conf, images, **kwargs):
        """Run this image and all dependency images"""
        with self._run_container(conf, images, **kwargs):
            pass

    @contextmanager
    def _run_container(self, conf, images, detach=False, started=None, dependency=False, tag=None, delete_anyway=False):
        if conf.container_id:
            yield
            return

        try:
            self.run_deps(conf, images)
            tty = not detach and (dependency or conf.harpoon.interactive)
            container_id = self.create_container(conf, detach, tty)

            conf.container_id = container_id

            try:
                self.wait_for_deps(conf, images)
            except KeyboardInterrupt:
                raise UserQuit()

            self.start_container(conf, tty=tty, detach=detach, is_dependency=dependency)
            yield
        finally:
            if delete_anyway or (not detach and not dependency):
                self.stop_deps(conf, images)
                self.stop_container(conf, tag=tag)
                self.delete_deps(conf, images)

    ########################
    ###   DEPS
    ########################

    def delete_deps(self, conf, images):
        """Delete any deleteable images"""
        for dependency_name, _ in conf.dependency_images():
            image = images[dependency_name]
            if image.deleteable_image:
                log.info("Removing un-needed image {0}".format(image.image_name))
                conf.harpoon.docker_api.remove_image(image.image_name)

    def run_deps(self, conf, images):
        """Start containers for all our dependencies"""
        for dependency_name, detached in conf.dependency_images(for_running=True):
            try:
                self.run_container(images[dependency_name], images, detach=detached, dependency=True)
            except Exception as error:
                raise BadImage("Failed to start dependency container", image=conf.name, dependency=dependency_name, error=error)

    def stop_deps(self, conf, images):
        """Stop the containers for all our dependencies"""
        for dependency, _ in conf.dependency_images():
            self.stop_deps(images[dependency], images)
            try:
                self.stop_container(images[dependency], fail_on_bad_exit=True, fail_reason="Failed to run dependency container")
            except BadImage:
                raise
            except Exception as error:
                log.warning("Failed to stop dependency container\timage=%s\tdependency=%s\tcontainer_name=%s\terror=%s", conf.name, dependency, images[dependency].container_name, error)

    def wait_for_deps(self, conf, images):
        """Wait for all our dependencies"""
        from harpoon.option_spec.image_objs import WaitCondition
        api = conf.harpoon.docker_context_maker().api

        waited = set()
        last_attempt = {}
        dependencies = set(dep for dep, _ in conf.dependency_images())

        # Wait conditions come from dependency_options first
        # Or if none specified there, they come from the image itself
        wait_conditions = {}
        for dependency in dependencies:
            if conf.dependency_options is not NotSpecified and dependency in conf.dependency_options and conf.dependency_options[dependency].wait_condition is not NotSpecified:
                wait_conditions[dependency] = conf.dependency_options[dependency].wait_condition
            elif images[dependency].wait_condition is not NotSpecified:
                wait_conditions[dependency] = images[dependency].wait_condition

        if not wait_conditions:
            return

        start = time.time()
        while True:
            this_round = []
            for dependency in dependencies:
                if dependency in waited:
                    continue

                image = images[dependency]
                if dependency in wait_conditions:
                    done = self.wait_for_dep(api, image, wait_conditions[dependency], start, last_attempt.get(dependency))
                    this_round.append(done)
                    if done is True:
                        waited.add(dependency)
                    elif done is False:
                        last_attempt[dependency] = time.time()
                    elif done is WaitCondition.Timedout:
                        log.warning("Stopping dependency because it timedout waiting\tcontainer_id=%s", image.container_id)
                        self.stop_container(image)
                else:
                    waited.add(dependency)

            if set(this_round) != set([WaitCondition.KeepWaiting]):
                if dependencies - waited == set():
                    log.info("Finished waiting for dependencies")
                    break
                else:
                    log.info("Still waiting for dependencies\twaiting_on=%s", list(dependencies-waited))

                couldnt_wait = set()
                container_ids = {}
                for dependency in dependencies:
                    if dependency in waited:
                        continue

                    image = images[dependency]
                    if image.container_id is None:
                        stopped = True
                        if dependency not in container_ids:
                            available = sorted([i for i in available if "/{0}".format(image.container_name) in i["Names"]], key=lambda i: i["Created"])
                            if available:
                                container_ids[dependency] = available[0]["Id"]
                    else:
                        if dependency not in container_ids:
                            container_ids[dependency] = image.container_id
                        stopped, _ = self.is_stopped(image, image.container_id)

                    if stopped:
                        couldnt_wait.add(dependency)

                if couldnt_wait:
                    for container in couldnt_wait:
                        if container not in images or container not in container_ids:
                            continue
                        image = images[container]
                        container_id = container_ids[container]
                        container_name = image.container_name
                        hp.write_to(conf.harpoon.stdout, "=================== Logs for failed container {0} ({1})\n".format(container_id, container_name))
                        for line in conf.harpoon.docker_api.logs(container_id).split("\n"):
                            hp.write_to(conf.harpoon.stdout, "{0}\n".format(line))
                        hp.write_to(conf.harpoon.stdout, "------------------- End logs for failed container\n")
                    raise BadImage("One or more of the dependencies stopped running whilst waiting for other dependencies", stopped=list(couldnt_wait))

            time.sleep(0.1)

    def wait_for_dep(self, api, conf, wait_condition, start, last_attempt):<commit_msg>Make sure logs are decoded when there is an error<commit_after>"""
The runner knows about starting a container from a docker image.

It is responsible for also starting any dependency containers that are included
via ``volumes.share_with`` or ``link``.

It will also ensure containers are killed and removed after use.

Finally, the Runner is also responsible for starting and cleaning up intervention
containers.
"""

from harpoon.errors import BadOption, BadImage, BadResult, UserQuit, AlreadyBoundPorts
from harpoon.option_spec.harpoon_specs import HarpoonSpec
from harpoon import helpers as hp
from harpoon.helpers import until

from docker.errors import APIError as DockerAPIError
from input_algorithms.spec_base import NotSpecified
from input_algorithms.meta import Meta
from contextlib import contextmanager
from harpoon import dockerpty
from six.moves import input
import docker.errors
import logging
import socket
import uuid
import time
import six
import os

log = logging.getLogger("harpoon.ship.runner")

class Runner(object):
    """Knows how to run containers given Image objects"""

    ########################
    ###   USAGE
    ########################

    def run_container(self, conf, images, **kwargs):
        """Run this image and all dependency images"""
        with self._run_container(conf, images, **kwargs):
            pass

    @contextmanager
    def _run_container(self, conf, images, detach=False, started=None, dependency=False, tag=None, delete_anyway=False):
        if conf.container_id:
            yield
            return

        try:
            self.run_deps(conf, images)
            tty = not detach and (dependency or conf.harpoon.interactive)
            container_id = self.create_container(conf, detach, tty)

            conf.container_id = container_id

            try:
                self.wait_for_deps(conf, images)
            except KeyboardInterrupt:
                raise UserQuit()

            self.start_container(conf, tty=tty, detach=detach, is_dependency=dependency)
            yield
        finally:
            if delete_anyway or (not detach and not dependency):
                self.stop_deps(conf, images)
                self.stop_container(conf, tag=tag)
                self.delete_deps(conf, images)

    ########################
    ###   DEPS
    ########################

    def delete_deps(self, conf, images):
        """Delete any deleteable images"""
        for dependency_name, _ in conf.dependency_images():
            image = images[dependency_name]
            if image.deleteable_image:
                log.info("Removing un-needed image {0}".format(image.image_name))
                conf.harpoon.docker_api.remove_image(image.image_name)

    def run_deps(self, conf, images):
        """Start containers for all our dependencies"""
        for dependency_name, detached in conf.dependency_images(for_running=True):
            try:
                self.run_container(images[dependency_name], images, detach=detached, dependency=True)
            except Exception as error:
                raise BadImage("Failed to start dependency container", image=conf.name, dependency=dependency_name, error=error)

    def stop_deps(self, conf, images):
        """Stop the containers for all our dependencies"""
        for dependency, _ in conf.dependency_images():
            self.stop_deps(images[dependency], images)
            try:
                self.stop_container(images[dependency], fail_on_bad_exit=True, fail_reason="Failed to run dependency container")
            except BadImage:
                raise
            except Exception as error:
                log.warning("Failed to stop dependency container\timage=%s\tdependency=%s\tcontainer_name=%s\terror=%s", conf.name, dependency, images[dependency].container_name, error)

    def wait_for_deps(self, conf, images):
        """Wait for all our dependencies"""
        from harpoon.option_spec.image_objs import WaitCondition
        api = conf.harpoon.docker_context_maker().api

        waited = set()
        last_attempt = {}
        dependencies = set(dep for dep, _ in conf.dependency_images())

        # Wait conditions come from dependency_options first
        # Or if none specified there, they come from the image itself
        wait_conditions = {}
        for dependency in dependencies:
            if conf.dependency_options is not NotSpecified and dependency in conf.dependency_options and conf.dependency_options[dependency].wait_condition is not NotSpecified:
                wait_conditions[dependency] = conf.dependency_options[dependency].wait_condition
            elif images[dependency].wait_condition is not NotSpecified:
                wait_conditions[dependency] = images[dependency].wait_condition

        if not wait_conditions:
            return

        start = time.time()
        while True:
            this_round = []
            for dependency in dependencies:
                if dependency in waited:
                    continue

                image = images[dependency]
                if dependency in wait_conditions:
                    done = self.wait_for_dep(api, image, wait_conditions[dependency], start, last_attempt.get(dependency))
                    this_round.append(done)
                    if done is True:
                        waited.add(dependency)
                    elif done is False:
                        last_attempt[dependency] = time.time()
                    elif done is WaitCondition.Timedout:
                        log.warning("Stopping dependency because it timedout waiting\tcontainer_id=%s", image.container_id)
                        self.stop_container(image)
                else:
                    waited.add(dependency)

            if set(this_round) != set([WaitCondition.KeepWaiting]):
                if dependencies - waited == set():
                    log.info("Finished waiting for dependencies")
                    break
                else:
                    log.info("Still waiting for dependencies\twaiting_on=%s", list(dependencies-waited))

                couldnt_wait = set()
                container_ids = {}
                for dependency in dependencies:
                    if dependency in waited:
                        continue

                    image = images[dependency]
                    if image.container_id is None:
                        stopped = True
                        if dependency not in container_ids:
                            available = sorted([i for i in available if "/{0}".format(image.container_name) in i["Names"]], key=lambda i: i["Created"])
                            if available:
                                container_ids[dependency] = available[0]["Id"]
                    else:
                        if dependency not in container_ids:
                            container_ids[dependency] = image.container_id
                        stopped, _ = self.is_stopped(image, image.container_id)

                    if stopped:
                        couldnt_wait.add(dependency)

                if couldnt_wait:
                    for container in couldnt_wait:
                        if container not in images or container not in container_ids:
                            continue
                        image = images[container]
                        container_id = container_ids[container]
                        container_name = image.container_name
                        hp.write_to(conf.harpoon.stdout, "=================== Logs for failed container {0} ({1})\n".format(container_id, container_name))

                        logs = conf.harpoon.docker_api.logs(container_id)
                        if isinstance(logs, six.binary_type):
                            logs = logs.decode()
                        for line in logs.split("\n"):
                            hp.write_to(conf.harpoon.stdout, "{0}\n".format(line))

                        hp.write_to(conf.harpoon.stdout, "------------------- End logs for failed container\n")
                    raise BadImage("One or more of the dependencies stopped running whilst waiting for other dependencies", stopped=list(couldnt_wait))

            time.sleep(0.1)

    def wait_for_dep(self, api, conf, wait_condition, start, last_attempt):
{'fraction_non_alphanumeric': 0.08972415883601091, 'fraction_numerical': 0.0009699909063352532, 'mean_word_length': 2.045227985970094, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3673048', 'n_tokens_mistral': 4139, 'n_tokens_neox': 3857, 'n_words': 1139}