| # Copyright 2019-2020 Fairphone B.V. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| |
| """Utilities for accessing a LAVA instance from Python.""" |
| |
| from enum import Enum |
| from pathlib import Path |
| import subprocess |
| from typing import Any, Optional, Set |
| |
| import yaml |
| |
| |
| class LavaError(Exception): |
| """Exception raised for failed calls to the LAVA XMLRPC API.""" |
| |
| def __init__(self, called_process_error: subprocess.CalledProcessError): |
| """Initialize a LavaError instance.""" |
| super().__init__() |
| self.called_process_error = called_process_error |
| |
| def __str__(self) -> str: |
| """Create a string representation of this object.""" |
| return ( |
| "Command '{}' failed with exit status {}.\n" |
| "stdout: {}\n" |
| "stderr: {}".format( |
| self.called_process_error.cmd, |
| self.called_process_error.returncode, |
| self.called_process_error.stdout, |
| self.called_process_error.stderr, |
| ) |
| ) |
| |
| |
| class Lavacli: |
| """Wrapper class for lavacli. |
| |
| This class allows to call lavacli from python. It assumes that connection |
| and authorization to a LAVA instance is set up in lavacli before using this |
| class. |
| """ |
| |
| def __init__(self, identity: str): |
| """Initialize a Lavacli instance. |
| |
| Arguments: |
| identity: |
| lavacli identity for connecting and authorizing at a LAVA |
| instance through lavacli. This identity needs to be configured |
| in lavacli before making calls through this class. |
| |
| """ |
| self._identity = identity |
| |
| def cmd(self, *argv: str) -> str: |
| """Execute a command through lavacli. |
| |
| Arguments: |
| argv: |
| List if command line arguments for lavacli. |
| |
| Returns: |
| Standard output string of the executed command. |
| |
| Throws: |
| LavaError: |
| If lavacli is not available or quits with non-zero exit code. |
| |
| """ |
| try: |
| # subprocess.check_output is not typed, so ignore type checks here. |
| return subprocess.check_output( # type: ignore |
| ["lavacli", "-i", self._identity, *argv], encoding="utf-8" |
| ) |
| except subprocess.CalledProcessError as e: |
| # Re-raise the exception in the own type to make semantics clearer. |
| raise LavaError(e) |
| |
| def cmd_yaml(self, *argv: str) -> Any: |
| """Call `cmd` and parse output as YAML document. |
| |
| This function assumes that the arguments given in argv instruct lavacli |
| to print a YAML document to the standard output (usually achieved with |
| `--yaml` parameter in supporting subcommands). The standard output |
| string must contain exactly one YAML document with only standard YAML |
| tags. |
| |
| Refer to `cmd` for arguments and exception specification. |
| |
| Returns: |
| Python object constructed from the standard output of the executed |
| command. |
| |
| """ |
| return yaml.safe_load(self.cmd(*argv)) |
| |
| def tags_cmd(self, *argv: str) -> str: |
| """Call the 'tag' subcommand in lavacli.""" |
| return self.cmd("tags", *argv) |
| |
| def tags_cmd_yaml(self, *argv: str) -> Any: |
| """Call `cmd_yaml` for the 'tag' subcommand in lavacli.""" |
| return self.cmd_yaml("tags", *argv) |
| |
| def device_types_cmd(self, *argv: str) -> str: |
| """Call the 'device-types' subcommand in lavacli.""" |
| return self.cmd("device-types", *argv) |
| |
| def device_types_cmd_yaml(self, *argv: str) -> Any: |
| """Call `cmd_yaml` for the 'device-types' subcommand in lavacli.""" |
| return self.cmd_yaml("device-types", *argv) |
| |
| def devices_cmd(self, *argv: str) -> str: |
| """Call the 'devices' subcommand in lavacli.""" |
| return self.cmd("devices", *argv) |
| |
| def devices_cmd_yaml(self, *argv: str) -> Any: |
| """Call `cmd_yaml` for the 'devices' subcommand in lavacli.""" |
| return self.cmd_yaml("devices", *argv) |
| |
| |
| class LavaDeviceHealth(Enum): |
| """Enumeration of possible LAVA device health states. |
| |
| This enumeration also provides a translation between different string |
| representations in lavacli: When setting the device health state, the |
| uppercase notation is required. When querying the state, it is notated with |
| first letter in uppercase only. |
| |
| """ |
| |
| GOOD = "Good" |
| UNKNOWN = "Unknown" |
| LOOPING = "Looping" |
| BAD = "Bad" |
| MAINTENANCE = "Maintenance" |
| RETIRED = "Retired" |
| |
| |
| class LavaDevice: |
| """Wrapper of a device in LAVA.""" |
| |
| def __init__( |
| self, |
| hostname: str, |
| lavacli: Lavacli, |
| health: Optional[LavaDeviceHealth] = None, |
| ): |
| """Initialize a LavaDevice instance. |
| |
| Arguments: |
| hostname: Device hostname in LAVA. |
| lavacli: lavacli wrapper instance. |
| health: |
| Health state of the device. This must be None if the device is |
| not yet present in the LAVA instance and a valid |
| LavaDeviceHealth enum value otherwise. |
| |
| """ |
| self._hostname = hostname |
| self._lavacli = lavacli |
| self._health = health |
| |
| @property |
| def hostname(self) -> str: |
| """Return the hostname that identifies the device in a LAVA instance.""" |
| return self._hostname |
| |
| @property |
| def health(self) -> Optional[LavaDeviceHealth]: |
| """Return the cached value of the device health state.""" |
| return self._health |
| |
| @health.setter |
| def health(self, health: LavaDeviceHealth) -> None: |
| """Set device state if requested and cache value differ.""" |
| if self.health == health: |
| return |
| |
| self._lavacli.devices_cmd( |
| "update", "--health", health.name, self.hostname |
| ) |
| self._health = health |
| |
| def set_tags(self, new_tags: Set[str]) -> bool: |
| """Set the list of tags assigned to the device. |
| |
| Returns: True if the set of tags changed, False otherwise. |
| |
| """ |
| current_tags = set( |
| self._lavacli.devices_cmd_yaml( |
| "tags", "list", "--yaml", self.hostname |
| ) |
| ) |
| if current_tags == new_tags: |
| return False |
| |
| for tag in new_tags - current_tags: |
| self._lavacli.devices_cmd("tags", "add", self.hostname, tag) |
| for tag in current_tags - new_tags: |
| self._lavacli.devices_cmd("tags", "delete", self.hostname, tag) |
| return True |
| |
| def set_device_dict(self, device_dict_file: Path) -> bool: |
| """Set the device dictionary. |
| |
| Returns: True if the dictionary changed, False otherwise. |
| """ |
| current_dict: Optional[str] |
| try: |
| # Fetch the dict if it is set. Silently ignore if this fails due to |
| # a not yet existing configuration. |
| current_dict = self._lavacli.devices_cmd( |
| "dict", "get", self.hostname |
| ) |
| except LavaError as e: |
| expected_error = ( |
| '''Fault 404: "Device '{}' does not have a configuration"''' |
| "".format(self.hostname) |
| ) |
| if expected_error not in e.called_process_error.stdout: |
| raise e |
| current_dict = None |
| |
| new_dict = device_dict_file.read_text() |
| if new_dict == current_dict: |
| return False |
| |
| self._lavacli.devices_cmd( |
| "dict", "set", self.hostname, str(device_dict_file) |
| ) |
| return True |
| |
| def add_update_config( |
| self, |
| lava_worker: str, |
| device_type_name: str, |
| tags: Set[str], |
| device_dict_file: Path, |
| ) -> None: |
| """Ensure that the device is present and configured. |
| |
| This adds the device to the LAVA instance if it is not present yet and |
| ensures that all supplied parameters are up-to-date. If the device was |
| previously retired (removed from the LAVA instance), it will be |
| reactivated. |
| |
| This will trigger a health check on the device if anything needed to be |
| changed on the configuration. |
| """ |
| request_health_check = False |
| maintenance_params = [ |
| "--health", |
| LavaDeviceHealth.MAINTENANCE.name, |
| "--worker", |
| lava_worker, |
| self.hostname, |
| ] |
| if self.health is None: # New to the LAVA instance. |
| # Start in maintenance mode so that the health check doesn't run |
| # before the device is fully configured. |
| self._lavacli.devices_cmd( |
| "add", "--type", device_type_name, *maintenance_params |
| ) |
| request_health_check = True |
| else: |
| if self.health == LavaDeviceHealth.RETIRED: |
| # Reactivate the device if it was previously retired. |
| request_health_check = True |
| current_worker = None |
| else: |
| # A health check will be required if the worker changes. |
| current_worker = self._lavacli.devices_cmd_yaml( |
| "show", "--yaml", self.hostname |
| )["worker"] |
| |
| if current_worker != lava_worker: |
| self._lavacli.devices_cmd("update", *maintenance_params) |
| request_health_check = True |
| |
| if self.set_tags(new_tags=set(tags)): |
| request_health_check = True |
| if self.set_device_dict(device_dict_file=device_dict_file): |
| request_health_check = True |
| |
| # Run a health check if something changed. |
| if request_health_check: |
| self.health = LavaDeviceHealth.UNKNOWN |