#!/usr/bin/python # -*- 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: kinesis_stream version_added: 1.0.0 short_description: Manage a Kinesis Stream. description: - Create or Delete a Kinesis Stream. - Update the retention period of a Kinesis Stream. - Update Tags on a Kinesis Stream. - Enable/disable server side encryption on a Kinesis Stream. author: - Allen Sanabria (@linuxdynasty) options: name: description: - The name of the Kinesis Stream you are managing. required: true type: str shards: description: - The number of shards you want to have with this stream. - This is required when I(state=present) type: int retention_period: description: - The length of time (in hours) data records are accessible after they are added to the stream. - The default retention period is 24 hours and can not be less than 24 hours. - The maximum retention period is 168 hours. - The retention period can be modified during any point in time. type: int state: description: - Create or Delete the Kinesis Stream. default: present choices: [ 'present', 'absent' ] type: str wait: description: - Wait for operation to complete before returning. default: true type: bool wait_timeout: description: - How many seconds to wait for an operation to complete before timing out. default: 300 type: int tags: description: - "A dictionary of resource tags of the form: C({ tag1: value1, tag2: value2 })." aliases: [ "resource_tags" ] type: dict encryption_state: description: - Enable or Disable encryption on the Kinesis Stream. choices: [ 'enabled', 'disabled' ] type: str encryption_type: description: - The type of encryption. - Defaults to C(KMS) choices: ['KMS', 'NONE'] type: str key_id: description: - The GUID or alias for the KMS key. type: str extends_documentation_fragment: - amazon.aws.common.modules - amazon.aws.region.modules - amazon.aws.boto3 """ EXAMPLES = r""" # Note: These examples do not set authentication details, see the AWS Guide for details. # Basic creation example: - name: Set up Kinesis Stream with 10 shards and wait for the stream to become ACTIVE community.aws.kinesis_stream: name: test-stream shards: 10 wait: true wait_timeout: 600 register: test_stream # Basic creation example with tags: - name: Set up Kinesis Stream with 10 shards, tag the environment, and wait for the stream to become ACTIVE community.aws.kinesis_stream: name: test-stream shards: 10 tags: Env: development wait: true wait_timeout: 600 register: test_stream # Basic creation example with tags and increase the retention period from the default 24 hours to 48 hours: - name: Set up Kinesis Stream with 10 shards, tag the environment, increase the retention period and wait for the stream to become ACTIVE community.aws.kinesis_stream: name: test-stream retention_period: 48 shards: 10 tags: Env: development wait: true wait_timeout: 600 register: test_stream # Basic delete example: - name: Delete Kinesis Stream test-stream and wait for it to finish deleting. community.aws.kinesis_stream: name: test-stream state: absent wait: true wait_timeout: 600 register: test_stream # Basic enable encryption example: - name: Encrypt Kinesis Stream test-stream. community.aws.kinesis_stream: name: test-stream state: present shards: 1 encryption_state: enabled encryption_type: KMS key_id: alias/aws/kinesis wait: true wait_timeout: 600 register: test_stream # Basic disable encryption example: - name: Encrypt Kinesis Stream test-stream. community.aws.kinesis_stream: name: test-stream state: present shards: 1 encryption_state: disabled encryption_type: KMS key_id: alias/aws/kinesis wait: true wait_timeout: 600 register: test_stream """ RETURN = r""" stream_name: description: The name of the Kinesis Stream. returned: when state == present. type: str sample: "test-stream" stream_arn: description: The amazon resource identifier returned: when state == present. type: str sample: "arn:aws:kinesis:east-side:123456789:stream/test-stream" stream_status: description: The current state of the Kinesis Stream. returned: when state == present. type: str sample: "ACTIVE" retention_period_hours: description: Number of hours messages will be kept for a Kinesis Stream. returned: when state == present. type: int sample: 24 tags: description: Dictionary containing all the tags associated with the Kinesis stream. returned: when state == present. type: dict sample: { "Name": "Splunk", "Env": "development" } """ import time try: import botocore except ImportError: pass # Handled 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.tagging import boto3_tag_list_to_ansible_dict from ansible_collections.amazon.aws.plugins.module_utils.tagging import compare_aws_tags from ansible_collections.community.aws.plugins.module_utils.modules import AnsibleCommunityAWSModule as AnsibleAWSModule def get_tags(client, stream_name): """Retrieve the tags for a Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): Name of the Kinesis stream. Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >> get_tags(client, stream_name) Returns: Tuple (bool, str, dict) """ err_msg = "" success = False params = { "StreamName": stream_name, } results = dict() try: results = client.list_tags_for_stream(**params)["Tags"] success = True except botocore.exceptions.ClientError as e: err_msg = to_native(e) return success, err_msg, boto3_tag_list_to_ansible_dict(results) def find_stream(client, stream_name): """Retrieve a Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): Name of the Kinesis stream. Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' Returns: Tuple (bool, str, dict) """ err_msg = "" success = False params = { "StreamName": stream_name, } results = dict() has_more_shards = True shards = list() try: while has_more_shards: results = client.describe_stream(**params)["StreamDescription"] shards.extend(results.pop("Shards")) has_more_shards = results["HasMoreShards"] if has_more_shards: params["ExclusiveStartShardId"] = shards[-1]["ShardId"] results["Shards"] = shards num_closed_shards = len([s for s in shards if "EndingSequenceNumber" in s["SequenceNumberRange"]]) results["OpenShardsCount"] = len(shards) - num_closed_shards results["ClosedShardsCount"] = num_closed_shards results["ShardsCount"] = len(shards) success = True except botocore.exceptions.ClientError as e: err_msg = to_native(e) return success, err_msg, results def wait_for_status(client, stream_name, status, wait_timeout=300, check_mode=False): """Wait for the status to change for a Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client stream_name (str): The name of the kinesis stream. status (str): The status to wait for. examples. status=available, status=deleted Kwargs: wait_timeout (int): Number of seconds to wait, until this timeout is reached. check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> wait_for_status(client, stream_name, 'ACTIVE', 300) Returns: Tuple (bool, str, dict) """ polling_increment_secs = 5 wait_timeout = time.time() + wait_timeout status_achieved = False stream = dict() err_msg = "" while wait_timeout > time.time(): try: find_success, find_msg, stream = find_stream(client, stream_name) if check_mode: status_achieved = True break if status != "DELETING": if find_success and stream: if stream.get("StreamStatus") == status: status_achieved = True break else: if not find_success: status_achieved = True break except botocore.exceptions.ClientError as e: err_msg = to_native(e) time.sleep(polling_increment_secs) if not status_achieved: err_msg = "Wait time out reached, while waiting for results" else: err_msg = f"Status {status} achieved successfully" return status_achieved, err_msg, stream def tags_action(client, stream_name, tags, action="create", check_mode=False): """Create or delete multiple tags from a Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. resource_id (str): The Amazon resource id. tags (list): List of dictionaries. examples.. [{Name: "", Values: [""]}] Kwargs: action (str): The action to perform. valid actions == create and delete default=create check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('ec2') >>> resource_id = 'pcx-123345678' >>> tags = {'env': 'development'} >>> update_tags(client, resource_id, tags) [True, ''] Returns: List (bool, str) """ success = False err_msg = "" params = {"StreamName": stream_name} try: if not check_mode: if action == "create": params["Tags"] = tags client.add_tags_to_stream(**params) success = True elif action == "delete": params["TagKeys"] = tags client.remove_tags_from_stream(**params) success = True else: err_msg = f"Invalid action {action}" else: if action == "create": success = True elif action == "delete": success = True else: err_msg = f"Invalid action {action}" except botocore.exceptions.ClientError as e: err_msg = to_native(e) return success, err_msg def update_tags(client, stream_name, tags, check_mode=False): """Update tags for an amazon resource. Args: resource_id (str): The Amazon resource id. tags (dict): Dictionary of tags you want applied to the Kinesis stream. Kwargs: check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('ec2') >>> stream_name = 'test-stream' >>> tags = {'env': 'development'} >>> update_tags(client, stream_name, tags) [True, ''] Return: Tuple (bool, str) """ success = False changed = False err_msg = "" tag_success, tag_msg, current_tags = get_tags(client, stream_name) tags_to_set, tags_to_delete = compare_aws_tags( current_tags, tags, purge_tags=True, ) if tags_to_delete: delete_success, delete_msg = tags_action( client, stream_name, tags_to_delete, action="delete", check_mode=check_mode ) if not delete_success: return delete_success, changed, delete_msg if tags_to_set: create_success, create_msg = tags_action( client, stream_name, tags_to_set, action="create", check_mode=check_mode ) if create_success: changed = True return create_success, changed, create_msg return success, changed, err_msg def stream_action(client, stream_name, shard_count=1, action="create", timeout=300, check_mode=False): """Create or Delete an Amazon Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: shard_count (int): Number of shards this stream will use. action (str): The action to perform. valid actions == create and delete default=create check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> shard_count = 20 >>> stream_action(client, stream_name, shard_count, action='create') Returns: List (bool, str) """ success = False err_msg = "" params = {"StreamName": stream_name} try: if not check_mode: if action == "create": params["ShardCount"] = shard_count client.create_stream(**params) success = True elif action == "delete": client.delete_stream(**params) success = True else: err_msg = f"Invalid action {action}" else: if action == "create": success = True elif action == "delete": success = True else: err_msg = f"Invalid action {action}" except botocore.exceptions.ClientError as e: err_msg = to_native(e) return success, err_msg def stream_encryption_action( client, stream_name, action="start_encryption", encryption_type="", key_id="", timeout=300, check_mode=False ): """Create, Encrypt or Delete an Amazon Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: shard_count (int): Number of shards this stream will use. action (str): The action to perform. valid actions == create and delete default=create encryption_type (str): NONE or KMS key_id (str): The GUID or alias for the KMS key check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> shard_count = 20 >>> stream_action(client, stream_name, shard_count, action='create', encryption_type='KMS',key_id='alias/aws') Returns: List (bool, str) """ success = False err_msg = "" params = {"StreamName": stream_name} try: if not check_mode: if action == "start_encryption": params["EncryptionType"] = encryption_type params["KeyId"] = key_id client.start_stream_encryption(**params) success = True elif action == "stop_encryption": params["EncryptionType"] = encryption_type params["KeyId"] = key_id client.stop_stream_encryption(**params) success = True else: err_msg = f"Invalid encryption action {action}" else: if action == "start_encryption": success = True elif action == "stop_encryption": success = True else: err_msg = f"Invalid encryption action {action}" except botocore.exceptions.ClientError as e: err_msg = to_native(e) return success, err_msg def retention_action(client, stream_name, retention_period=24, action="increase", check_mode=False): """Increase or Decrease the retention of messages in the Kinesis stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: retention_period (int): This is how long messages will be kept before they are discarded. This can not be less than 24 hours. action (str): The action to perform. valid actions == create and delete default=create check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> retention_period = 48 >>> retention_action(client, stream_name, retention_period, action='increase') Returns: Tuple (bool, str) """ success = False err_msg = "" params = {"StreamName": stream_name} try: if not check_mode: if action == "increase": params["RetentionPeriodHours"] = retention_period client.increase_stream_retention_period(**params) success = True err_msg = f"Retention Period increased successfully to {retention_period}" elif action == "decrease": params["RetentionPeriodHours"] = retention_period client.decrease_stream_retention_period(**params) success = True err_msg = f"Retention Period decreased successfully to {retention_period}" else: err_msg = f"Invalid action {action}" else: if action == "increase": success = True elif action == "decrease": success = True else: err_msg = f"Invalid action {action}" except botocore.exceptions.ClientError as e: err_msg = to_native(e) return success, err_msg def update_shard_count(client, stream_name, number_of_shards=1, check_mode=False): """Increase or Decrease the number of shards in the Kinesis stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: number_of_shards (int): Number of shards this stream will use. default=1 check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> number_of_shards = 3 >>> update_shard_count(client, stream_name, number_of_shards) Returns: Tuple (bool, str) """ success = True err_msg = "" params = {"StreamName": stream_name, "ScalingType": "UNIFORM_SCALING"} if not check_mode: params["TargetShardCount"] = number_of_shards try: client.update_shard_count(**params) except botocore.exceptions.ClientError as e: return False, str(e) return success, err_msg def update( client, current_stream, stream_name, number_of_shards=1, retention_period=None, tags=None, wait=False, wait_timeout=300, check_mode=False, ): """Update an Amazon Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: number_of_shards (int): Number of shards this stream will use. default=1 retention_period (int): This is how long messages will be kept before they are discarded. This can not be less than 24 hours. tags (dict): The tags you want applied. wait (bool): Wait until Stream is ACTIVE. default=False wait_timeout (int): How long to wait until this operation is considered failed. default=300 check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> current_stream = { 'ShardCount': 3, 'HasMoreShards': True, 'RetentionPeriodHours': 24, 'StreamName': 'test-stream', 'StreamARN': 'arn:aws:kinesis:us-west-2:123456789:stream/test-stream', 'StreamStatus': "ACTIVE' } >>> stream_name = 'test-stream' >>> retention_period = 48 >>> number_of_shards = 10 >>> update(client, current_stream, stream_name, number_of_shards, retention_period ) Returns: Tuple (bool, bool, str) """ success = True changed = False err_msg = "" if retention_period: if wait: wait_success, wait_msg, current_stream = wait_for_status( client, stream_name, "ACTIVE", wait_timeout, check_mode=check_mode ) if not wait_success: return wait_success, False, wait_msg if current_stream.get("StreamStatus") == "ACTIVE": retention_changed = False if retention_period > current_stream["RetentionPeriodHours"]: retention_changed, retention_msg = retention_action( client, stream_name, retention_period, action="increase", check_mode=check_mode ) elif retention_period < current_stream["RetentionPeriodHours"]: retention_changed, retention_msg = retention_action( client, stream_name, retention_period, action="decrease", check_mode=check_mode ) elif retention_period == current_stream["RetentionPeriodHours"]: retention_msg = f"Retention {retention_period} is the same as {current_stream['RetentionPeriodHours']}" success = True if retention_changed: success = True changed = True err_msg = retention_msg if changed and wait: wait_success, wait_msg, current_stream = wait_for_status( client, stream_name, "ACTIVE", wait_timeout, check_mode=check_mode ) if not wait_success: return wait_success, False, wait_msg elif changed and not wait: stream_found, stream_msg, current_stream = find_stream(client, stream_name) if stream_found: if current_stream["StreamStatus"] != "ACTIVE": err_msg = f"Retention Period for {stream_name} is in the process of updating" return success, changed, err_msg else: err_msg = ( "StreamStatus has to be ACTIVE in order to modify the retention period." f" Current status is {current_stream.get('StreamStatus', 'UNKNOWN')}" ) return success, changed, err_msg if current_stream["OpenShardsCount"] != number_of_shards: success, err_msg = update_shard_count(client, stream_name, number_of_shards, check_mode=check_mode) if not success: return success, changed, err_msg changed = True if wait: wait_success, wait_msg, current_stream = wait_for_status( client, stream_name, "ACTIVE", wait_timeout, check_mode=check_mode ) if not wait_success: return wait_success, changed, wait_msg else: stream_found, stream_msg, current_stream = find_stream(client, stream_name) if stream_found and current_stream["StreamStatus"] != "ACTIVE": err_msg = f"Number of shards for {stream_name} is in the process of updating" return success, changed, err_msg if tags: tag_success, tag_changed, err_msg = update_tags(client, stream_name, tags, check_mode=check_mode) changed |= tag_changed if wait: success, err_msg, status_stream = wait_for_status( client, stream_name, "ACTIVE", wait_timeout, check_mode=check_mode ) if success and changed: err_msg = f"Kinesis Stream {stream_name} updated successfully." elif success and not changed: err_msg = f"Kinesis Stream {stream_name} did not change." return success, changed, err_msg def create_stream( client, stream_name, number_of_shards=1, retention_period=None, tags=None, wait=False, wait_timeout=300, check_mode=False, ): """Create an Amazon Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: number_of_shards (int): Number of shards this stream will use. default=1 retention_period (int): Can not be less than 24 hours default=None tags (dict): The tags you want applied. default=None wait (bool): Wait until Stream is ACTIVE. default=False wait_timeout (int): How long to wait until this operation is considered failed. default=300 check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> number_of_shards = 10 >>> tags = {'env': 'test'} >>> create_stream(client, stream_name, number_of_shards, tags=tags) Returns: Tuple (bool, bool, str, dict) """ success = False changed = False err_msg = "" results = dict() stream_found, stream_msg, current_stream = find_stream(client, stream_name) if stream_found and current_stream.get("StreamStatus") == "DELETING" and wait: wait_success, wait_msg, current_stream = wait_for_status( client, stream_name, "ACTIVE", wait_timeout, check_mode=check_mode ) if stream_found and current_stream.get("StreamStatus") != "DELETING": success, changed, err_msg = update( client, current_stream, stream_name, number_of_shards, retention_period, tags, wait, wait_timeout, check_mode=check_mode, ) else: create_success, create_msg = stream_action( client, stream_name, number_of_shards, action="create", check_mode=check_mode ) if not create_success: changed = True err_msg = f"Failed to create Kinesis stream: {create_msg}" return False, True, err_msg, {} else: changed = True if wait: wait_success, wait_msg, results = wait_for_status( client, stream_name, "ACTIVE", wait_timeout, check_mode=check_mode ) err_msg = f"Kinesis Stream {stream_name} is in the process of being created" if not wait_success: return wait_success, True, wait_msg, results else: err_msg = f"Kinesis Stream {stream_name} created successfully" if tags: changed, err_msg = tags_action(client, stream_name, tags, action="create", check_mode=check_mode) if changed: success = True if not success: return success, changed, err_msg, results stream_found, stream_msg, current_stream = find_stream(client, stream_name) if retention_period and current_stream.get("StreamStatus") == "ACTIVE": changed, err_msg = retention_action( client, stream_name, retention_period, action="increase", check_mode=check_mode ) if changed: success = True if not success: return success, changed, err_msg, results else: err_msg = ( "StreamStatus has to be ACTIVE in order to modify the retention period." f" Current status is {current_stream.get('StreamStatus', 'UNKNOWN')}" ) success = create_success changed = True if success: stream_found, stream_msg, results = find_stream(client, stream_name) tag_success, tag_msg, current_tags = get_tags(client, stream_name) if check_mode: current_tags = tags if not current_tags: current_tags = dict() results = camel_dict_to_snake_dict(results) results["tags"] = current_tags return success, changed, err_msg, results def delete_stream(client, stream_name, wait=False, wait_timeout=300, check_mode=False): """Delete an Amazon Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: wait (bool): Wait until Stream is ACTIVE. default=False wait_timeout (int): How long to wait until this operation is considered failed. default=300 check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> delete_stream(client, stream_name) Returns: Tuple (bool, bool, str, dict) """ success = False changed = False err_msg = "" results = dict() stream_found, stream_msg, current_stream = find_stream(client, stream_name) if stream_found: success, err_msg = stream_action(client, stream_name, action="delete", check_mode=check_mode) if success: changed = True if wait: success, err_msg, results = wait_for_status( client, stream_name, "DELETING", wait_timeout, check_mode=check_mode ) err_msg = f"Stream {stream_name} deleted successfully" if not success: return success, True, err_msg, results else: err_msg = f"Stream {stream_name} is in the process of being deleted" else: success = True changed = False err_msg = f"Stream {stream_name} does not exist" return success, changed, err_msg, results def start_stream_encryption( client, stream_name, encryption_type="", key_id="", wait=False, wait_timeout=300, check_mode=False ): """Start encryption on an Amazon Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: encryption_type (str): KMS or NONE key_id (str): KMS key GUID or alias wait (bool): Wait until Stream is ACTIVE. default=False wait_timeout (int): How long to wait until this operation is considered failed. default=300 check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> key_id = 'alias/aws' >>> encryption_type = 'KMS' >>> start_stream_encryption(client, stream_name,encryption_type,key_id) Returns: Tuple (bool, bool, str, dict) """ success = False changed = False err_msg = "" results = dict() stream_found, stream_msg, current_stream = find_stream(client, stream_name) if stream_found: if current_stream.get("EncryptionType") == encryption_type and current_stream.get("KeyId") == key_id: changed = False success = True err_msg = f"Kinesis Stream {stream_name} encryption already configured." else: success, err_msg = stream_encryption_action( client, stream_name, action="start_encryption", encryption_type=encryption_type, key_id=key_id, check_mode=check_mode, ) if success: changed = True if wait: success, err_msg, results = wait_for_status( client, stream_name, "ACTIVE", wait_timeout, check_mode=check_mode ) err_msg = f"Kinesis Stream {stream_name} encryption started successfully." if not success: return success, True, err_msg, results else: err_msg = f"Kinesis Stream {stream_name} is in the process of starting encryption." else: success = True changed = False err_msg = f"Kinesis Stream {stream_name} does not exist" if success: stream_found, stream_msg, results = find_stream(client, stream_name) tag_success, tag_msg, current_tags = get_tags(client, stream_name) if not current_tags: current_tags = dict() results = camel_dict_to_snake_dict(results) results["tags"] = current_tags return success, changed, err_msg, results def stop_stream_encryption( client, stream_name, encryption_type="", key_id="", wait=True, wait_timeout=300, check_mode=False ): """Stop encryption on an Amazon Kinesis Stream. Args: client (botocore.client.EC2): Boto3 client. stream_name (str): The name of the kinesis stream. Kwargs: encryption_type (str): KMS or NONE key_id (str): KMS key GUID or alias wait (bool): Wait until Stream is ACTIVE. default=False wait_timeout (int): How long to wait until this operation is considered failed. default=300 check_mode (bool): This will pass DryRun as one of the parameters to the aws api. default=False Basic Usage: >>> client = boto3.client('kinesis') >>> stream_name = 'test-stream' >>> stop_stream_encryption(client, stream_name,encryption_type, key_id) Returns: Tuple (bool, bool, str, dict) """ success = False changed = False err_msg = "" results = dict() stream_found, stream_msg, current_stream = find_stream(client, stream_name) if stream_found: if current_stream.get("EncryptionType") == "KMS": success, err_msg = stream_encryption_action( client, stream_name, action="stop_encryption", key_id=key_id, encryption_type=encryption_type, check_mode=check_mode, ) changed = success if wait: success, err_msg, results = wait_for_status( client, stream_name, "ACTIVE", wait_timeout, check_mode=check_mode ) if not success: return success, True, err_msg, results err_msg = f"Kinesis Stream {stream_name} encryption stopped successfully." else: err_msg = f"Stream {stream_name} is in the process of stopping encryption." elif current_stream.get("EncryptionType") == "NONE": success = True err_msg = f"Kinesis Stream {stream_name} encryption already stopped." else: success = True changed = False err_msg = f"Stream {stream_name} does not exist." if success: stream_found, stream_msg, results = find_stream(client, stream_name) tag_success, tag_msg, current_tags = get_tags(client, stream_name) if not current_tags: current_tags = dict() results = camel_dict_to_snake_dict(results) results["tags"] = current_tags return success, changed, err_msg, results def main(): argument_spec = dict( name=dict(required=True), shards=dict(default=None, required=False, type="int"), retention_period=dict(default=None, required=False, type="int"), tags=dict(default=None, required=False, type="dict", aliases=["resource_tags"]), wait=dict(default=True, required=False, type="bool"), wait_timeout=dict(default=300, required=False, type="int"), state=dict(default="present", choices=["present", "absent"]), encryption_type=dict(required=False, choices=["NONE", "KMS"]), key_id=dict(required=False, type="str"), encryption_state=dict(required=False, choices=["enabled", "disabled"]), ) module = AnsibleAWSModule( argument_spec=argument_spec, supports_check_mode=True, ) retention_period = module.params.get("retention_period") stream_name = module.params.get("name") shards = module.params.get("shards") state = module.params.get("state") tags = module.params.get("tags") wait = module.params.get("wait") wait_timeout = module.params.get("wait_timeout") encryption_type = module.params.get("encryption_type") key_id = module.params.get("key_id") encryption_state = module.params.get("encryption_state") if state == "present" and not shards: module.fail_json(msg="Shards is required when state == present.") if retention_period: if retention_period < 24: module.fail_json(msg="Retention period can not be less than 24 hours.") check_mode = module.check_mode try: client = module.client("kinesis") except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e: module.fail_json_aws(e, msg="Failed to connect to AWS") if state == "present": success, changed, err_msg, results = create_stream( client, stream_name, shards, retention_period, tags, wait, wait_timeout, check_mode ) if encryption_state == "enabled": success, changed, err_msg, results = start_stream_encryption( client, stream_name, encryption_type, key_id, wait, wait_timeout, check_mode ) elif encryption_state == "disabled": success, changed, err_msg, results = stop_stream_encryption( client, stream_name, encryption_type, key_id, wait, wait_timeout, check_mode ) elif state == "absent": success, changed, err_msg, results = delete_stream(client, stream_name, wait, wait_timeout, check_mode) if success: module.exit_json(success=success, changed=changed, msg=err_msg, **results) else: module.fail_json(success=success, changed=changed, msg=err_msg, result=results) if __name__ == "__main__": main()