#!/usr/bin/env python # # Copyright 2016 - The Android Open Source Project # # 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. """Config manager. Three protobuf messages are defined in driver/internal/config/proto/internal_config.proto driver/internal/config/proto/user_config.proto Internal config file User config file | | v v InternalConfig UserConfig (proto message) (proto message) | | | | |-> AcloudConfig <-| At runtime, AcloudConfigManager performs the following steps. - Load driver config file into a InternalConfig message instance. - Load user config file into a UserConfig message instance. - Create AcloudConfig using InternalConfig and UserConfig. TODO: 1. Add support for override configs with command line args. 2. Scan all configs to find the right config for given branch and build_id. Raise an error if the given build_id is smaller than min_build_id only applies to release build id. Raise an error if the branch is not supported. """ import logging import os from google.protobuf import text_format # pylint: disable=no-name-in-module,import-error from acloud import errors from acloud.internal import constants from acloud.internal.proto import internal_config_pb2 from acloud.internal.proto import user_config_pb2 from acloud.create import create_args _CONFIG_DATA_PATH = os.path.join( os.path.dirname(os.path.abspath(__file__)), "data") _DEFAULT_CONFIG_FILE = "acloud.config" logger = logging.getLogger(__name__) def GetDefaultConfigFile(): """Return path to default config file.""" config_path = os.path.join(os.path.expanduser("~"), ".config", "acloud") # Create the default config dir if it doesn't exist. if not os.path.exists(config_path): os.makedirs(config_path) return os.path.join(config_path, _DEFAULT_CONFIG_FILE) def GetAcloudConfig(args): """Helper function to initialize Config object. Args: args: Namespace object from argparse.parse_args. Return: An instance of AcloudConfig. """ config_mgr = AcloudConfigManager(args.config_file) cfg = config_mgr.Load() cfg.OverrideWithArgs(args) return cfg class AcloudConfig(object): """A class that holds all configurations for acloud.""" REQUIRED_FIELD = [ "machine_type", "network", "min_machine_size", "disk_image_name", "disk_image_mime_type" ] # pylint: disable=too-many-statements def __init__(self, usr_cfg, internal_cfg): """Initialize. Args: usr_cfg: A protobuf object that holds the user configurations. internal_cfg: A protobuf object that holds internal configurations. """ self.service_account_name = usr_cfg.service_account_name # pylint: disable=invalid-name self.service_account_private_key_path = ( usr_cfg.service_account_private_key_path) self.service_account_json_private_key_path = ( usr_cfg.service_account_json_private_key_path) self.creds_cache_file = internal_cfg.creds_cache_file self.user_agent = internal_cfg.user_agent self.client_id = usr_cfg.client_id self.client_secret = usr_cfg.client_secret self.project = usr_cfg.project self.zone = usr_cfg.zone self.machine_type = (usr_cfg.machine_type or internal_cfg.default_usr_cfg.machine_type) self.network = usr_cfg.network or internal_cfg.default_usr_cfg.network self.ssh_private_key_path = usr_cfg.ssh_private_key_path self.ssh_public_key_path = usr_cfg.ssh_public_key_path self.storage_bucket_name = usr_cfg.storage_bucket_name self.metadata_variable = { key: val for key, val in internal_cfg.default_usr_cfg.metadata_variable.iteritems() } self.metadata_variable.update(usr_cfg.metadata_variable) self.device_resolution_map = { device: resolution for device, resolution in internal_cfg.device_resolution_map.iteritems() } self.device_default_orientation_map = { device: orientation for device, orientation in internal_cfg.device_default_orientation_map.iteritems() } self.no_project_access_msg_map = { project: msg for project, msg in internal_cfg.no_project_access_msg_map.iteritems() } self.min_machine_size = internal_cfg.min_machine_size self.disk_image_name = internal_cfg.disk_image_name self.disk_image_mime_type = internal_cfg.disk_image_mime_type self.disk_image_extension = internal_cfg.disk_image_extension self.disk_raw_image_name = internal_cfg.disk_raw_image_name self.disk_raw_image_extension = internal_cfg.disk_raw_image_extension self.valid_branch_and_min_build_id = { branch: min_build_id for branch, min_build_id in internal_cfg.valid_branch_and_min_build_id.iteritems() } self.precreated_data_image_map = { size_gb: image_name for size_gb, image_name in internal_cfg.precreated_data_image.iteritems() } self.extra_data_disk_size_gb = ( usr_cfg.extra_data_disk_size_gb or internal_cfg.default_usr_cfg.extra_data_disk_size_gb) if self.extra_data_disk_size_gb > 0: if "cfg_sta_persistent_data_device" not in usr_cfg.metadata_variable: # If user did not set it explicity, use default. self.metadata_variable["cfg_sta_persistent_data_device"] = ( internal_cfg.default_extra_data_disk_device) if "cfg_sta_ephemeral_data_size_mb" in usr_cfg.metadata_variable: raise errors.ConfigError( "The following settings can't be set at the same time: " "extra_data_disk_size_gb and" "metadata variable cfg_sta_ephemeral_data_size_mb.") if "cfg_sta_ephemeral_data_size_mb" in self.metadata_variable: del self.metadata_variable["cfg_sta_ephemeral_data_size_mb"] # Additional scopes to be passed to the created instance self.extra_scopes = usr_cfg.extra_scopes # Fields that can be overriden by args self.orientation = usr_cfg.orientation self.resolution = usr_cfg.resolution self.stable_host_image_name = ( usr_cfg.stable_host_image_name or internal_cfg.default_usr_cfg.stable_host_image_name) self.stable_host_image_project = ( usr_cfg.stable_host_image_project or internal_cfg.default_usr_cfg.stable_host_image_project) self.kernel_build_target = internal_cfg.kernel_build_target self.emulator_build_target = internal_cfg.emulator_build_target self.stable_goldfish_host_image_name = ( usr_cfg.stable_goldfish_host_image_name or internal_cfg.default_usr_cfg.stable_goldfish_host_image_name) self.stable_goldfish_host_image_project = ( usr_cfg.stable_goldfish_host_image_project or internal_cfg.default_usr_cfg.stable_goldfish_host_image_project) self.stable_cheeps_host_image_name = ( usr_cfg.stable_cheeps_host_image_name or internal_cfg.default_usr_cfg.stable_cheeps_host_image_name) self.stable_cheeps_host_image_project = ( usr_cfg.stable_cheeps_host_image_project or internal_cfg.default_usr_cfg.stable_cheeps_host_image_project) self.common_hw_property_map = internal_cfg.common_hw_property_map self.hw_property = usr_cfg.hw_property self.launch_args = usr_cfg.launch_args self.instance_name_pattern = ( usr_cfg.instance_name_pattern or internal_cfg.default_usr_cfg.instance_name_pattern) # Verify validity of configurations. self.Verify() def OverrideWithArgs(self, parsed_args): """Override configuration values with args passed in from cmd line. Args: parsed_args: Args parsed from command line. """ if parsed_args.which == create_args.CMD_CREATE and parsed_args.spec: if not self.resolution: self.resolution = self.device_resolution_map.get( parsed_args.spec, "") if not self.orientation: self.orientation = self.device_default_orientation_map.get( parsed_args.spec, "") if parsed_args.email: self.service_account_name = parsed_args.email if parsed_args.service_account_json_private_key_path: self.service_account_json_private_key_path = ( parsed_args.service_account_json_private_key_path) if parsed_args.which == "create_gf" and parsed_args.base_image: self.stable_goldfish_host_image_name = parsed_args.base_image if parsed_args.which == create_args.CMD_CREATE and not self.hw_property: flavor = parsed_args.flavor or constants.FLAVOR_PHONE self.hw_property = self.common_hw_property_map.get(flavor, "") if parsed_args.which in [create_args.CMD_CREATE, "create_cf"]: if parsed_args.network: self.network = parsed_args.network def OverrideHwPropertyWithFlavor(self, flavor): """Override hw configuration values with flavor name. HwProperty will be overrided according to the change of flavor. If flavor is None, set hw configuration with phone(default flavor). Args: flavor: string of flavor name. """ self.hw_property = self.common_hw_property_map.get( flavor, constants.FLAVOR_PHONE) def Verify(self): """Verify configuration fields.""" missing = [f for f in self.REQUIRED_FIELD if not getattr(self, f)] if missing: raise errors.ConfigError( "Missing required configuration fields: %s" % missing) if (self.extra_data_disk_size_gb and self.extra_data_disk_size_gb not in self.precreated_data_image_map): raise errors.ConfigError( "Supported extra_data_disk_size_gb options(gb): %s, " "invalid value: %d" % (self.precreated_data_image_map.keys(), self.extra_data_disk_size_gb)) class AcloudConfigManager(object): """A class that loads configurations.""" _DEFAULT_INTERNAL_CONFIG_PATH = os.path.join(_CONFIG_DATA_PATH, "default.config") def __init__(self, user_config_path, internal_config_path=_DEFAULT_INTERNAL_CONFIG_PATH): """Initialize with user specified paths to configs. Args: user_config_path: path to the user config. internal_config_path: path to the internal conifg. """ self.user_config_path = user_config_path self._internal_config_path = internal_config_path def Load(self): """Load the configurations. Load user config with some special design. 1. User specified user config: a.User config exist: Load config. b.User config didn't exist: Raise exception. 2. User didn't specify user config, use default config: a.Default config exist: Load config. b.Default config didn't exist: provide empty usr_cfg. """ internal_cfg = None usr_cfg = None try: with open(self._internal_config_path) as config_file: internal_cfg = self.LoadConfigFromProtocolBuffer( config_file, internal_config_pb2.InternalConfig) except OSError as e: raise errors.ConfigError("Could not load config files: %s" % str(e)) # Load user config file if self.user_config_path: if os.path.exists(self.user_config_path): with open(self.user_config_path, "r") as config_file: usr_cfg = self.LoadConfigFromProtocolBuffer( config_file, user_config_pb2.UserConfig) else: raise errors.ConfigError("The file doesn't exist: %s" % (self.user_config_path)) else: self.user_config_path = GetDefaultConfigFile() if os.path.exists(self.user_config_path): with open(self.user_config_path, "r") as config_file: usr_cfg = self.LoadConfigFromProtocolBuffer( config_file, user_config_pb2.UserConfig) else: usr_cfg = user_config_pb2.UserConfig() return AcloudConfig(usr_cfg, internal_cfg) @staticmethod def LoadConfigFromProtocolBuffer(config_file, message_type): """Load config from a text-based protocol buffer file. Args: config_file: A python File object. message_type: A proto message class. Returns: An instance of type "message_type" populated with data from the file. """ try: config = message_type() text_format.Merge(config_file.read(), config) return config except text_format.ParseError as e: raise errors.ConfigError("Could not parse config: %s" % str(e))