scheduler/luciferlib.py

# Copyright 2017 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.

"""Library providing an API to lucifer."""

import os
import logging
import pipes
import socket
import subprocess

import common
from autotest_lib.client.bin import local_host
from autotest_lib.client.common_lib import global_config
from autotest_lib.scheduler.drone_manager import PidfileId
from autotest_lib.server.hosts import ssh_host
from autotest_lib.frontend.afe import models

_config = global_config.global_config
_SECTION = 'LUCIFER'

# TODO(crbug.com/748234): Move these to shadow_config.ini
# See also drones.AUTOTEST_INSTALL_DIR
_ENV = '/usr/bin/env'
_AUTOTEST_DIR = '/usr/local/autotest'
_JOB_REPORTER_PATH = os.path.join(_AUTOTEST_DIR, 'bin', 'job_reporter')

logger = logging.getLogger(__name__)


def is_lucifer_enabled():
    """Return True if lucifer is enabled in the config."""
    return True


def is_enabled_for(level):
    """Return True if lucifer is enabled for the given level.

    @param level: string, e.g. 'PARSING', 'GATHERING'
    """
    if not is_lucifer_enabled():
        return False
    config_level = (_config.get_config_value(_SECTION, 'lucifer_level')
                    .upper())
    return level.upper() == config_level


def is_lucifer_owned(job):
    """Return True if job is already sent to lucifer.

    @param job: frontend.afe.models.Job instance
    """
    assert isinstance(job, models.Job)
    return hasattr(job, 'jobhandoff')


def is_lucifer_owned_by_id(job_id):
    """Return True if job is already sent to lucifer."""
    return models.JobHandoff.objects.filter(job_id=job_id).exists()


def is_split_job(hqe_id):
    """Return True if HQE is part of a job with HQEs in a different group.

    For examples if the given HQE have execution_subdir=foo and the job
    has an HQE with execution_subdir=bar, then return True.  The only
    situation where this happens is if provisioning in a multi-DUT job
    fails, the HQEs will each be in their own group.

    See https://bugs.chromium.org/p/chromium/issues/detail?id=811877

    @param hqe_id: HQE id
    """
    hqe = models.HostQueueEntry.objects.get(id=hqe_id)
    hqes = hqe.job.hostqueueentry_set.all()
    try:
        _get_consistent_execution_path(hqes)
    except ExecutionPathError:
        return True
    return False


# TODO(crbug.com/748234): This is temporary to enable toggling
# lucifer rollouts with an option.
def spawn_starting_job_handler(manager, job):
    """Spawn job_reporter to handle a job.

    Pass all arguments by keyword.

    @param manager: scheduler.drone_manager.DroneManager instance
    @param job: Job instance
    @returns: Drone instance
    """
    manager = _DroneManager(manager)
    drone = manager.pick_drone_to_use()
    results_dir = _results_dir(manager, job)
    args = [
            _JOB_REPORTER_PATH,

            # General configuration
            '--jobdir', _get_jobdir(),
            '--lucifer-path', _get_lucifer_path(),

            # Job specific
            '--lucifer-level', 'STARTING',
            '--job-id', str(job.id),
            '--results-dir', results_dir,

            # STARTING specific
            '--execution-tag', _working_directory(job),
    ]
    if _get_gcp_creds():
        args = [
                'GOOGLE_APPLICATION_CREDENTIALS=%s'
                % pipes.quote(_get_gcp_creds()),
        ] + args
    drone.spawn(_ENV, args,
                output_file=_prepare_output_file(drone, results_dir))
    drone.add_active_processes(1)
    manager.reorder_drone_queue()
    manager.register_pidfile_processes(
            os.path.join(results_dir, '.autoserv_execute'), 1)
    return drone


# TODO(crbug.com/748234): This is temporary to enable toggling
# lucifer rollouts with an option.
def spawn_parsing_job_handler(manager, job, autoserv_exit, pidfile_id=None):
    """Spawn job_reporter to handle a job.

    Pass all arguments by keyword.

    @param manager: scheduler.drone_manager.DroneManager instance
    @param job: Job instance
    @param autoserv_exit: autoserv exit status
    @param pidfile_id: PidfileId instance
    @returns: Drone instance
    """
    manager = _DroneManager(manager)
    if pidfile_id is None:
        drone = manager.pick_drone_to_use()
    else:
        drone = manager.get_drone_for_pidfile(pidfile_id)
    results_dir = _results_dir(manager, job)
    args = [
            _JOB_REPORTER_PATH,

            # General configuration
            '--jobdir', _get_jobdir(),
            '--lucifer-path', _get_lucifer_path(),

            # Job specific
            '--job-id', str(job.id),
            '--lucifer-level', 'STARTING',
            '--parsing-only',
            '--results-dir', results_dir,
    ]
    if _get_gcp_creds():
        args = [
                'GOOGLE_APPLICATION_CREDENTIALS=%s'
                % pipes.quote(_get_gcp_creds()),
        ] + args
    drone.spawn(_ENV, args,
                output_file=_prepare_output_file(drone, results_dir))
    drone.add_active_processes(1)
    manager.reorder_drone_queue()
    manager.register_pidfile_processes(
            os.path.join(results_dir, '.autoserv_execute'), 1)
    return drone


_LUCIFER_DIR = 'lucifer'


def _prepare_output_file(drone, results_dir):
    logdir = os.path.join(results_dir, _LUCIFER_DIR)
    drone.run('mkdir', ['-p', logdir])
    return os.path.join(logdir, 'job_reporter_output.log')


def _get_jobdir():
    return _config.get_config_value(_SECTION, 'jobdir')


def _get_lucifer_path():
    return os.path.join(_get_binaries_path(), 'lucifer')


def _get_binaries_path():
    """Get binaries dir path from config.."""
    return _config.get_config_value(_SECTION, 'binaries_path')


def _get_gcp_creds():
    """Return path to GCP service account credentials.

    This is the empty string by default, if no credentials will be used.
    """
    return _config.get_config_value(_SECTION, 'gcp_creds', default='')


class _DroneManager(object):
    """Simplified drone API."""

    def __init__(self, old_manager):
        """Initialize instance.

        @param old_manager: old style DroneManager
        """
        self._manager = old_manager

    def get_num_tests_failed(self, pidfile_id):
        """Return the number of tests failed for autoserv by pidfile.

        @param pidfile_id: PidfileId instance.
        @returns: int (-1 if missing)
        """
        state = self._manager.get_pidfile_contents(pidfile_id)
        if state.num_tests_failed is None:
            return -1
        return state.num_tests_failed

    def get_drone_for_pidfile(self, pidfile_id):
        """Return a drone to use from a pidfile.

        @param pidfile_id: PidfileId instance.
        """
        return _wrap_drone(self._manager.get_drone_for_pidfile_id(pidfile_id))

    def pick_drone_to_use(self, num_processes=1):
        """Return a drone to use.

        Various options can be passed to optimize drone selection.

        @param num_processes: number of processes the drone is intended
            to run
        """
        old_drone = self._manager.pick_drone_to_use(
                num_processes=num_processes,
        )
        return _wrap_drone(old_drone)

    def absolute_path(self, path):
        """Return absolute path for drone results.

        The returned path might be remote.
        """
        return self._manager.absolute_path(path)

    def register_pidfile_processes(self, path, count):
        """Register a pidfile with the given number of processes.

        This should be done to allow the drone manager to check the
        number of processes still alive.  This may be used to select
        drones based on the number of active processes as a proxy for
        load.

        The exact semantics depends on the drone manager implementation;
        implementation specific comments follow:

        Pidfiles are kept in memory to track process count.  Pidfiles
        are rediscovered when the scheduler restarts.  Thus, errors in
        pidfile tracking can be fixed by restarting the scheduler.xo
        """
        pidfile_id = PidfileId(path)
        self._manager.register_pidfile(pidfile_id)
        self._manager._registered_pidfile_info[pidfile_id].num_processes = count

    def reorder_drone_queue(self):
        """Reorder drone queue according to modified process counts.

        Call this after Drone.add_active_processes().
        """
        self._manager.reorder_drone_queue()


def _wrap_drone(old_drone):
    """Wrap an old style drone."""
    host = old_drone._host
    if isinstance(host, local_host.LocalHost):
        return LocalDrone()
    elif isinstance(host, ssh_host.SSHHost):
        return RemoteDrone(old_drone)
    else:
        raise TypeError('Drone has an unknown host type')


def _results_dir(manager, job):
    """Return results dir for a job.

    Path may be on a remote host.
    """
    return manager.absolute_path(_working_directory(job))


def _working_directory(job):
    return _get_consistent_execution_path(job.hostqueueentry_set.all())


def _get_consistent_execution_path(execution_entries):
    first_execution_path = execution_entries[0].execution_path()
    for execution_entry in execution_entries[1:]:
        if execution_entry.execution_path() != first_execution_path:
            raise ExecutionPathError(
                    '%s (%s) != %s (%s)'
                    % (execution_entry.execution_path(),
                       execution_entry,
                       first_execution_path,
                       execution_entries[0]))
    return first_execution_path


class ExecutionPathError(Exception):
    """Raised by _get_consistent_execution_path()."""


class Drone(object):
    """Simplified drone API."""

    def hostname(self):
        """Return the hostname of the drone."""

    def run(self, path, args):
        """Run a command synchronously.

        path must be an absolute path.  path may be on a remote machine.
        args is a list of arguments.

        The process may or may not have its own session.  The process
        should be short-lived.  It should not try to obtain a
        controlling terminal.

        The new process will have stdin, stdout, and stderr opened to
        /dev/null.

        This method intentionally has a very restrictive API.  It should
        be used to perform setup local to the drone, when the drone may
        be a remote machine.
        """

    def spawn(self, path, args, output_file):
        """Spawn an independent process.

        path must be an absolute path.  path may be on a remote machine.
        args is a list of arguments.

        The process is spawned in its own session.  It should not try to
        obtain a controlling terminal.

        The new process will have stdin opened to /dev/null and stdout,
        stderr opened to output_file.

        output_file is a pathname, but how it is interpreted is
        implementation defined, e.g., it may be a remote file.
        """

    def add_active_processes(self, count):
        """Track additional number of active processes.

        This may be used to select drones based on the number of active
        processes as a proxy for load.

        _DroneManager.register_pidfile_processes() and
        _DroneManager.reorder_drone_queue() should also be called.

        The exact semantics depends on the drone manager implementation;
        implementation specific comments follow:

        Process count is used as a proxy for workload, and one process
        equals the workload of one autoserv or one job.  This count is
        recalculated during each scheduler tick, using pidfiles tracked
        by the drone manager (so the count added by this function only
        applies for one tick).
        """


class LocalDrone(Drone):
    """Local implementation of Drone."""

    def hostname(self):
        return socket.gethostname()

    def run(self, path, args):
        with open(os.devnull, 'r+b') as null:
            subprocess.call([path] + args, stdin=null,
                            stdout=null, stderr=null)

    def spawn(self, path, args, output_file):
        _spawn(path, [path] + args, output_file)


class RemoteDrone(Drone):
    """Remote implementation of Drone through SSH."""

    def __init__(self, drone):
        host = drone._host
        if not isinstance(host, ssh_host.SSHHost):
            raise TypeError('RemoteDrone must be passed a drone with SSHHost')
        self._drone = drone
        self._host = drone._host

    def hostname(self):
        return self._host.hostname

    def run(self, path, args):
        cmd_parts = [path] + args
        safe_cmd = ' '.join(pipes.quote(part) for part in cmd_parts)
        self._host.run('%(cmd)s <%(null)s >%(null)s 2>&1'
                       % {'cmd': safe_cmd, 'null': os.devnull})

    def spawn(self, path, args, output_file):
        cmd_parts = [path] + args
        safe_cmd = ' '.join(pipes.quote(part) for part in cmd_parts)
        safe_file = pipes.quote(output_file)
        # SSH creates a session for each command, so we do not have to
        # do it.
        self._host.run('%(cmd)s <%(null)s >>%(file)s 2>&1 &'
                       % {'cmd': safe_cmd,
                          'file': safe_file,
                          'null': os.devnull})

    def add_active_processes(self, count):
        self._drone.active_processes += count


def _spawn(path, argv, output_file):
    """Spawn a new process in its own session.

    path must be an absolute path.  The first item in argv should be
    path.

    In the calling process, this function returns on success.
    The forked process puts itself in its own session and execs.

    The new process will have stdin opened to /dev/null and stdout,
    stderr opened to output_file.
    """
    logger.info('Spawning %r, %r, %r', path, argv, output_file)
    assert all(isinstance(arg, basestring) for arg in argv)
    pid = os.fork()
    if pid:
        os.waitpid(pid, 0)
        return
    # Double fork to reparent to init since monitor_db does not reap.
    if os.fork():
        os._exit(os.EX_OK)
    os.setsid()
    null_fd = os.open(os.devnull, os.O_RDONLY)
    os.dup2(null_fd, 0)
    os.close(null_fd)
    out_fd = os.open(output_file, os.O_WRONLY | os.O_APPEND | os.O_CREAT)
    os.dup2(out_fd, 1)
    os.dup2(out_fd, 2)
    os.close(out_fd)
    os.execv(path, argv)