diff --git a/plugins/modules/ec2_vpc_vgw.py b/plugins/modules/ec2_vpc_vgw.py
new file mode 100644
index 00000000000..a8ba032058c
--- /dev/null
+++ b/plugins/modules/ec2_vpc_vgw.py
@@ -0,0 +1,580 @@
+#!/usr/bin/python
+# Copyright: Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+ANSIBLE_METADATA = {'metadata_version': '1.1',
+                    'status': ['stableinterface'],
+                    'supported_by': 'community'}
+
+
+DOCUMENTATION = '''
+module: ec2_vpc_vgw
+short_description: Create and delete AWS VPN Virtual Gateways.
+description:
+  - Creates AWS VPN Virtual Gateways
+  - Deletes AWS VPN Virtual Gateways
+  - Attaches Virtual Gateways to VPCs
+  - Detaches Virtual Gateways from VPCs
+requirements: [ boto3 ]
+options:
+  state:
+    description:
+        - present to ensure resource is created.
+        - absent to remove resource
+    default: present
+    choices: [ "present", "absent"]
+    type: str
+  name:
+    description:
+        - name of the vgw to be created or deleted
+    type: str
+  type:
+    description:
+        - type of the virtual gateway to be created
+    choices: [ "ipsec.1" ]
+    default: "ipsec.1"
+    type: str
+  vpn_gateway_id:
+    description:
+        - vpn gateway id of an existing virtual gateway
+    type: str
+  vpc_id:
+    description:
+        - the vpc-id of a vpc to attach or detach
+    type: str
+  asn:
+    description:
+        - the BGP ASN of the amazon side
+    type: int
+  wait_timeout:
+    description:
+        - number of seconds to wait for status during vpc attach and detach
+    default: 320
+    type: int
+  tags:
+    description:
+        - dictionary of resource tags
+    aliases: [ "resource_tags" ]
+    type: dict
+author: Nick Aslanidis (@naslanidis)
+extends_documentation_fragment:
+- ansible.amazon.ec2
+- ansible.amazon.aws
+
+'''
+
+EXAMPLES = '''
+- name: Create a new vgw attached to a specific VPC
+  ec2_vpc_vgw:
+    state: present
+    region: ap-southeast-2
+    profile: personal
+    vpc_id: vpc-12345678
+    name: personal-testing
+    type: ipsec.1
+  register: created_vgw
+
+- name: Create a new unattached vgw
+  ec2_vpc_vgw:
+    state: present
+    region: ap-southeast-2
+    profile: personal
+    name: personal-testing
+    type: ipsec.1
+    tags:
+      environment: production
+      owner: ABC
+  register: created_vgw
+
+- name: Remove a new vgw using the name
+  ec2_vpc_vgw:
+    state: absent
+    region: ap-southeast-2
+    profile: personal
+    name: personal-testing
+    type: ipsec.1
+  register: deleted_vgw
+
+- name: Remove a new vgw using the vpn_gateway_id
+  ec2_vpc_vgw:
+    state: absent
+    region: ap-southeast-2
+    profile: personal
+    vpn_gateway_id: vgw-3a9aa123
+  register: deleted_vgw
+'''
+
+RETURN = '''
+result:
+  description: The result of the create, or delete action.
+  returned: success
+  type: dict
+'''
+
+import time
+import traceback
+
+try:
+    import botocore
+    import boto3
+    HAS_BOTO3 = True
+except ImportError:
+    HAS_BOTO3 = False
+
+from ansible_collections.ansible.amazon.plugins.module_utils.aws.core import is_boto3_error_code
+from ansible_collections.ansible.amazon.plugins.module_utils.aws.waiters import get_waiter
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.ansible.amazon.plugins.module_utils.ec2 import HAS_BOTO3, boto3_conn, ec2_argument_spec, get_aws_connection_info, AWSRetry
+from ansible.module_utils._text import to_native
+
+
+def get_vgw_info(vgws):
+    if not isinstance(vgws, list):
+        return
+
+    for vgw in vgws:
+        vgw_info = {
+            'id': vgw['VpnGatewayId'],
+            'type': vgw['Type'],
+            'state': vgw['State'],
+            'vpc_id': None,
+            'tags': dict()
+        }
+
+        for tag in vgw['Tags']:
+            vgw_info['tags'][tag['Key']] = tag['Value']
+
+        if len(vgw['VpcAttachments']) != 0 and vgw['VpcAttachments'][0]['State'] == 'attached':
+            vgw_info['vpc_id'] = vgw['VpcAttachments'][0]['VpcId']
+
+        return vgw_info
+
+
+def wait_for_status(client, module, vpn_gateway_id, status):
+    polling_increment_secs = 15
+    max_retries = (module.params.get('wait_timeout') // polling_increment_secs)
+    status_achieved = False
+
+    for x in range(0, max_retries):
+        try:
+            response = find_vgw(client, module, vpn_gateway_id)
+            if response[0]['VpcAttachments'][0]['State'] == status:
+                status_achieved = True
+                break
+            else:
+                time.sleep(polling_increment_secs)
+        except botocore.exceptions.ClientError as e:
+            module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    result = response
+    return status_achieved, result
+
+
+def attach_vgw(client, module, vpn_gateway_id):
+    params = dict()
+    params['VpcId'] = module.params.get('vpc_id')
+
+    try:
+        # Immediately after a detachment, the EC2 API sometimes will report the VpnGateways[0].State
+        # as available several seconds before actually permitting a new attachment.
+        # So we catch and retry that error.  See https://github.com/ansible/ansible/issues/53185
+        response = AWSRetry.jittered_backoff(retries=5,
+                                             catch_extra_error_codes=['InvalidParameterValue']
+                                             )(client.attach_vpn_gateway)(VpnGatewayId=vpn_gateway_id,
+                                                                          VpcId=params['VpcId'])
+    except botocore.exceptions.ClientError as e:
+        module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    status_achieved, vgw = wait_for_status(client, module, [vpn_gateway_id], 'attached')
+    if not status_achieved:
+        module.fail_json(msg='Error waiting for vpc to attach to vgw - please check the AWS console')
+
+    result = response
+    return result
+
+
+def detach_vgw(client, module, vpn_gateway_id, vpc_id=None):
+    params = dict()
+    params['VpcId'] = module.params.get('vpc_id')
+
+    if vpc_id:
+        try:
+            response = client.detach_vpn_gateway(VpnGatewayId=vpn_gateway_id, VpcId=vpc_id)
+        except botocore.exceptions.ClientError as e:
+            module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+    else:
+        try:
+            response = client.detach_vpn_gateway(VpnGatewayId=vpn_gateway_id, VpcId=params['VpcId'])
+        except botocore.exceptions.ClientError as e:
+            module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    status_achieved, vgw = wait_for_status(client, module, [vpn_gateway_id], 'detached')
+    if not status_achieved:
+        module.fail_json(msg='Error waiting for  vpc to detach from vgw - please check the AWS console')
+
+    result = response
+    return result
+
+
+def create_vgw(client, module):
+    params = dict()
+    params['Type'] = module.params.get('type')
+    if module.params.get('asn'):
+        params['AmazonSideAsn'] = module.params.get('asn')
+
+    try:
+        response = client.create_vpn_gateway(**params)
+        get_waiter(
+            client, 'vpn_gateway_exists'
+        ).wait(
+            VpnGatewayIds=[response['VpnGateway']['VpnGatewayId']]
+        )
+    except botocore.exceptions.WaiterError as e:
+        module.fail_json(msg="Failed to wait for Vpn Gateway {0} to be available".format(response['VpnGateway']['VpnGatewayId']),
+                         exception=traceback.format_exc())
+    except is_boto3_error_code('VpnGatewayLimitExceeded'):
+        module.fail_json(msg="Too many VPN gateways exist in this account.", exception=traceback.format_exc())
+    except botocore.exceptions.ClientError as e:  # pylint: disable=duplicate-except
+        module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    result = response
+    return result
+
+
+def delete_vgw(client, module, vpn_gateway_id):
+
+    try:
+        response = client.delete_vpn_gateway(VpnGatewayId=vpn_gateway_id)
+    except botocore.exceptions.ClientError as e:
+        module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    # return the deleted VpnGatewayId as this is not included in the above response
+    result = vpn_gateway_id
+    return result
+
+
+def create_tags(client, module, vpn_gateway_id):
+    params = dict()
+
+    try:
+        response = client.create_tags(Resources=[vpn_gateway_id], Tags=load_tags(module))
+    except botocore.exceptions.ClientError as e:
+        module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    result = response
+    return result
+
+
+def delete_tags(client, module, vpn_gateway_id, tags_to_delete=None):
+    params = dict()
+
+    if tags_to_delete:
+        try:
+            response = client.delete_tags(Resources=[vpn_gateway_id], Tags=tags_to_delete)
+        except botocore.exceptions.ClientError as e:
+            module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+    else:
+        try:
+            response = client.delete_tags(Resources=[vpn_gateway_id])
+        except botocore.exceptions.ClientError as e:
+            module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    result = response
+    return result
+
+
+def load_tags(module):
+    tags = []
+
+    if module.params.get('tags'):
+        for name, value in module.params.get('tags').items():
+            tags.append({'Key': name, 'Value': str(value)})
+        tags.append({'Key': "Name", 'Value': module.params.get('name')})
+    else:
+        tags.append({'Key': "Name", 'Value': module.params.get('name')})
+    return tags
+
+
+def find_tags(client, module, resource_id=None):
+
+    if resource_id:
+        try:
+            response = client.describe_tags(Filters=[
+                {'Name': 'resource-id', 'Values': [resource_id]}
+            ])
+        except botocore.exceptions.ClientError as e:
+            module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    result = response
+    return result
+
+
+def check_tags(client, module, existing_vgw, vpn_gateway_id):
+    params = dict()
+    params['Tags'] = module.params.get('tags')
+    vgw = existing_vgw
+    changed = False
+    tags_list = {}
+
+    # format tags for comparison
+    for tags in existing_vgw[0]['Tags']:
+        if tags['Key'] != 'Name':
+            tags_list[tags['Key']] = tags['Value']
+
+    # if existing tags don't match the tags arg, delete existing and recreate with new list
+    if params['Tags'] is not None and tags_list != params['Tags']:
+        delete_tags(client, module, vpn_gateway_id)
+        create_tags(client, module, vpn_gateway_id)
+        vgw = find_vgw(client, module)
+        changed = True
+
+    # if no tag args are supplied, delete any existing tags with the exception of the name tag
+    if params['Tags'] is None and tags_list != {}:
+        tags_to_delete = []
+        for tags in existing_vgw[0]['Tags']:
+            if tags['Key'] != 'Name':
+                tags_to_delete.append(tags)
+
+        delete_tags(client, module, vpn_gateway_id, tags_to_delete)
+        vgw = find_vgw(client, module)
+        changed = True
+
+    return vgw, changed
+
+
+def find_vpc(client, module):
+    params = dict()
+    params['vpc_id'] = module.params.get('vpc_id')
+
+    if params['vpc_id']:
+        try:
+            response = client.describe_vpcs(VpcIds=[params['vpc_id']])
+        except botocore.exceptions.ClientError as e:
+            module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    result = response
+    return result
+
+
+def find_vgw(client, module, vpn_gateway_id=None):
+    params = dict()
+    if vpn_gateway_id:
+        params['VpnGatewayIds'] = vpn_gateway_id
+    else:
+        params['Filters'] = [
+            {'Name': 'type', 'Values': [module.params.get('type')]},
+            {'Name': 'tag:Name', 'Values': [module.params.get('name')]},
+        ]
+        if module.params.get('state') == 'present':
+            params['Filters'].append({'Name': 'state', 'Values': ['pending', 'available']})
+    try:
+        response = client.describe_vpn_gateways(**params)
+    except botocore.exceptions.ClientError as e:
+        module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    return sorted(response['VpnGateways'], key=lambda k: k['VpnGatewayId'])
+
+
+def ensure_vgw_present(client, module):
+
+    # If an existing vgw name and type matches our args, then a match is considered to have been
+    # found and we will not create another vgw.
+
+    changed = False
+    params = dict()
+    result = dict()
+    params['Name'] = module.params.get('name')
+    params['VpcId'] = module.params.get('vpc_id')
+    params['Type'] = module.params.get('type')
+    params['Tags'] = module.params.get('tags')
+    params['VpnGatewayIds'] = module.params.get('vpn_gateway_id')
+
+    # check that the vpc_id exists. If not, an exception is thrown
+    if params['VpcId']:
+        vpc = find_vpc(client, module)
+
+    # check if a gateway matching our module args already exists
+    existing_vgw = find_vgw(client, module)
+
+    if existing_vgw != []:
+        vpn_gateway_id = existing_vgw[0]['VpnGatewayId']
+        vgw, changed = check_tags(client, module, existing_vgw, vpn_gateway_id)
+
+        # if a vpc_id was provided, check if it exists and if it's attached
+        if params['VpcId']:
+
+            current_vpc_attachments = existing_vgw[0]['VpcAttachments']
+
+            if current_vpc_attachments != [] and current_vpc_attachments[0]['State'] == 'attached':
+                if current_vpc_attachments[0]['VpcId'] != params['VpcId'] or current_vpc_attachments[0]['State'] != 'attached':
+                    # detach the existing vpc from the virtual gateway
+                    vpc_to_detach = current_vpc_attachments[0]['VpcId']
+                    detach_vgw(client, module, vpn_gateway_id, vpc_to_detach)
+                    get_waiter(client, 'vpn_gateway_detached').wait(VpnGatewayIds=[vpn_gateway_id])
+                    attached_vgw = attach_vgw(client, module, vpn_gateway_id)
+                    changed = True
+            else:
+                # attach the vgw to the supplied vpc
+                attached_vgw = attach_vgw(client, module, vpn_gateway_id)
+                changed = True
+
+        # if params['VpcId'] is not provided, check the vgw is attached to a vpc. if so, detach it.
+        else:
+            existing_vgw = find_vgw(client, module, [vpn_gateway_id])
+
+            if existing_vgw[0]['VpcAttachments'] != []:
+                if existing_vgw[0]['VpcAttachments'][0]['State'] == 'attached':
+                    # detach the vpc from the vgw
+                    vpc_to_detach = existing_vgw[0]['VpcAttachments'][0]['VpcId']
+                    detach_vgw(client, module, vpn_gateway_id, vpc_to_detach)
+                    changed = True
+
+    else:
+        # create a new vgw
+        new_vgw = create_vgw(client, module)
+        changed = True
+        vpn_gateway_id = new_vgw['VpnGateway']['VpnGatewayId']
+
+        # tag the new virtual gateway
+        create_tags(client, module, vpn_gateway_id)
+
+        # if a vpc-id was supplied, attempt to attach it to the vgw
+        if params['VpcId']:
+            attached_vgw = attach_vgw(client, module, vpn_gateway_id)
+            changed = True
+
+    # return current state of the vgw
+    vgw = find_vgw(client, module, [vpn_gateway_id])
+    result = get_vgw_info(vgw)
+    return changed, result
+
+
+def ensure_vgw_absent(client, module):
+
+    # If an existing vgw name and type matches our args, then a match is considered to have been
+    # found and we will take steps to delete it.
+
+    changed = False
+    params = dict()
+    result = dict()
+    params['Name'] = module.params.get('name')
+    params['VpcId'] = module.params.get('vpc_id')
+    params['Type'] = module.params.get('type')
+    params['Tags'] = module.params.get('tags')
+    params['VpnGatewayIds'] = module.params.get('vpn_gateway_id')
+
+    # check if a gateway matching our module args already exists
+    if params['VpnGatewayIds']:
+        existing_vgw_with_id = find_vgw(client, module, [params['VpnGatewayIds']])
+        if existing_vgw_with_id != [] and existing_vgw_with_id[0]['State'] != 'deleted':
+            existing_vgw = existing_vgw_with_id
+            if existing_vgw[0]['VpcAttachments'] != [] and existing_vgw[0]['VpcAttachments'][0]['State'] == 'attached':
+                if params['VpcId']:
+                    if params['VpcId'] != existing_vgw[0]['VpcAttachments'][0]['VpcId']:
+                        module.fail_json(msg='The vpc-id provided does not match the vpc-id currently attached - please check the AWS console')
+
+                    else:
+                        # detach the vpc from the vgw
+                        detach_vgw(client, module, params['VpnGatewayIds'], params['VpcId'])
+                        deleted_vgw = delete_vgw(client, module, params['VpnGatewayIds'])
+                        changed = True
+
+                else:
+                    # attempt to detach any attached vpcs
+                    vpc_to_detach = existing_vgw[0]['VpcAttachments'][0]['VpcId']
+                    detach_vgw(client, module, params['VpnGatewayIds'], vpc_to_detach)
+                    deleted_vgw = delete_vgw(client, module, params['VpnGatewayIds'])
+                    changed = True
+
+            else:
+                # no vpc's are attached so attempt to delete the vgw
+                deleted_vgw = delete_vgw(client, module, params['VpnGatewayIds'])
+                changed = True
+
+        else:
+            changed = False
+            deleted_vgw = "Nothing to do"
+
+    else:
+        # Check that a name and type argument has been supplied if no vgw-id
+        if not module.params.get('name') or not module.params.get('type'):
+            module.fail_json(msg='A name and type is required when no vgw-id and a status of \'absent\' is supplied')
+
+        existing_vgw = find_vgw(client, module)
+        if existing_vgw != [] and existing_vgw[0]['State'] != 'deleted':
+            vpn_gateway_id = existing_vgw[0]['VpnGatewayId']
+            if existing_vgw[0]['VpcAttachments'] != [] and existing_vgw[0]['VpcAttachments'][0]['State'] == 'attached':
+                if params['VpcId']:
+                    if params['VpcId'] != existing_vgw[0]['VpcAttachments'][0]['VpcId']:
+                        module.fail_json(msg='The vpc-id provided does not match the vpc-id currently attached - please check the AWS console')
+
+                    else:
+                        # detach the vpc from the vgw
+                        detach_vgw(client, module, vpn_gateway_id, params['VpcId'])
+
+                        # now that the vpc has been detached, delete the vgw
+                        deleted_vgw = delete_vgw(client, module, vpn_gateway_id)
+                        changed = True
+
+                else:
+                    # attempt to detach any attached vpcs
+                    vpc_to_detach = existing_vgw[0]['VpcAttachments'][0]['VpcId']
+                    detach_vgw(client, module, vpn_gateway_id, vpc_to_detach)
+                    changed = True
+
+                    # now that the vpc has been detached, delete the vgw
+                    deleted_vgw = delete_vgw(client, module, vpn_gateway_id)
+
+            else:
+                # no vpc's are attached so attempt to delete the vgw
+                deleted_vgw = delete_vgw(client, module, vpn_gateway_id)
+                changed = True
+
+        else:
+            changed = False
+            deleted_vgw = None
+
+    result = deleted_vgw
+    return changed, result
+
+
+def main():
+    argument_spec = ec2_argument_spec()
+    argument_spec.update(dict(
+        state=dict(default='present', choices=['present', 'absent']),
+        name=dict(),
+        vpn_gateway_id=dict(),
+        vpc_id=dict(),
+        asn=dict(type='int'),
+        wait_timeout=dict(type='int', default=320),
+        type=dict(default='ipsec.1', choices=['ipsec.1']),
+        tags=dict(default=None, required=False, type='dict', aliases=['resource_tags']),
+    )
+    )
+    module = AnsibleModule(argument_spec=argument_spec,
+                           required_if=[['state', 'present', ['name']]])
+
+    if not HAS_BOTO3:
+        module.fail_json(msg='json and boto3 is required.')
+
+    state = module.params.get('state').lower()
+
+    try:
+        region, ec2_url, aws_connect_kwargs = get_aws_connection_info(module, boto3=True)
+        client = boto3_conn(module, conn_type='client', resource='ec2', region=region, endpoint=ec2_url, **aws_connect_kwargs)
+    except botocore.exceptions.NoCredentialsError as e:
+        module.fail_json(msg="Can't authorize connection - %s" % to_native(e), exception=traceback.format_exc())
+
+    if state == 'present':
+        (changed, results) = ensure_vgw_present(client, module)
+    else:
+        (changed, results) = ensure_vgw_absent(client, module)
+    module.exit_json(changed=changed, vgw=results)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/plugins/modules/ec2_vpc_vgw_facts.py b/plugins/modules/ec2_vpc_vgw_facts.py
new file mode 120000
index 00000000000..bbcf44bef40
--- /dev/null
+++ b/plugins/modules/ec2_vpc_vgw_facts.py
@@ -0,0 +1 @@
+ec2_vpc_vgw_info.py
\ No newline at end of file
diff --git a/plugins/modules/ec2_vpc_vgw_info.py b/plugins/modules/ec2_vpc_vgw_info.py
new file mode 100644
index 00000000000..a1ac5b04c7b
--- /dev/null
+++ b/plugins/modules/ec2_vpc_vgw_info.py
@@ -0,0 +1,165 @@
+#!/usr/bin/python
+# Copyright: Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+ANSIBLE_METADATA = {'metadata_version': '1.1',
+                    'status': ['preview'],
+                    'supported_by': 'community'}
+
+
+DOCUMENTATION = '''
+---
+module: ec2_vpc_vgw_info
+short_description: Gather information about virtual gateways in AWS
+description:
+    - Gather information about virtual gateways in AWS.
+    - This module was called C(ec2_vpc_vgw_facts) before Ansible 2.9. The usage did not change.
+requirements: [ boto3 ]
+options:
+  filters:
+    description:
+      - A dict of filters to apply. Each dict item consists of a filter key and a filter value.
+        See U(https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeVpnGateways.html) for possible filters.
+    type: dict
+  vpn_gateway_ids:
+    description:
+      - Get details of a specific Virtual Gateway ID. This value should be provided as a list.
+    type: list
+    elements: str
+author: "Nick Aslanidis (@naslanidis)"
+extends_documentation_fragment:
+- ansible.amazon.aws
+- ansible.amazon.ec2
+
+'''
+
+EXAMPLES = '''
+# # Note: These examples do not set authentication details, see the AWS Guide for details.
+
+- name: Gather information about all virtual gateways for an account or profile
+  ec2_vpc_vgw_info:
+    region: ap-southeast-2
+    profile: production
+  register: vgw_info
+
+- name: Gather information about a filtered list of Virtual Gateways
+  ec2_vpc_vgw_info:
+    region: ap-southeast-2
+    profile: production
+    filters:
+        "tag:Name": "main-virt-gateway"
+  register: vgw_info
+
+- name: Gather information about a specific virtual gateway by VpnGatewayIds
+  ec2_vpc_vgw_info:
+    region: ap-southeast-2
+    profile: production
+    vpn_gateway_ids: vgw-c432f6a7
+  register: vgw_info
+'''
+
+RETURN = '''
+virtual_gateways:
+    description: The virtual gateways for the account.
+    returned: always
+    type: list
+    sample: [
+        {
+            "state": "available",
+            "tags": [
+                {
+                    "key": "Name",
+                    "value": "TEST-VGW"
+                }
+            ],
+            "type": "ipsec.1",
+            "vpc_attachments": [
+                {
+                    "state": "attached",
+                    "vpc_id": "vpc-22a93c74"
+                }
+            ],
+            "vpn_gateway_id": "vgw-23e3d64e"
+        }
+    ]
+
+changed:
+    description: True if listing the virtual gateways succeeds.
+    returned: always
+    type: bool
+    sample: "false"
+'''
+import traceback
+
+try:
+    import botocore
+except ImportError:
+    pass  # will be captured by imported HAS_BOTO3
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.ansible.amazon.plugins.module_utils.ec2 import (ec2_argument_spec, get_aws_connection_info, boto3_conn,
+                                      camel_dict_to_snake_dict, ansible_dict_to_boto3_filter_list, HAS_BOTO3)
+
+
+def get_virtual_gateway_info(virtual_gateway):
+    virtual_gateway_info = {'VpnGatewayId': virtual_gateway['VpnGatewayId'],
+                            'State': virtual_gateway['State'],
+                            'Type': virtual_gateway['Type'],
+                            'VpcAttachments': virtual_gateway['VpcAttachments'],
+                            'Tags': virtual_gateway.get('Tags', [])}
+    return virtual_gateway_info
+
+
+def list_virtual_gateways(client, module):
+    params = dict()
+
+    params['Filters'] = ansible_dict_to_boto3_filter_list(module.params.get('filters'))
+    params['DryRun'] = module.check_mode
+
+    if module.params.get("vpn_gateway_ids"):
+        params['VpnGatewayIds'] = module.params.get("vpn_gateway_ids")
+
+    try:
+        all_virtual_gateways = client.describe_vpn_gateways(**params)
+    except botocore.exceptions.ClientError as e:
+        module.fail_json(msg=str(e), exception=traceback.format_exc())
+
+    return [camel_dict_to_snake_dict(get_virtual_gateway_info(vgw))
+            for vgw in all_virtual_gateways['VpnGateways']]
+
+
+def main():
+    argument_spec = ec2_argument_spec()
+    argument_spec.update(
+        dict(
+            filters=dict(type='dict', default=dict()),
+            vpn_gateway_ids=dict(type='list', default=None)
+        )
+    )
+
+    module = AnsibleModule(argument_spec=argument_spec, supports_check_mode=True)
+    if module._name == 'ec2_vpc_vgw_facts':
+        module.deprecate("The 'ec2_vpc_vgw_facts' module has been renamed to 'ec2_vpc_vgw_info'", version='2.13')
+
+    # Validate Requirements
+    if not HAS_BOTO3:
+        module.fail_json(msg='json and boto3 is required.')
+
+    try:
+        region, ec2_url, aws_connect_kwargs = get_aws_connection_info(module, boto3=True)
+        connection = boto3_conn(module, conn_type='client', resource='ec2', region=region, endpoint=ec2_url, **aws_connect_kwargs)
+    except botocore.exceptions.NoCredentialsError as e:
+        module.fail_json(msg="Can't authorize connection - " + str(e))
+
+    # call your function here
+    results = list_virtual_gateways(connection, module)
+
+    module.exit_json(virtual_gateways=results)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/integration/targets/ec2_vpc_vgw/aliases b/tests/integration/targets/ec2_vpc_vgw/aliases
new file mode 100644
index 00000000000..6e3860bee23
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_vgw/aliases
@@ -0,0 +1,2 @@
+cloud/aws
+shippable/aws/group2
diff --git a/tests/integration/targets/ec2_vpc_vgw/tasks/main.yml b/tests/integration/targets/ec2_vpc_vgw/tasks/main.yml
new file mode 100644
index 00000000000..13365146e14
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_vgw/tasks/main.yml
@@ -0,0 +1,171 @@
+---
+- block:
+
+    # ============================================================
+    - name: set up aws connection info
+      set_fact:
+        aws_connection_info: &aws_connection_info
+          aws_access_key: "{{ aws_access_key }}"
+          aws_secret_key: "{{ aws_secret_key }}"
+          security_token: "{{ security_token }}"
+          region: "{{ aws_region }}"
+      no_log: yes
+
+    # ============================================================
+    - debug: msg="Setting up test dependencies"
+
+    - name: create a VPC
+      ec2_vpc_net:
+        name: "{{ resource_prefix }}-vpc-{{ item }}"
+        state: present
+        cidr_block: "10.0.0.0/26"
+        <<: *aws_connection_info
+        tags:
+          Name: "{{ resource_prefix }}-vpc-{{ item }}"
+          Description: "Created by ansible-test"
+      register: vpc_result
+      loop: [1, 2]
+
+    - name: use set fact for vpc ids
+      set_fact:
+        vpc_id_1: '{{ vpc_result.results.0.vpc.id }}'
+        vpc_id_2: '{{ vpc_result.results.1.vpc.id }}'
+
+    # ============================================================
+    - debug: msg="Running tests"
+
+    - name: create vpn gateway and attach it to vpc
+      ec2_vpc_vgw:
+        state: present
+        vpc_id: '{{ vpc_id_1 }}'
+        name: "{{ resource_prefix }}-vgw"
+        <<: *aws_connection_info
+      register: vgw
+
+    - assert:
+        that:
+          - vgw.changed
+          - "{{ vgw.vgw.vpc_id == vpc_id_1 }}"
+          - '"{{ vgw.vgw.tags.Name }}" == "{{ resource_prefix }}-vgw"'
+
+    - name: test idempotence
+      ec2_vpc_vgw:
+        state: present
+        vpc_id: '{{ vpc_id_1 }}'
+        name: "{{ resource_prefix }}-vgw"
+        <<: *aws_connection_info
+      register: vgw
+
+    - assert:
+        that:
+          - not vgw.changed
+
+    # ============================================================
+    - name: attach vpn gateway to the other VPC
+      ec2_vpc_vgw:
+        state: present
+        vpc_id: '{{ vpc_id_2 }}'
+        name: "{{ resource_prefix }}-vgw"
+        <<: *aws_connection_info
+      register: vgw
+
+    - assert:
+        that:
+          - vgw.changed
+          - "{{ vgw.vgw.vpc_id == vpc_id_2 }}"
+
+    # ============================================================
+    - name: add tags to the VGW
+      ec2_vpc_vgw:
+        state: present
+        vpc_id: '{{ vpc_id_2 }}'
+        name: "{{ resource_prefix }}-vgw"
+        tags:
+          created_by: ec2_vpc_vgw integration tests
+        <<: *aws_connection_info
+      register: vgw
+
+    - assert:
+        that:
+          - vgw.changed
+          - vgw.vgw.tags | length == 2
+          - "'created_by' in vgw.vgw.tags"
+
+    - name: test idempotence
+      ec2_vpc_vgw:
+        state: present
+        vpc_id: '{{ vpc_id_2 }}'
+        name: "{{ resource_prefix }}-vgw"
+        tags:
+          created_by: ec2_vpc_vgw integration tests
+        <<: *aws_connection_info
+      register: vgw
+
+    - assert:
+        that:
+          - not vgw.changed
+
+    # ============================================================
+    - name: remove tags from the VGW
+      ec2_vpc_vgw:
+        state: present
+        vpc_id: '{{ vpc_id_2 }}'
+        name: "{{ resource_prefix }}-vgw"
+        <<: *aws_connection_info
+      register: vgw
+
+    - assert:
+        that:
+          - vgw.changed
+          - vgw.vgw.tags | length == 1
+          - '"{{ vgw.vgw.tags.Name }}" == "{{ resource_prefix }}-vgw"'
+
+    # ============================================================
+    - name: detach vpn gateway
+      ec2_vpc_vgw:
+        state: present
+        name: "{{ resource_prefix }}-vgw"
+        <<: *aws_connection_info
+      register: vgw
+
+    - assert:
+        that:
+          - vgw.changed
+          - not vgw.vgw.vpc_id
+
+    - name: test idempotence
+      ec2_vpc_vgw:
+        state: present
+        name: "{{ resource_prefix }}-vgw"
+        <<: *aws_connection_info
+      register: vgw
+
+    - assert:
+        that:
+          - not vgw.changed
+
+    # ============================================================
+
+  always:
+
+    - debug: msg="Removing test dependencies"
+
+    - name: delete vpn gateway
+      ec2_vpc_vgw:
+        state: absent
+        vpn_gateway_id: '{{ vgw.vgw.id }}'
+        <<: *aws_connection_info
+      ignore_errors: yes
+
+    - name: delete vpc
+      ec2_vpc_net:
+        name: "{{ resource_prefix }}-vpc-{{ item }}"
+        state: absent
+        cidr_block: "10.0.0.0/26"
+        <<: *aws_connection_info
+      loop: [1, 2]
+      register: result
+      retries: 10
+      delay: 5
+      until: result is not failed
+      ignore_errors: true