文件操作 - efs_info.py
返回文件管理
返回主菜单
删除本文件
文件: /usr/lib/python3.12/site-packages/ansible_collections/community/aws/plugins/modules/efs_info.py
编辑文件内容
# -*- coding: utf-8 -*- # Copyright: Ansible Project # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) DOCUMENTATION = r""" --- module: efs_info version_added: 1.0.0 short_description: Get information about Amazon EFS file systems description: - This module can be used to search Amazon EFS file systems. Note that the M(community.aws.efs_info) module no longer returns C(ansible_facts)! author: - "Ryan Sydnor (@ryansydnor)" options: name: description: - Creation Token of Amazon EFS file system. aliases: [ creation_token ] type: str id: description: - ID of Amazon EFS. type: str tags: description: - List of tags of Amazon EFS. Should be defined as dictionary. type: dict default: {} targets: description: - List of targets on which to filter the returned results. - Result must match all of the specified targets, each of which can be a security group ID, a subnet ID or an IP address. type: list elements: str default: [] extends_documentation_fragment: - amazon.aws.common.modules - amazon.aws.region.modules - amazon.aws.boto3 """ EXAMPLES = r""" - name: Find all existing efs community.aws.efs_info: register: result - name: Find efs using id community.aws.efs_info: id: fs-1234abcd register: result - name: Searching all EFS instances with tag Name = 'myTestNameTag', in subnet 'subnet-1a2b3c4d' and with security group 'sg-4d3c2b1a' community.aws.efs_info: tags: Name: myTestNameTag targets: - subnet-1a2b3c4d - sg-4d3c2b1a register: result - ansible.builtin.debug: msg: "{{ result['efs'] }}" """ RETURN = r""" creation_time: description: timestamp of creation date returned: always type: str sample: "2015-11-16 07:30:57-05:00" creation_token: description: EFS creation token returned: always type: str sample: console-88609e04-9a0e-4a2e-912c-feaa99509961 file_system_id: description: ID of the file system returned: always type: str sample: fs-xxxxxxxx life_cycle_state: description: state of the EFS file system returned: always type: str sample: creating, available, deleting, deleted mount_point: description: url of file system with leading dot from the time AWS EFS required to add network suffix to EFS address returned: always type: str sample: .fs-xxxxxxxx.efs.us-west-2.amazonaws.com:/ filesystem_address: description: url of file system returned: always type: str sample: fs-xxxxxxxx.efs.us-west-2.amazonaws.com:/ mount_targets: description: list of mount targets returned: always type: list sample: [ { "file_system_id": "fs-a7ad440e", "ip_address": "172.31.17.173", "life_cycle_state": "available", "mount_target_id": "fsmt-d8907871", "network_interface_id": "eni-6e387e26", "owner_id": "123456789012", "security_groups": [ "sg-a30b22c6" ], "subnet_id": "subnet-e265c895" }, ... ] name: description: name of the file system returned: always type: str sample: my-efs number_of_mount_targets: description: the number of targets mounted returned: always type: int sample: 3 owner_id: description: AWS account ID of EFS owner returned: always type: str sample: XXXXXXXXXXXX size_in_bytes: description: size of the file system in bytes as of a timestamp returned: always type: dict sample: { "timestamp": "2015-12-21 13:59:59-05:00", "value": 12288 } performance_mode: description: performance mode of the file system returned: always type: str sample: "generalPurpose" throughput_mode: description: mode of throughput for the file system returned: always type: str sample: "bursting" provisioned_throughput_in_mibps: description: throughput provisioned in Mibps returned: when throughput_mode is set to "provisioned" type: float sample: 15.0 tags: description: tags on the efs instance returned: always type: dict sample: { "name": "my-efs", "key": "Value" } """ from collections import defaultdict try: import botocore except ImportError: pass # caught by AnsibleAWSModule from ansible.module_utils._text import to_native from ansible.module_utils.common.dict_transformations import camel_dict_to_snake_dict from ansible_collections.amazon.aws.plugins.module_utils.retries import AWSRetry from ansible_collections.amazon.aws.plugins.module_utils.tagging import boto3_tag_list_to_ansible_dict from ansible_collections.community.aws.plugins.module_utils.modules import AnsibleCommunityAWSModule as AnsibleAWSModule class EFSConnection(object): STATE_CREATING = "creating" STATE_AVAILABLE = "available" STATE_DELETING = "deleting" STATE_DELETED = "deleted" def __init__(self, module): try: self.connection = module.client("efs") self.module = module except Exception as e: module.fail_json(msg=f"Failed to connect to AWS: {to_native(e)}") self.region = module.region @AWSRetry.exponential_backoff(catch_extra_error_codes=["ThrottlingException"]) def list_file_systems(self, **kwargs): """ Returns generator of file systems including all attributes of FS """ paginator = self.connection.get_paginator("describe_file_systems") return paginator.paginate(**kwargs).build_full_result()["FileSystems"] @AWSRetry.exponential_backoff(catch_extra_error_codes=["ThrottlingException"]) def get_tags(self, file_system_id): """ Returns tag list for selected instance of EFS """ paginator = self.connection.get_paginator("describe_tags") return boto3_tag_list_to_ansible_dict( paginator.paginate(FileSystemId=file_system_id).build_full_result()["Tags"] ) @AWSRetry.exponential_backoff(catch_extra_error_codes=["ThrottlingException"]) def get_mount_targets(self, file_system_id): """ Returns mount targets for selected instance of EFS """ paginator = self.connection.get_paginator("describe_mount_targets") return paginator.paginate(FileSystemId=file_system_id).build_full_result()["MountTargets"] @AWSRetry.jittered_backoff(catch_extra_error_codes=["ThrottlingException"]) def get_security_groups(self, mount_target_id): """ Returns security groups for selected instance of EFS """ return self.connection.describe_mount_target_security_groups(MountTargetId=mount_target_id)["SecurityGroups"] def get_mount_targets_data(self, file_systems): for item in file_systems: if item["life_cycle_state"] == self.STATE_AVAILABLE: try: mount_targets = self.get_mount_targets(item["file_system_id"]) except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e: self.module.fail_json_aws(e, msg="Couldn't get EFS targets") for mt in mount_targets: item["mount_targets"].append(camel_dict_to_snake_dict(mt)) return file_systems def get_security_groups_data(self, file_systems): for item in file_systems: if item["life_cycle_state"] == self.STATE_AVAILABLE: for target in item["mount_targets"]: if target["life_cycle_state"] == self.STATE_AVAILABLE: try: target["security_groups"] = self.get_security_groups(target["mount_target_id"]) except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e: self.module.fail_json_aws(e, msg="Couldn't get EFS security groups") else: target["security_groups"] = [] else: item["tags"] = {} item["mount_targets"] = [] return file_systems def get_file_systems(self, file_system_id=None, creation_token=None): kwargs = dict() if file_system_id: kwargs["FileSystemId"] = file_system_id if creation_token: kwargs["CreationToken"] = creation_token try: file_systems = self.list_file_systems(**kwargs) except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e: self.module.fail_json_aws(e, msg="Couldn't get EFS file systems") results = list() for item in file_systems: item["CreationTime"] = str(item["CreationTime"]) """ In the time when MountPoint was introduced there was a need to add a suffix of network path before one could use it AWS updated it and now there is no need to add a suffix. MountPoint is left for back-compatibility purpose And new FilesystemAddress variable is introduced for direct use with other modules (e.g. mount) AWS documentation is available here: U(https://docs.aws.amazon.com/efs/latest/ug/gs-step-three-connect-to-ec2-instance.html) """ item["MountPoint"] = f".{item['FileSystemId']}.efs.{self.region}.amazonaws.com:/" item["FilesystemAddress"] = f"{item['FileSystemId']}.efs.{self.region}.amazonaws.com:/" if "Timestamp" in item["SizeInBytes"]: item["SizeInBytes"]["Timestamp"] = str(item["SizeInBytes"]["Timestamp"]) result = camel_dict_to_snake_dict(item) result["tags"] = {} result["mount_targets"] = [] # Set tags *after* doing camel to snake if result["life_cycle_state"] == self.STATE_AVAILABLE: try: result["tags"] = self.get_tags(result["file_system_id"]) except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e: self.module.fail_json_aws(e, msg="Couldn't get EFS tags") results.append(result) return results def prefix_to_attr(attr_id): """ Helper method to convert ID prefix to mount target attribute """ attr_by_prefix = { "fsmt-": "mount_target_id", "subnet-": "subnet_id", "eni-": "network_interface_id", "sg-": "security_groups", } return first_or_default( [attr_name for (prefix, attr_name) in attr_by_prefix.items() if str(attr_id).startswith(prefix)], "ip_address" ) def first_or_default(items, default=None): """ Helper method to fetch first element of list (if exists) """ for item in items: return item return default def has_tags(available, required): """ Helper method to determine if tag requested already exists """ for key, value in required.items(): if key not in available or value != available[key]: return False return True def has_targets(available, required): """ Helper method to determine if mount target requested already exists """ grouped = group_list_of_dict(available) for value, field in required: if field not in grouped or value not in grouped[field]: return False return True def group_list_of_dict(array): """ Helper method to group list of dict to dict with all possible values """ result = defaultdict(list) for item in array: for key, value in item.items(): result[key] += value if isinstance(value, list) else [value] return result def main(): """ Module action handler """ argument_spec = dict( id=dict(), name=dict(aliases=["creation_token"]), tags=dict(type="dict", default={}), targets=dict(type="list", default=[], elements="str"), ) module = AnsibleAWSModule(argument_spec=argument_spec, supports_check_mode=True) connection = EFSConnection(module) name = module.params.get("name") fs_id = module.params.get("id") tags = module.params.get("tags") targets = module.params.get("targets") file_systems_info = connection.get_file_systems(fs_id, name) if tags: file_systems_info = [item for item in file_systems_info if has_tags(item["tags"], tags)] file_systems_info = connection.get_mount_targets_data(file_systems_info) file_systems_info = connection.get_security_groups_data(file_systems_info) if targets: targets = [(item, prefix_to_attr(item)) for item in targets] file_systems_info = [item for item in file_systems_info if has_targets(item["mount_targets"], targets)] module.exit_json(changed=False, efs=file_systems_info) if __name__ == "__main__": main()
修改文件时间
将文件时间修改为当前时间的前一年
删除文件