blob: 3a368039207848f7195630ad78ac07a4aaaed7cf [file] [log] [blame]
# Copyright 2018 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import os
import pipes
import socket
import time
from import containers
UNIVERSAL_CROS_HOSTNAME = 'variable_chromeos_device_hostname'
class CrosContainerDescriptor(containers.ContainerDescriptorBase):
def __init__(self, device_hostname, ssh_id_path):
super(CrosContainerDescriptor, self).__init__()
self._device_hostname = device_hostname
self._ssh_id_path = ssh_id_path
def name(self):
return 'cros_%s' % self._device_hostname
def shutdown_file(self):
return '/b/%s.shutdown.stamp' % self._device_hostname
def lock_file(self):
return '/var/lock/cros_docker.%s.lock' % self._device_hostname
def hostname(self):
this_host = socket.gethostname().split('.')[0]
return '%s--%s' % (this_host, self._device_hostname)
def device_hostname(self):
return self._device_hostname
def ssh_id_path(self):
return self._ssh_id_path
def log_started(self):
logging.debug('Launched new container (%s) for device %s.',, self._device_hostname)
class CrosDockerClient(containers.DockerClient):
def __init__(self):
super(CrosDockerClient, self).__init__()
# The 'cros flash' tool creates loopback devices to mount the CrOS images.
# Due to a bug, Docker is unable to create new loop devices in /dev/ within
# a container. So mount the host's /dev/ within each container as a
# workaround. See for more info.
self.volumes['/dev/'] = '/dev/'
def create_container(self, container_desc, image_name, swarming_url, labels,
assert isinstance(container_desc, CrosContainerDescriptor)
env = {
CROS_SSH_ID_ENV_VAR: container_desc.ssh_id_path,
# The `cros flash` tool uses IPv6 loopback to communicate to the local dev
# server. IPv6 is disabled by default on later versions of Docker, so
# explicitly enable it.
sysctls = {
'net.ipv6.conf.lo.disable_ipv6': 0,
# Create the container with elevated privileges so the CrOS chroot (needed
# for DUT flashing) can be created.
container = super(CrosDockerClient, self).create_container(
container_desc, image_name, swarming_url, labels,
additional_env=env, privileged=True, sysctls=sysctls)
# Add an entry to /etc/hosts that points a universal hostname to the
# cros device.
ip = None
ip = socket.gethostbyname(container_desc.device_hostname)'Fetched IPv4 of %s: %s', container_desc.device_hostname, ip)
except socket.gaierror:
'Unable to get IPv4 of %s. Hosts file will remain unchanged.',
if ip:
append_cmd = 'echo "%s %s" >> /etc/hosts' % (ip, UNIVERSAL_CROS_HOSTNAME)
out = container.exec_run('/bin/bash -c %s' % pipes.quote(append_cmd))'Result from modifying hosts file: %s', out)
return container