blob: 0f353d2cd747d6cb05e3740c82dfcdd702048556 [file] [log] [blame]
# -*- coding: utf-8 -*-
# Copyright 2019 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.
"""Interface to the AFE RPC server for all moblab code."""
import http.client
import json
import sys
import multiprocessing
from datetime import datetime
import logging
import docker
import requests
MOBLAB_RPC_SERVER = "autotest-afe:8888"
_LOGGER = logging.getLogger(__name__)
_LOGGER.setLevel(logging.DEBUG)
def enroll_dut(ip_address):
client = docker.from_env()
afe_container = client.containers.get("autotest-afe")
cmd = "/usr/local/autotest/cli/atest host create %s" % ip_address
_LOGGER.debug(cmd)
(_, output) = afe_container.exec_run(cmd)
_LOGGER.debug(output)
class AFEConnector(object):
"""Class that provides moblab an inferface to AFE."""
def __init__(self):
"""Setup the JSON encoder/decoder."""
self.json_encoder = json.encoder.JSONEncoder()
self.json_decoder = json.decoder.JSONDecoder()
def send_rpc_command(self, method, params=None):
"""Call the AFE RPC server.
Args:
method (string): The name of the AFE RPC
params (string, optional): Defaults to None. A JSON encoded string
with any paramaters required for the RPC.
"""
if not params:
params = {}
rpc_url = "http://%s/afe/server/rpc/" % MOBLAB_RPC_SERVER
data = self.json_encoder.encode(
{"id": 0, "method": method, "params": [params]}
)
_LOGGER.debug("URL: %s Data: %s", rpc_url, data)
try:
response = requests.post(rpc_url, data=data)
error = self.decode_response(response, "error")
if error:
_LOGGER.error("AFE %s failed with error %s", method, error)
_LOGGER.debug("Response: %s", response.text)
return response
except requests.exceptions.RequestException as e:
_LOGGER.error(e)
def decode_response(self, response, key, expected_code=http.client.OK):
"""Get a specific value from the return of an afe call.
Args:
response (string): json formatted data.
key (string): they key of the data to retrieve.
expected_code (int, optional): Defaults to http.client.OK.
Returns:
dict: {} if the AFE server returned an error else a map k,v
of the response.
"""
if not response or response.status_code != expected_code:
if response:
# TODO(haddowk): Figure out how to return a useful error
# message.
_LOGGER.error("AFE RPC Failed %s", response.get("error", ""))
return {}
return self.json_decoder.decode(response.text)[key]
def get_config_values(self):
"""Request all the configuration details from AFE.
Returns:
dict: { <section name> : { <config_name> : <config_value> } }
"""
response = self.send_rpc_command("get_config_values")
raw_result = self.decode_response(response, "result")
result = {}
if not raw_result:
return result
for k, v in list(raw_result.items()):
try:
result[k] = dict(v)
except ValueError:
_LOGGER.error(
"Error getting config values key: %s value: %s", k, v
)
return result
def get_connected_devices(self, hostname=None, get_current_jobs=False):
"""Get the list of hosts (DUT's) configured in AFE ( which match
the specified hostname, if provided).
hostname (string): ip for which to return information.
get_current_jobs (bool): return info about curently executing
jobs and special tasks on DUT.
Returns:
A list of dictionaries, each dict with host information.
"""
params = {}
if hostname:
params["hostname"] = hostname
if get_current_jobs:
params["include_current_job"] = "true"
response = self.send_rpc_command("get_hosts", params)
return self.decode_response(response, "result")
def get_num_jobs(
self,
id_filter=None,
name_filter=None,
created_time_lt=None,
created_time_gt=None,
sub=False,
suite=False,
not_yet_run=False,
running=False,
finished=False,
completed_status=None,
parent_id_filter=None,
):
"""
Returns number of jobs that match filter arguments. Any arg that
evaluates to false is not used for filtering.
Args:
id_filter: Filter for only jobs that contain id string, this is a
contains match rather than an exact match.
name_filter: Filter for only jobs that contain name string, this
is a contains match rather than an exact match.
created_time_lt: Upper bound created time filter. Second
granularity timestamp.
created_time_gt: Lower bound created time filter. Second
granularity timestamp.
sub: true iff only child jobs are not to be filtered.
suite: true iff only parent jobs are not to be filtered.
not_yet_run: true iff only not-yet-run jobs are not to be filtered.
running: true iff only running jobs are not to be filtered.
finished: true iff only finished jobs are not to be filtered.
parent_id_filter: Filter for only jobs that are children of this
job id, this is an exact match:
Returns:
A dict containing 'result' number.
"""
params = {
"completed_status": completed_status,
}
if id_filter:
params["id__icontains"] = id_filter
if name_filter:
params["name__icontains"] = name_filter
if created_time_lt:
params["created_on__lte"] = str(
datetime.fromtimestamp(created_time_lt).strftime(
"%Y-%m-%d %H:%M:%S"
)
)
if created_time_gt:
params["created_on__gte"] = str(
datetime.fromtimestamp(created_time_gt).strftime(
"%Y-%m-%d %H:%M:%S"
)
)
if sub:
params["sub"] = "true"
elif suite:
params["suite"] = "true"
if not_yet_run:
params["not_yet_run"] = "true"
elif running:
params["running"] = "true"
elif finished:
params["finished"] = "true"
if parent_id_filter:
params["parent_job"] = parent_id_filter
response = self.send_rpc_command("get_num_jobs", params)
_LOGGER.error(response)
return self.decode_response(response, "result")
def get_jobs(
self,
query_start,
query_limit,
id_filter=None,
name_filter=None,
created_time_lt=None,
created_time_gt=None,
sub=False,
suite=False,
not_yet_run=False,
running=False,
finished=False,
completed_status=None,
parent_id_filter=None,
sort_by=None,
):
"""Get attributes of the jobs specified.
Args:
id_filter: Filter for only jobs that contain id string, this is a
contains match rather than an exact match.
name_filter: Filter for only jobs that contain name string, this
is a contains match rather than an exact match.
created_time_lt: Upper bound created time filter. Second
granularity timestamp.
created_time_gt: Lower bound created time filter. Second
granularity timestamp.
sub: true iff only child jobs are not to be filtered.
suite: true iff only parent jobs are not to be filtered.
not_yet_run: true iff only not-yet-run jobs are not to be filtered.
running: true iff only running jobs are not to be filtered.
finished: true iff only finished jobs are not to be filtered.
parent_id_filter: Filter for only jobs that are children of this
Returns:
A list of dict. The key, value of the dict is the job attribute
name and the value.
"""
if not sort_by:
sort_by = "-created_on"
params = {
"query_start": query_start,
"query_limit": query_limit,
"sort_by": [sort_by],
"completed_status": completed_status,
}
if id_filter:
params["id__iexact"] = id_filter
if name_filter:
params["name__icontains"] = name_filter
if created_time_lt:
params["created_on__lte"] = str(
datetime.fromtimestamp(created_time_lt).strftime(
"%Y-%m-%d %H:%M:%S"
)
)
if created_time_gt:
params["created_on__gte"] = str(
datetime.fromtimestamp(created_time_gt).strftime(
"%Y-%m-%d %H:%M:%S"
)
)
if sub:
params["sub"] = "true"
elif suite:
params["suite"] = "true"
if not_yet_run:
params["not_yet_run"] = "true"
elif running:
params["running"] = "true"
elif finished:
params["finished"] = "true"
if parent_id_filter:
params["parent_job"] = parent_id_filter
response = self.send_rpc_command("get_jobs_summary", params)
return self.decode_response(response, "result")
def get_jobs_by_ids(self, id_list):
assert isinstance(id_list, list)
params = {"id__in": id_list}
response = self.send_rpc_command("get_jobs_summary", params)
return self.decode_response(response, "result")
def get_job_history(self, job_id):
params = {"job_id": str(job_id)}
response = self.send_rpc_command("get_job_history", params)
return self.decode_response(response, "result")
def get_num_special_tasks(
self, id_list=[], host_id_list=[], start_time_gt=None, end_time_lt=None
):
"""Get attributes of the special tasks specified.
Args:
id_list: List of special task id's,
host_id_list: List of AFE host id's
start_time_gt(int): Seconds timestamp of a filter to DUT tasks by
start time.
end_time_lt(int): Seconds timestamp of a filter to DUT tasks by
end time.
Returns:
Number of tasks that match the given filters ( special_tasks in AFE
vernacular ).
"""
params = {}
if id_list:
assert isinstance(id_list, list)
params = {"id__in": id_list}
if host_id_list:
assert isinstance(host_id_list, list)
params = {"host__in": host_id_list}
if start_time_gt:
params["start_time"] = str(
datetime.fromtimestamp(start_time_gt).strftime(
"%Y-%m-%d %H:%M:%S"
)
)
if end_time_lt:
params["end_time"] = str(
datetime.fromtimestamp(end_time_lt).strftime(
"%Y-%m-%d %H:%M:%S"
)
)
response = self.send_rpc_command("get_num_special_tasks", params)
return self.decode_response(response, "result")
def get_special_tasks(
self,
id_list=[],
host_id_list=[],
query_start=None,
query_limit=None,
start_time_gt=None,
end_time_lt=None,
sort_by_most_recent=True,
):
"""Get attributes of the special tasks specified.
Args:
id_list: A list of special task id to be retrieved.
host_id_list: A list of AFE host id's to filter results by.
query_start(int): Parameter specifying index on where to start when
returning paged results.
query_limit(int): Upper limit on number of results to return.
start_time_gt(int): Seconds timestamp of a filter to DUT tasks by
start time.
end_time_lt(int): Seconds timestamp of a filter to DUT tasks by
end time.
Returns:
A list of dict. The key, value of the dict is the special task
attribute name and the value.
"""
params = {}
if id_list:
assert isinstance(id_list, list)
params["id__in"] = id_list
if host_id_list:
assert isinstance(host_id_list, list)
params["host__in"] = host_id_list
if query_start is not None and query_limit is not None:
params["query_start"] = query_start
params["query_limit"] = query_limit
if start_time_gt:
params["start_time"] = str(
datetime.fromtimestamp(start_time_gt).strftime(
"%Y-%m-%d %H:%M:%S"
)
)
if end_time_lt:
params["end_time"] = str(
datetime.fromtimestamp(end_time_lt).strftime(
"%Y-%m-%d %H:%M:%S"
)
)
if sort_by_most_recent:
params["sort_by"] = ["-id"]
response = self.send_rpc_command("get_special_tasks", params)
return self.decode_response(response, "result")
def get_num_host_queue_entries(self, job_id=None, host_id=None):
"""
Get number of HQE's that match given query parameters.
Args:
job_id(int): The job id which HQEs associated with.
host_id(int): The AFE DUT id which HQEs associated with.
Returns:
An integer.
"""
params = {}
if job_id:
params["job__id"] = job_id
if host_id:
params["host__id"] = host_id
response = self.send_rpc_command("get_num_host_queue_entries", params)
return self.decode_response(response, "result")
def get_host_queue_entries(
self, job_id=None, host_id=None, query_start=None, query_limit=None
):
"""Get attributes of all HQE of given id.
Args:
job_id(int): The job id which HQEs associated with.
host_id(int): The AFE DUT id which HQEs associated with.
query_start(int): Parameter specifying index on where to start when
returning paged results.
query_limit(int): Upper limit on number of results to return.
Returns:
A list of dict. The dict is the information about the HQE.
"""
params = {}
if query_start is not None and query_limit is not None:
params["query_start"] = query_start
params["query_limit"] = query_limit
if job_id:
params["job__id"] = job_id
if host_id:
params["host__id"] = host_id
response = self.send_rpc_command("get_host_queue_entries", params)
result = self.decode_response(response, "result")
error = self.decode_response(response, "error")
return result, error
def run_suite(
self,
board,
build,
builds,
suite,
suite_timeout_mins,
pool=None,
model=None,
suite_args=None,
test_args=None,
):
"""Request AFE run a given suite.
Args:
board (string): Software build to run, e.g. octopus
build (string): Software version to run e.g. R73-11647.24.0
builds (list): Software versions to run e.g. R73-11647.24.0
suite (string): Test suite to run e.g. cts_P
suite_timeout_mins (integer, optional): Number of minutes until the
suite times out.
pool (string, optional): Pool label to target suite on.
model (string, optional): Specific model to run on e.g. phaser
suite_args (string, optional): Defaults to None. delimited
key=val pairs passed to suite control file.
test_args (string, optional): Defaults to None. delimited
key=val pairs passed to test control file.
Returns:
Tuple (int, string): The first element will be an int representing
ID of the created suite job. The second value is an error message.
Exactly one of the values will ever not be None. A None ID implies
failure of the AFE call-through.
"""
params = {
"board": board,
"builds": builds,
"name": suite,
"pool": pool,
"run_prod_code": False,
"test_source_build": build,
"wait_for_results": True,
"suite_args": suite_args,
"test_args": test_args,
"job_retry": True,
"max_retries": sys.maxsize,
"model": model,
"timeout_mins": suite_timeout_mins,
"max_runtime_mins": suite_timeout_mins,
}
response = self.send_rpc_command("create_suite_job", params)
result = self.decode_response(response, "result")
error = self.decode_response(response, "error")
return result, error
def enroll_duts(self, ip_addresses):
_LOGGER.debug("Enroll hosts %s", ip_addresses)
p = multiprocessing.Pool(20)
p.map(enroll_dut, ip_addresses)
p.close()
p.join()
def unenroll_duts(self, ip_addresses):
hosts = self.get_connected_devices()
for host in hosts:
hostname = host.get("hostname")
params = {"id": host.get("id")}
_LOGGER.error(params)
if hostname in ip_addresses:
response = self.send_rpc_command("delete_host", params)
_LOGGER.error(response)
def abort_host_queue_entries(self, job_ids):
response = self.send_rpc_command(
"abort_host_queue_entries", {"job__id__in": job_ids}
)
return self.decode_response(response, "result")
def _package_edit_action_results(self, response):
"""
Add/remove attribute label actions are currently implemented by
calling an AFE endpoint that will return (True, <message>)
on success cases, but not the same format on failure cases. This
logic packages failure cases in the same format.
TO-DO: This logic should disappear when the underlying AFE calls
are no longer relied upon.
"""
if self.decode_response(response, "result"):
return self.decode_response(response, "result")
else:
return (False, self.decode_response(response, "error")["message"])
def add_label_to_host(self, dut_hostname, label):
"""Add label to the specified DUT.
Args:
dut_hostname(string): hostname (IP) of targeted DUT.
label (string): Label to be added.
Returns:
A tuple with first value indicating whether action succeeded and
the second value a message summarizing the result (+ reason for
failure, if action failed).
"""
response = self.send_rpc_command(
"add_moblab_label",
{"ipaddress": dut_hostname, "label_name": label},
)
return self._package_edit_action_results(response)
def remove_label_from_host(self, dut_hostname, label):
"""
Removes label from the specified DUT.
Args:
dut_hostname(string): hostname (IP) of targeted DUT.
label (string): Label to be removed.
Returns:
A tuple with first value indicating whether action succeeded and
the second value a message summarizing the result (+ reason for
failure, if action failed).
"""
response = self.send_rpc_command(
"remove_moblab_label",
{"ipaddress": dut_hostname, "label_name": label},
)
return self._package_edit_action_results(response)
def add_attribute_to_host(self, dut_hostname, key, value):
"""
Adds attribute key:value to the specified DUT.
Args:
dut_hostname(string): hostname (IP) of targeted DUT.
key (string): Key of attribute to be added.
value (string): Value of attribute to be added.
Returns:
A tuple with first value indicating whether action succeeded and
the second value a message summarizing the result (+ reason for
failure, if action failed).
"""
response = self.send_rpc_command(
"set_host_attrib",
{"ipaddress": dut_hostname, "attribute": key, "value": value},
)
return self._package_edit_action_results(response)
def remove_attribute_from_host(self, dut_hostname, key):
"""
Removes attribute ( referenced by key ) from the specified DUT.
Args:
dut_hostname(string): hostname (IP) of targeted DUT.
key (string): Key of attribute to be removed.
Returns:
A tuple with first value indicating whether action succeeded and
the second value a message summarizing the result (+ reason for
failure, if action failed).
"""
response = self.send_rpc_command(
"delete_host_attrib",
{
"ipaddress": dut_hostname,
"attribute": key,
},
)
return self._package_edit_action_results(response)
def reverify_host(self, hostnames):
"""
Sets off reverify job on DUT of given ID.
Args:
hostnames([string]): hostnames of DUTs.
"""
response = self.send_rpc_command(
"reverify_hosts", {"hostname__in": hostnames}
)
return self.decode_response(response, "result")
def repair_host(self, host_id):
"""Set off repair job on DUT of given ID.
Args:
host_id(int): ID of DUT in AFE backend.
"""
response = self.send_rpc_command("repair_hosts", {"id": host_id})
return self.decode_response(response, "result")
def get_cloud_storage_info(self):
"""Get cloud configuration info from AFE.
Returns:
A dict of cloud configuration info ( specifically,
'gs_access_key_id', 'gs_secret_access_key',
'image_storage_server' )
"""
response = self.send_rpc_command("get_cloud_storage_info", {})
return self.decode_response(response, "result")
def get_dut_wifi_info(self):
"""
Gets DUT wifi info from AFE.
Returns:
A dict of DUT wifi info ( specifically, 'dut_wifi_name,
'dut_wifi_password' )
"""
response = self.send_rpc_command("get_dut_wifi_info", {})
return self.decode_response(response, "result")
def _format_configuration_dict(
self,
is_cloud_enabled=True,
gs_access_key_id="",
gs_secret_access_key="",
image_storage_server="",
dut_wifi_name=None,
dut_wifi_password="",
):
"""
Helper method for returning params shared between calls to set
config and validate cloud config. Will assume that submissions
are only for either DUT WIFI *or* cloud configurations.
Args:
is_cloud_enabled (boolean): True iff user has checked 'integrate
with cloud' on UI
gs_access_key_id (string): boto ID string
gs_secret_access_key (string): boto secret string
image_storage_server (string): GCS bucket URL
dut_wifi_name (string): Wifi name for DUTs
dut_wifi_password (string): Wifi password for DUTs
Returns:
A dictionary of cloud configuration parameters.
"""
configuration_dict = {}
if dut_wifi_name:
configuration_dict["wifi_info"] = {
"wifi_dut_ap_name": dut_wifi_name,
"wifi_dut_ap_pass": dut_wifi_password,
}
else:
configuration_dict["cloud_storage_info"] = {
"is_cloud_enabled": is_cloud_enabled,
"gs_access_key_id": gs_access_key_id,
"gs_secret_access_key": gs_secret_access_key,
"image_storage_server": image_storage_server,
"results_storage_server": "",
# AFE backend services configuration of other values in
# addition to cloud configs, but Moblab only needs to change
# these values, so this value is always False.
"use_existing_boto_file": False,
}
return configuration_dict
def set_configuration_info(
self,
is_cloud_enabled=True,
gs_access_key_id="",
gs_secret_access_key="",
image_storage_server="",
dut_wifi_name=None,
dut_wifi_password="",
):
"""
Applies given configuration values to AFE.
Args:
is_cloud_enabled (boolean): True iff user has checked 'integrate
with cloud' on UI
gs_access_key_id (string): boto ID string
gs_secret_access_key (string): boto secret string
image_storage_server (string): GCS bucket URL
dut_wifi_name (string): Wifi name for DUTs
dut_wifi_password (string): Wifi password for DUTs
Returns:
A boolean, true IFF application of new cloud configurations
succeeded.
"""
response = self.send_rpc_command(
"submit_wizard_config_info",
self._format_configuration_dict(
is_cloud_enabled,
gs_access_key_id,
gs_secret_access_key,
image_storage_server,
dut_wifi_name,
dut_wifi_password,
),
)
response = self.decode_response(response, "result")
if "status_ok" in response and response["status_ok"]:
return True
else:
return False
def validate_cloud_storage_info(
self, gs_access_key_id, gs_secret_access_key, image_storage_server
):
"""Call through to AFE to check validity of boto and bucket
configuration, returns true iff configurations are valid, and if
false returns a reason that configuration is not valid.
Args:
gs_access_key_id (string): boto ID string
gs_secret_access_key (string): boto secret string
image_storage_server (string): GCS bucket URL
Returns:
A tuple (boolean, string).
"""
response = self.send_rpc_command(
"validate_cloud_storage_info",
self._format_configuration_dict(
gs_access_key_id=gs_access_key_id,
gs_secret_access_key=gs_secret_access_key,
image_storage_server=image_storage_server,
),
)
result = self.decode_response(response, "result")
return result["status_ok"], result.get("status_details", "")
def set_is_remote_agent_enabled(self, is_enabled=False):
"""Apply given remote agent configuration values to AFE.
Args:
is_enabled (bool): enable the remote agent or not
Returns:
A boolean, true IFF application of remote agent configurations
succeeded.
"""
response = self.send_rpc_command(
"set_remote_agent_config", {"enabled": is_enabled}
)
response = self.decode_response(response, "result")
return (
True
if "status_ok" in response and response["status_ok"]
else False
)
def set_is_remote_command_enabled(self, is_enabled=False):
"""Apply given remote command configuration values to AFE.
Args:
is_enabled (bool): enable the remote command or not
Returns:
A boolean, true IFF application of remote command configurations
succeeded.
"""
response = self.send_rpc_command(
"set_remote_command_config", {"enabled": is_enabled}
)
response = self.decode_response(response, "result")
return (
True
if "status_ok" in response and response["status_ok"]
else False
)
def modify_hosts_lock_state(
self, new_locked_state, hosts_names, lock_reason
):
"""Lock/unlock given DUTs by host name
Args:
new_locked_state (bool): lock or unlock DUTs
hosts_names ([str]): list of host names (DUTs' IPs)
lock_reason: user facing text message
Returns:
A string. Errors message returned by AFE or None.
"""
connected_duts = self.get_connected_devices()
_LOGGER.info("Connected DUTs: %s", connected_duts)
hosts_names = [
dut["hostname"]
for dut in connected_duts
if dut["hostname"] in hosts_names
and dut["locked"] != new_locked_state
]
params = {
"host_filter_data": {"hostname__in": hosts_names},
"update_data": {
"locked": new_locked_state,
"lock_reason": lock_reason,
},
}
response = self.send_rpc_command("modify_hosts", params)
logging.debug(
"modify_hosts call to modified lock state "
"request: %s ,response: %s",
params,
response,
)
return self.decode_response(response, "error")