mirror of
https://github.com/ansible-collections/google.cloud.git
synced 2025-04-05 10:20:26 -07:00
This commit is contained in:
parent
e23c44c612
commit
8b9a2c70dd
3 changed files with 139 additions and 2 deletions
2
changelogs/gcp_pubsub_subscription_gcs_feature.yaml
Normal file
2
changelogs/gcp_pubsub_subscription_gcs_feature.yaml
Normal file
|
@ -0,0 +1,2 @@
|
||||||
|
features:
|
||||||
|
- gcp_pubsub_subscription - allows to create GCS subscription
|
|
@ -63,6 +63,65 @@ options:
|
||||||
}}"'
|
}}"'
|
||||||
required: true
|
required: true
|
||||||
type: dict
|
type: dict
|
||||||
|
cloud_storage:
|
||||||
|
description:
|
||||||
|
- Cloud Storage Subscription configuration.
|
||||||
|
required: false
|
||||||
|
type: dict
|
||||||
|
suboptions:
|
||||||
|
bucket:
|
||||||
|
description:
|
||||||
|
- A reference to a Bucket resource.
|
||||||
|
- This field represents a name of a Bucket resource in GCP where messages read from a Topic will be stored.
|
||||||
|
required: true
|
||||||
|
type: str
|
||||||
|
file_prefix:
|
||||||
|
description:
|
||||||
|
- File object name prefix stored in a Bucket.
|
||||||
|
required: false
|
||||||
|
type: str
|
||||||
|
file_suffix:
|
||||||
|
description:
|
||||||
|
- File object name suffix stored in a Bucket.
|
||||||
|
required: false
|
||||||
|
type: str
|
||||||
|
file_datetime_format:
|
||||||
|
description:
|
||||||
|
- File object datetime format stored in a Bucket.
|
||||||
|
required: false
|
||||||
|
type: str
|
||||||
|
max_duration:
|
||||||
|
description:
|
||||||
|
- Subscription writes a new output file if the specified value of max duration is exceeded. Min: 1m, max: 10m.
|
||||||
|
required: true
|
||||||
|
type: str
|
||||||
|
max_bytes:
|
||||||
|
description:
|
||||||
|
- Cloud Storage Subscription writes a new output file if the specified value of max bytes is exceeded. Min: 1KB, max: 10GiB.
|
||||||
|
required: false
|
||||||
|
type: str
|
||||||
|
max_messages:
|
||||||
|
description:
|
||||||
|
- Cloud Storage Subscription writes a new output file if the specified number of messages is exceeded. Min: 1000.
|
||||||
|
required: false
|
||||||
|
type: int
|
||||||
|
output_format:
|
||||||
|
description:
|
||||||
|
- Specify the format of the output files that are to be stored in a Cloud Storage bucket as text or avro.
|
||||||
|
required: true
|
||||||
|
type: str
|
||||||
|
write_metadata:
|
||||||
|
description:
|
||||||
|
- This option allows to store the message metadata along with the message, e.g. message headers.
|
||||||
|
- This field is valid for avro message format only.
|
||||||
|
required: false
|
||||||
|
type: bool
|
||||||
|
use_topic_schema:
|
||||||
|
description:
|
||||||
|
- When true, the output Cloud Storage file will be serialized using the topic schema, if it exists.
|
||||||
|
- This field is valid for avro message format only.
|
||||||
|
required: false
|
||||||
|
type: bool
|
||||||
labels:
|
labels:
|
||||||
description:
|
description:
|
||||||
- A set of key/value label pairs to assign to this Subscription.
|
- A set of key/value label pairs to assign to this Subscription.
|
||||||
|
@ -576,6 +635,21 @@ def main():
|
||||||
state=dict(default='present', choices=['present', 'absent'], type='str'),
|
state=dict(default='present', choices=['present', 'absent'], type='str'),
|
||||||
name=dict(required=True, type='str'),
|
name=dict(required=True, type='str'),
|
||||||
topic=dict(required=True, type='dict'),
|
topic=dict(required=True, type='dict'),
|
||||||
|
cloud_storage=dict(
|
||||||
|
type='dict',
|
||||||
|
options=dict(
|
||||||
|
bucket=dict(required=True, type='str'),
|
||||||
|
file_prefix=dict(type='str'),
|
||||||
|
file_suffix=dict(type='str'),
|
||||||
|
file_datetime_format=dict(type='str'),
|
||||||
|
max_duration=dict(type='str'),
|
||||||
|
max_bytes=dict(type='str'),
|
||||||
|
max_messages=dict(type='int'),
|
||||||
|
output_format=dict(type='str'),
|
||||||
|
write_metadata=dict(type='bool'),
|
||||||
|
use_topic_schema=dict(type='bool'),
|
||||||
|
),
|
||||||
|
),
|
||||||
labels=dict(type='dict'),
|
labels=dict(type='dict'),
|
||||||
push_config=dict(
|
push_config=dict(
|
||||||
type='dict',
|
type='dict',
|
||||||
|
@ -642,6 +716,8 @@ def update(module, link, fetch):
|
||||||
|
|
||||||
def updateMask(request, response):
|
def updateMask(request, response):
|
||||||
update_mask = []
|
update_mask = []
|
||||||
|
if request.get('cloudStorageConfig') != response.get('cloudStorageConfig'):
|
||||||
|
update_mask.append('cloudStorageConfig')
|
||||||
if request.get('labels') != response.get('labels'):
|
if request.get('labels') != response.get('labels'):
|
||||||
update_mask.append('labels')
|
update_mask.append('labels')
|
||||||
if request.get('pushConfig') != response.get('pushConfig'):
|
if request.get('pushConfig') != response.get('pushConfig'):
|
||||||
|
@ -671,6 +747,7 @@ def resource_to_request(module):
|
||||||
u'name': name_pattern(module.params.get('name'), module),
|
u'name': name_pattern(module.params.get('name'), module),
|
||||||
u'topic': topic_pattern(replace_resource_dict(module.params.get(u'topic', {}), 'name'), module),
|
u'topic': topic_pattern(replace_resource_dict(module.params.get(u'topic', {}), 'name'), module),
|
||||||
u'labels': module.params.get('labels'),
|
u'labels': module.params.get('labels'),
|
||||||
|
u'cloudStorageConfig': SubscriptionCloudStorageConfig(module.params.get(u'cloud_storage', {}), module).to_request(),
|
||||||
u'pushConfig': SubscriptionPushconfig(module.params.get('push_config', {}), module).to_request(),
|
u'pushConfig': SubscriptionPushconfig(module.params.get('push_config', {}), module).to_request(),
|
||||||
u'ackDeadlineSeconds': module.params.get('ack_deadline_seconds'),
|
u'ackDeadlineSeconds': module.params.get('ack_deadline_seconds'),
|
||||||
u'messageRetentionDuration': module.params.get('message_retention_duration'),
|
u'messageRetentionDuration': module.params.get('message_retention_duration'),
|
||||||
|
@ -748,6 +825,7 @@ def response_to_hash(module, response):
|
||||||
u'name': name_pattern(module.params.get('name'), module),
|
u'name': name_pattern(module.params.get('name'), module),
|
||||||
u'topic': topic_pattern(replace_resource_dict(module.params.get(u'topic', {}), 'name'), module),
|
u'topic': topic_pattern(replace_resource_dict(module.params.get(u'topic', {}), 'name'), module),
|
||||||
u'labels': response.get(u'labels'),
|
u'labels': response.get(u'labels'),
|
||||||
|
u'cloudStorageConfig': SubscriptionCloudStorageConfig(response.get(u'cloudStorageConfig', {}), module).from_response(),
|
||||||
u'pushConfig': SubscriptionPushconfig(response.get(u'pushConfig', {}), module).from_response(),
|
u'pushConfig': SubscriptionPushconfig(response.get(u'pushConfig', {}), module).from_response(),
|
||||||
u'ackDeadlineSeconds': response.get(u'ackDeadlineSeconds'),
|
u'ackDeadlineSeconds': response.get(u'ackDeadlineSeconds'),
|
||||||
u'messageRetentionDuration': response.get(u'messageRetentionDuration'),
|
u'messageRetentionDuration': response.get(u'messageRetentionDuration'),
|
||||||
|
@ -880,6 +958,48 @@ class SubscriptionRetrypolicy(object):
|
||||||
def from_response(self):
|
def from_response(self):
|
||||||
return remove_nones_from_dict({u'minimumBackoff': self.request.get(u'minimumBackoff'), u'maximumBackoff': self.request.get(u'maximumBackoff')})
|
return remove_nones_from_dict({u'minimumBackoff': self.request.get(u'minimumBackoff'), u'maximumBackoff': self.request.get(u'maximumBackoff')})
|
||||||
|
|
||||||
|
class SubscriptionCloudStorageConfig(object):
|
||||||
|
def __init__(self, request, module):
|
||||||
|
self.module = module
|
||||||
|
if request:
|
||||||
|
self.request = request
|
||||||
|
else:
|
||||||
|
self.request = {}
|
||||||
|
|
||||||
|
def to_request(self):
|
||||||
|
return remove_nones_from_dict(
|
||||||
|
{
|
||||||
|
u'bucket': self.request.get('bucket'),
|
||||||
|
u'filenamePrefix': self.request.get('file_prefix', {}),
|
||||||
|
u'filenameSuffix': self.request.get('file_suffix', {}),
|
||||||
|
u'filenameDatetimeFormat': self.request.get('file_datetime_format', {}),
|
||||||
|
u'maxDuration': self.request.get('max_duration', {}),
|
||||||
|
u'maxBytes': self.request.get('max_bytes', {}),
|
||||||
|
u'maxMessages': self.request.get('max_messages', {}),
|
||||||
|
u'avroConfig': {'writeMetadata': self.request.get('write_metadata', False),
|
||||||
|
'useTopicSchema': self.request.get('use_topic_schema', False)}
|
||||||
|
if self.request.get('output_format', {}) == 'avro'
|
||||||
|
else {},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
def from_response(self):
|
||||||
|
return remove_nones_from_dict(
|
||||||
|
{
|
||||||
|
u'bucket': self.request.get('bucket'),
|
||||||
|
u'filenamePrefix': self.request.get('filenamePrefix', {}),
|
||||||
|
u'filenameSuffix': self.request.get('filenameSuffix', {}),
|
||||||
|
u'filenameDatetimeFormat': self.request.get('filenameDatetimeFormat', {}),
|
||||||
|
u'maxDuration': self.request.get('maxDuration', {}),
|
||||||
|
u'maxBytes': self.request.get('maxBytes', {}),
|
||||||
|
u'maxMessages': self.request.get('maxMessages', {}),
|
||||||
|
u'avroConfig': {'writeMetadata': self.request.get('avroConfig', {}).get('writeMetadata', False),
|
||||||
|
'useTopicSchema': self.request.get('avroConfig', {}).get('useTopicSchema', False)}
|
||||||
|
if self.request.get('avroConfig', {})
|
||||||
|
else {},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
main()
|
main()
|
||||||
|
|
|
@ -73,11 +73,26 @@
|
||||||
that:
|
that:
|
||||||
- result.changed == false
|
- result.changed == false
|
||||||
#----------------------------------------------------------
|
#----------------------------------------------------------
|
||||||
|
- name: Update ack_deadline_seconds of a subscription that already exists
|
||||||
|
google.cloud.gcp_pubsub_subscription:
|
||||||
|
name: "{{ resource_name }}"
|
||||||
|
topic: "{{ topic }}"
|
||||||
|
ack_deadline_seconds: 500
|
||||||
|
project: "{{ gcp_project }}"
|
||||||
|
auth_kind: "{{ gcp_cred_kind }}"
|
||||||
|
service_account_file: "{{ gcp_cred_file | default(omit) }}"
|
||||||
|
state: present
|
||||||
|
register: result
|
||||||
|
- name: Assert changed is true
|
||||||
|
ansible.builtin.assert:
|
||||||
|
that:
|
||||||
|
- result.changed == true
|
||||||
|
#----------------------------------------------------------
|
||||||
- name: Delete a subscription
|
- name: Delete a subscription
|
||||||
google.cloud.gcp_pubsub_subscription:
|
google.cloud.gcp_pubsub_subscription:
|
||||||
name: "{{ resource_name }}"
|
name: "{{ resource_name }}"
|
||||||
topic: "{{ topic }}"
|
topic: "{{ topic }}"
|
||||||
ack_deadline_seconds: 300
|
ack_deadline_seconds: 500
|
||||||
project: "{{ gcp_project }}"
|
project: "{{ gcp_project }}"
|
||||||
auth_kind: "{{ gcp_cred_kind }}"
|
auth_kind: "{{ gcp_cred_kind }}"
|
||||||
service_account_file: "{{ gcp_cred_file | default(omit) }}"
|
service_account_file: "{{ gcp_cred_file | default(omit) }}"
|
||||||
|
@ -104,7 +119,7 @@
|
||||||
google.cloud.gcp_pubsub_subscription:
|
google.cloud.gcp_pubsub_subscription:
|
||||||
name: "{{ resource_name }}"
|
name: "{{ resource_name }}"
|
||||||
topic: "{{ topic }}"
|
topic: "{{ topic }}"
|
||||||
ack_deadline_seconds: 300
|
ack_deadline_seconds: 500
|
||||||
project: "{{ gcp_project }}"
|
project: "{{ gcp_project }}"
|
||||||
auth_kind: "{{ gcp_cred_kind }}"
|
auth_kind: "{{ gcp_cred_kind }}"
|
||||||
service_account_file: "{{ gcp_cred_file | default(omit) }}"
|
service_account_file: "{{ gcp_cred_file | default(omit) }}"
|
||||||
|
|
Loading…
Add table
Reference in a new issue