# -*- coding: utf-8 -*- # # Copyright 2020 Google LLC. All Rights Reserved. # # 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. """This class will store in-memory instance of ops agent policy.""" from __future__ import absolute_import from __future__ import division from __future__ import unicode_literals import enum import json import sys from googlecloudsdk.core.resource import resource_property _StrEnum = ( (enum.StrEnum,) if sys.version_info[:2] >= (3, 11) else (str, enum.Enum) ) class OpsAgentPolicy(object): """An Ops Agent policy encapsulates the underlying OS Config Guest Policy.""" class AgentRule(object): """An Ops agent rule contains agent type, version, enable_autoupgrade.""" class Type(*_StrEnum): LOGGING = 'logging' METRICS = 'metrics' OPS_AGENT = 'ops-agent' class PackageState(*_StrEnum): INSTALLED = 'installed' REMOVED = 'removed' class Version(*_StrEnum): LATEST_OF_ALL = 'latest' CURRENT_MAJOR = 'current-major' def __init__(self, agent_type, enable_autoupgrade, version=Version.CURRENT_MAJOR, package_state=PackageState.INSTALLED): """Initialize AgentRule instance. Args: agent_type: Type, agent type to be installed. enable_autoupgrade: bool, enable autoupgrade for the package or not. version: str, agent version, e.g. 'latest', '5.5.2', '5.*.*'. package_state: Optional PackageState, desiredState for the package. """ self.type = agent_type self.enable_autoupgrade = enable_autoupgrade self.version = version self.package_state = package_state def __eq__(self, other): return self.__dict__ == other.__dict__ def ToJson(self): """Generate JSON with camel-cased key.""" key_camel_cased_dict = { resource_property.ConvertToCamelCase(key): value for key, value in self.__dict__.items() } return json.dumps(key_camel_cased_dict, default=str, sort_keys=True) class Assignment(object): """The group or groups of VM instances that the policy applies to.""" class OsType(object): """The criteria for selecting VM Instances by OS type.""" class OsShortName(*_StrEnum): CENTOS = 'centos' DEBIAN = 'debian' WINDOWS = 'windows' RHEL = 'rhel' ROCKY = 'rocky' SLES = 'sles' SLES_SAP = 'sles-sap' UBUNTU = 'ubuntu' def __init__(self, short_name, version): """Initialize OsType instance. Args: short_name: str, OS distro name, e.g. 'centos', 'debian'. version: str, OS version, e.g. '19.10', '7', '7.8'. """ self.short_name = short_name self.version = version def __eq__(self, other): return self.__dict__ == other.__dict__ def __init__(self, group_labels, zones, instances, os_types): """Initialize Assignment Instance. Args: group_labels: list of dict, VM group label matchers, or None. zones: list, VM zone matchers, or None. instances: list, instance name matchers, or None. os_types: OsType, VM OS type matchers, or None. """ self.group_labels = group_labels or [] self.zones = zones or [] self.instances = instances or [] self.os_types = os_types or [] def __eq__(self, other): return self.__dict__ == other.__dict__ def __init__(self, assignment, agent_rules, description, etag, name, update_time, create_time): """Initialize an ops agent policy instance. Args: assignment: Assignment, selection criteria for applying policy to VMs. agent_rules: list of AgentRule, the agent rules to be applied to VMs. description: str, user specified description of the policy. etag: str, unique tag for policy, generated by the API, or None. name: str, user specified name of the policy, or None. update_time: str, update time in RFC3339 format, or None. create_time: str, create time in RFC3339 format, or None. """ self.assignment = assignment self.agent_rules = agent_rules self.description = description self.etag = etag self.id = name self.update_time = update_time self.create_time = create_time def __eq__(self, other): return self.__dict__ == other.__dict__ def __repr__(self): """JSON format string representation for testing.""" return json.dumps(self, default=lambda o: o.__dict__, indent=2, separators=(',', ': '), sort_keys=True) def CreateOsTypes(os_types): """Create Os Types in Ops Agent Policy. Args: os_types: dict, VM OS type matchers, or None. Returns: A list of OpsAgentPolicy.Assignment.OsType objects. """ OsType = OpsAgentPolicy.Assignment.OsType # pylint: disable=invalid-name return [ OsType(OsType.OsShortName(os_type['short-name']), os_type['version']) for os_type in os_types or [] ] def CreateAgentRules(agent_rules): """Create agent rules in ops agent policy. Args: agent_rules: list of dict, fields describing agent rules from the command line. Returns: An OpsAgentPolicy.AgentRules object. """ ops_agents = [] for agent_rule in agent_rules or []: ops_agents.append( OpsAgentPolicy.AgentRule( OpsAgentPolicy.AgentRule.Type(agent_rule['type']), agent_rule['enable-autoupgrade'], agent_rule.get('version', OpsAgentPolicy.AgentRule.Version.CURRENT_MAJOR), OpsAgentPolicy.AgentRule.PackageState(agent_rule.get( 'package-state', OpsAgentPolicy.AgentRule.PackageState.INSTALLED)))) return ops_agents def CreateOpsAgentPolicy(description, agent_rules, group_labels, os_types, zones, instances): """Create Ops Agent Policy. Args: description: str, ops agent policy description. agent_rules: list of dict, fields describing agent rules from the command line. group_labels: list of dict, VM group label matchers. os_types: dict, VM OS type matchers. zones: list, VM zone matchers. instances: list, instance name matchers. Returns: ops agent policy. """ return OpsAgentPolicy( assignment=OpsAgentPolicy.Assignment( group_labels=group_labels, zones=zones, instances=instances, os_types=CreateOsTypes(os_types)), agent_rules=CreateAgentRules(agent_rules), description=description, etag=None, name=None, update_time=None, create_time=None) def UpdateOpsAgentsPolicy(ops_agents_policy, description, etag, agent_rules, os_types, group_labels, zones, instances): """Merge existing ops agent policy with user updates. Unless explicitly mentioned, a None value means "leave unchanged". Args: ops_agents_policy: OpsAgentPolicy, ops agent policy. description: str, ops agent policy description, or None. etag: str, unique tag for policy to prevent race conditions, or None. agent_rules: list of dict, fields describing agent rules from the command line, or None. An empty list means the same as None. os_types: dict, VM OS type matchers, or None. An empty dict means the same as None. group_labels: list of dict, VM group label matchers, or None. zones: list of zones, VM zone matchers, or None. instances: list of instances, instance name matchers, or None. Returns: Updated ops agents policy. """ updated_description = ( ops_agents_policy.description if description is None else description) # TODO(b/164141164): Decide what should happen when etag=''. # Unless supplied, keep the etag from the last policy read from the server. # If the etag is stale, the RPC will error out to prevent race conditions. updated_etag = ops_agents_policy.etag if etag is None else etag assignment = ops_agents_policy.assignment updated_assignment = OpsAgentPolicy.Assignment( group_labels=( assignment.group_labels if group_labels is None else group_labels), zones=assignment.zones if zones is None else zones, instances=assignment.instances if instances is None else instances, os_types=CreateOsTypes(os_types) or assignment.os_types) updated_agent_rules = ( CreateAgentRules(agent_rules) or ops_agents_policy.agent_rules) return OpsAgentPolicy( assignment=updated_assignment, agent_rules=updated_agent_rules, description=updated_description, etag=updated_etag, name=ops_agents_policy.id, update_time=None, create_time=ops_agents_policy.create_time)