# 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']