xref: /aosp_15_r20/external/autotest/server/hosts/gsc_devboard_host.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright (c) 2022 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
#
# Expects to be run in an environment with sudo and no interactive password
# prompt, such as within the Chromium OS development chroot.
"""Host class for GSC devboard connected host."""

import contextlib
import logging
try:
    import docker
except ImportError:
    logging.info("Docker API is not installed in this environment")

DOCKER_IMAGE = "gcr.io/satlab-images/gsc_dev_board:release"

SATLAB_DOCKER_HOST = 'tcp://192.168.231.1:2375'
LOCAL_DOCKER_HOST = 'tcp://127.0.0.1:2375'
DEFAULT_DOCKER_HOST = 'unix:///var/run/docker.sock'

DEFAULT_SERVICE_PORT = 39999

ULTRADEBUG = '18d1:0304'


class GSCDevboardHost(object):
    """
    A host that is physically connected to a GSC devboard.

    It could either be a SDK workstation (chroot) or a SatLab box.
    """

    def _initialize(self,
                    hostname,
                    service_debugger_serial=None,
                    service_ip=None,
                    service_port=DEFAULT_SERVICE_PORT,
                    *args,
                    **dargs):
        """Construct a GSCDevboardHost object.

        @hostname: Name of the devboard host, will be used in future to look up
                   the debugger serial, not currently used.
        @service_debugger_serial: debugger connected to devboard, defaults to
                                  the first one found on the container.
        @service_ip: devboard service ip, default is to start a new container.
        @service_port: devboard service port, defaults to 39999.
        """

        # Use docker host from environment or by probing a list of candidates.
        self._client = None
        try:
            self._client = docker.from_env()
            logging.info("Created docker host from env")
        except NameError:
            raise NameError('Please install docker using '
                            '"autotest/files/utils/install_docker_chroot.sh"')
        except docker.errors.DockerException:
            docker_host = None
            candidate_hosts = [
                    SATLAB_DOCKER_HOST, DEFAULT_DOCKER_HOST, LOCAL_DOCKER_HOST
            ]
            for h in candidate_hosts:
                try:
                    c = docker.DockerClient(base_url=h, timeout=2)
                    c.close()
                    docker_host = h
                    break
                except docker.errors.DockerException:
                    pass
            if docker_host is not None:
                self._client = docker.DockerClient(base_url=docker_host,
                                                   timeout=300)
            else:
                raise ValueError('Invalid DOCKER_HOST, ensure dockerd is'
                                 ' running.')
            logging.info("Using docker host at %s", docker_host)

        self._satlab = False
        # GSCDevboardHost should only be created on Satlab or localhost, so
        # assume Satlab if a drone container is running.
        if len(self._client.containers.list(filters={'name': 'drone'})) > 0:
            logging.info("In Satlab")
            self._satlab = True

        self._service_debugger_serial = service_debugger_serial
        self._service_ip = service_ip
        self._service_port = service_port
        logging.info("Using service port %s", self._service_port)

        self._docker_network = 'default_satlab' if self._satlab else 'host'
        self._docker_container = None

        serials = self._list_debugger_serials()
        if len(serials) == 0:
            raise ValueError('No debuggers found')
        logging.info("Available debuggers: [%s]", ', '.join(serials))

        if self._service_debugger_serial is None:
            self._service_debugger_serial = serials[0]
        else:
            if self._service_debugger_serial not in serials:
                raise ValueError(
                        '%s debugger not found in [%s]' %
                        (self._service_debugger_serial, ', '.join(serials)))
        logging.info("Using debugger %s", self._service_debugger_serial)
        self._docker_container_name = "gsc_dev_board_{}".format(
                self._service_debugger_serial)

    def _list_debugger_serials(self):
        """List all attached debuggers."""

        c = self._client.containers.run(DOCKER_IMAGE,
                                        remove=True,
                                        privileged=True,
                                        name='list_debugger_serial',
                                        hostname='list_debugger_serial',
                                        detach=True,
                                        volumes=["/dev:/hostdev"],
                                        command=['sleep', '5'])

        res, output = c.exec_run(['lsusb', '-v', '-d', ULTRADEBUG],
                                 stderr=False,
                                 privileged=True)
        c.kill()
        if res != 0:
            return []
        output = output.decode("utf-8").split('\n')
        serials = [
                l.strip().split(' ')[-1] for l in output
                if l.strip()[:7] == 'iSerial'
        ]
        return serials

    @contextlib.contextmanager
    def service_context(self):
        """Service context manager that provides the service endpoint."""
        self.start_service()
        try:
            yield "{}:{}".format(self.service_ip, self.service_port)
        finally:
            self.stop_service()

    def start_service(self):
        """Starts service if needed."""
        if self._docker_container is not None:
            return

        if self._service_ip:
            # Assume container was manually started if service_ip was set
            logging.info("Skip start_service due to set service_ip")
            return

        #TODO(b/215767105): Pull image onto Satlab box if not present.

        environment = {
                'DEVBOARDSVC_PORT': self._service_port,
                'DEBUGGER_SERIAL': self._service_debugger_serial
        }
        start_cmd = ['/opt/gscdevboard/start_devboardsvc.sh']

        # Stop any leftover containers
        try:
            c = self._client.containers.get(self._docker_container_name)
            c.kill()
        except docker.errors.NotFound:
            pass

        self._client.containers.run(DOCKER_IMAGE,
                                    remove=True,
                                    privileged=True,
                                    name=self._docker_container_name,
                                    hostname=self._docker_container_name,
                                    network=self._docker_network,
                                    cap_add=["NET_ADMIN"],
                                    detach=True,
                                    volumes=["/dev:/hostdev"],
                                    environment=environment,
                                    command=start_cmd)

        # A separate containers.get call is needed to capture network attributes
        self._docker_container = self._client.containers.get(
                self._docker_container_name)

    def stop_service(self):
        """Stops service by killing the container."""
        if self._docker_container is None:
            return
        self._docker_container.kill()
        self._docker_container = None

    @property
    def service_port(self):
        """Return service port (local to the container host)."""
        return self._service_port

    @property
    def service_ip(self):
        """Return service ip (local to the container host)."""
        if self._service_ip is not None:
            return self._service_ip

        if self._docker_network == 'host':
            return '127.0.0.1'
        else:
            if self._docker_container is None:
                return ''
            else:
                settings = self._docker_container.attrs['NetworkSettings']
                return settings['Networks'][self._docker_network]['IPAddress']