diff --git a/changelogs/fragments/356-ec2_snapshot-boto3-migration.yml b/changelogs/fragments/356-ec2_snapshot-boto3-migration.yml
new file mode 100644
index 00000000000..f7a032c8408
--- /dev/null
+++ b/changelogs/fragments/356-ec2_snapshot-boto3-migration.yml
@@ -0,0 +1,4 @@
+minor_changes:
+- ec2_snapshot - migrated to use the boto3 python library (https://github.com/ansible-collections/amazon.aws/pull/356).
+breaking_changes:
+- ec2_snapshot - support for waiting indefinitely has been dropped, new default is 10 minutes (https://github.com/ansible-collections/amazon.aws/pull/356).
diff --git a/plugins/module_utils/waiters.py b/plugins/module_utils/waiters.py
index 0996c4a0781..c69454c2af5 100644
--- a/plugins/module_utils/waiters.py
+++ b/plugins/module_utils/waiters.py
@@ -162,6 +162,19 @@
                 },
             ]
         },
+        "SnapshotCompleted": {
+            "delay": 15,
+            "operation": "DescribeSnapshots",
+            "maxAttempts": 40,
+            "acceptors": [
+                {
+                    "expected": "completed",
+                    "matcher": "pathAll",
+                    "state": "success",
+                    "argument": "Snapshots[].State"
+                }
+            ]
+        },
         "SubnetAvailable": {
             "delay": 15,
             "operation": "DescribeSubnets",
@@ -582,6 +595,12 @@ def route53_model(name):
         core_waiter.NormalizedOperationMethod(
             ec2.describe_security_groups
         )),
+    ('EC2', 'snapshot_completed'): lambda ec2: core_waiter.Waiter(
+        'snapshot_completed',
+        ec2_model('SnapshotCompleted'),
+        core_waiter.NormalizedOperationMethod(
+            ec2.describe_snapshots
+        )),
     ('EC2', 'subnet_available'): lambda ec2: core_waiter.Waiter(
         'subnet_available',
         ec2_model('SubnetAvailable'),
diff --git a/plugins/modules/ec2_snapshot.py b/plugins/modules/ec2_snapshot.py
index cf4762dd414..00e9ac3cdb9 100644
--- a/plugins/modules/ec2_snapshot.py
+++ b/plugins/modules/ec2_snapshot.py
@@ -37,6 +37,8 @@
   snapshot_tags:
     description:
       - A dictionary of tags to add to the snapshot.
+      - If the volume has a C(Name) tag this will be automatically added to the
+        snapshot.
     type: dict
     required: false
   wait:
@@ -48,9 +50,8 @@
   wait_timeout:
     description:
       - How long before wait gives up, in seconds.
-      - Specify 0 to wait forever.
     required: false
-    default: 0
+    default: 600
     type: int
   state:
     description:
@@ -132,22 +133,22 @@
     sample: 8
 '''
 
-import time
 import datetime
 
 try:
-    import boto.exception
+    import botocore
 except ImportError:
-    pass  # Taken care of by ec2.HAS_BOTO
-
-from ..module_utils.core import AnsibleAWSModule
-from ..module_utils.ec2 import HAS_BOTO
-from ..module_utils.ec2 import ec2_connect
+    pass  # Taken care of by AnsibleAWSModule
 
+from ansible.module_utils.common.dict_transformations import camel_dict_to_snake_dict
 
-# Find the most recent snapshot
-def _get_snapshot_starttime(snap):
-    return datetime.datetime.strptime(snap.start_time, '%Y-%m-%dT%H:%M:%S.%fZ')
+from ..module_utils.core import AnsibleAWSModule
+from ..module_utils.core import is_boto3_error_code
+from ..module_utils.ec2 import AWSRetry
+from ..module_utils.ec2 import ansible_dict_to_boto3_filter_list
+from ..module_utils.ec2 import ansible_dict_to_boto3_tag_list
+from ..module_utils.ec2 import boto3_tag_list_to_ansible_dict
+from ..module_utils.waiters import get_waiter
 
 
 def _get_most_recent_snapshot(snapshots, max_snapshot_age_secs=None, now=None):
@@ -163,12 +164,10 @@ def _get_most_recent_snapshot(snapshots, max_snapshot_age_secs=None, now=None):
         return None
 
     if not now:
-        now = datetime.datetime.utcnow()
-
-    youngest_snapshot = max(snapshots, key=_get_snapshot_starttime)
+        now = datetime.datetime.now(datetime.timezone.utc)
 
-    # See if the snapshot is younger that the given max age
-    snapshot_start = datetime.datetime.strptime(youngest_snapshot.start_time, '%Y-%m-%dT%H:%M:%S.%fZ')
+    youngest_snapshot = max(snapshots, key=lambda s: s['StartTime'])
+    snapshot_start = youngest_snapshot['StartTime']
     snapshot_age = now - snapshot_start
 
     if max_snapshot_age_secs is not None:
@@ -178,92 +177,158 @@ def _get_most_recent_snapshot(snapshots, max_snapshot_age_secs=None, now=None):
     return youngest_snapshot
 
 
-def _create_with_wait(snapshot, wait_timeout_secs, sleep_func=time.sleep):
-    """
-    Wait for the snapshot to be created
-    :param snapshot:
-    :param wait_timeout_secs: fail this step after this many seconds
-    :param sleep_func:
-    :return:
-    """
-    time_waited = 0
-    snapshot.update()
-    while snapshot.status != 'completed':
-        sleep_func(3)
-        snapshot.update()
-        time_waited += 3
-        if wait_timeout_secs and time_waited > wait_timeout_secs:
-            return False
-    return True
-
-
-def create_snapshot(module, ec2, state=None, description=None, wait=None,
+def get_volume_by_instance(module, ec2, device_name, instance_id):
+    try:
+        _filter = {
+            'attachment.instance-id': instance_id,
+            'attachment.device': device_name
+        }
+        volumes = ec2.describe_volumes(
+            aws_retry=True,
+            Filters=ansible_dict_to_boto3_filter_list(_filter)
+        )['Volumes']
+    except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:
+        module.fail_json_aws(e, msg="Failed to describe Volume")
+
+    if not volumes:
+        module.fail_json(
+            msg="Could not find volume with name {0} attached to instance {1}".format(
+                device_name, instance_id
+            )
+        )
+
+    volume = volumes[0]
+    return volume
+
+
+def get_volume_by_id(module, ec2, volume):
+    try:
+        volumes = ec2.describe_volumes(
+            aws_retry=True,
+            VolumeIds=[volume],
+        )['Volumes']
+    except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:
+        module.fail_json_aws(e, msg="Failed to describe Volume")
+
+    if not volumes:
+        module.fail_json(
+            msg="Could not find volume with id {0}".format(volume)
+        )
+
+    volume = volumes[0]
+    return volume
+
+
+@AWSRetry.jittered_backoff()
+def _describe_snapshots(ec2, **params):
+    paginator = ec2.get_paginator('describe_snapshots')
+    return paginator.paginate(**params).build_full_result()
+
+
+# Handle SnapshotCreationPerVolumeRateExceeded separately because we need a much
+# longer delay than normal
+@AWSRetry.jittered_backoff(catch_extra_error_codes=['SnapshotCreationPerVolumeRateExceeded'], delay=15)
+def _create_snapshot(ec2, **params):
+    # Fast retry on common failures ('global' rate limits)
+    return ec2.create_snapshot(aws_retry=True, **params)
+
+
+def get_snapshots_by_volume(module, ec2, volume_id):
+    _filter = {'volume-id': volume_id}
+    try:
+        results = _describe_snapshots(
+            ec2,
+            Filters=ansible_dict_to_boto3_filter_list(_filter)
+        )
+    except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:
+        module.fail_json_aws(e, msg="Failed to describe snapshots from volume")
+
+    return results['Snapshots']
+
+
+def create_snapshot(module, ec2, description=None, wait=None,
                     wait_timeout=None, volume_id=None, instance_id=None,
                     snapshot_id=None, device_name=None, snapshot_tags=None,
                     last_snapshot_min_age=None):
     snapshot = None
     changed = False
 
-    required = [volume_id, snapshot_id, instance_id]
-    if required.count(None) != len(required) - 1:  # only 1 must be set
-        module.fail_json(msg='One and only one of volume_id or instance_id or snapshot_id must be specified')
-    if instance_id and not device_name or device_name and not instance_id:
-        module.fail_json(msg='Instance ID and device name must both be specified')
-
     if instance_id:
-        try:
-            volumes = ec2.get_all_volumes(filters={'attachment.instance-id': instance_id, 'attachment.device': device_name})
-        except boto.exception.BotoServerError as e:
-            module.fail_json_aws(e)
-
-        if not volumes:
-            module.fail_json(msg="Could not find volume with name %s attached to instance %s" % (device_name, instance_id))
-
-        volume_id = volumes[0].id
-
-    if state == 'absent':
-        if not snapshot_id:
-            module.fail_json(msg='snapshot_id must be set when state is absent')
-        try:
-            ec2.delete_snapshot(snapshot_id)
-        except boto.exception.BotoServerError as e:
-            # exception is raised if snapshot does not exist
-            if e.error_code == 'InvalidSnapshot.NotFound':
-                module.exit_json(changed=False)
-            else:
-                module.fail_json_aws(e)
-
-        # successful delete
-        module.exit_json(changed=True)
+        volume = get_volume_by_instance(
+            module, ec2, device_name, instance_id
+        )
+        volume_id = volume['VolumeId']
+    else:
+        volume = get_volume_by_id(module, ec2, volume_id)
 
     if last_snapshot_min_age > 0:
-        try:
-            current_snapshots = ec2.get_all_snapshots(filters={'volume_id': volume_id})
-        except boto.exception.BotoServerError as e:
-            module.fail_json_aws(e)
-
+        current_snapshots = get_snapshots_by_volume(module, ec2, volume_id)
         last_snapshot_min_age = last_snapshot_min_age * 60  # Convert to seconds
-        snapshot = _get_most_recent_snapshot(current_snapshots,
-                                             max_snapshot_age_secs=last_snapshot_min_age)
-    try:
-        # Create a new snapshot if we didn't find an existing one to use
-        if snapshot is None:
-            snapshot = ec2.create_snapshot(volume_id, description=description)
-            changed = True
-        if wait:
-            if not _create_with_wait(snapshot, wait_timeout):
-                module.fail_json(msg='Timed out while creating snapshot.')
+        snapshot = _get_most_recent_snapshot(
+            current_snapshots,
+            max_snapshot_age_secs=last_snapshot_min_age
+        )
+    # Create a new snapshot if we didn't find an existing one to use
+    if snapshot is None:
+        volume_tags = boto3_tag_list_to_ansible_dict(volume['Tags'])
+        volume_name = volume_tags.get('Name')
+        _tags = dict()
+        if volume_name:
+            _tags['Name'] = volume_name
         if snapshot_tags:
-            for k, v in snapshot_tags.items():
-                snapshot.add_tag(k, v)
-    except boto.exception.BotoServerError as e:
-        module.fail_json_aws(e)
+            _tags.update(snapshot_tags)
+
+        params = {'VolumeId': volume_id}
+        if description:
+            params['Description'] = description
+        if _tags:
+            params['TagSpecifications'] = [{
+                'ResourceType': 'snapshot',
+                'Tags': ansible_dict_to_boto3_tag_list(_tags),
+            }]
+        try:
+            snapshot = _create_snapshot(ec2, **params)
+        except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:
+            module.fail_json_aws(e, msg="Failed to create snapshot")
+        changed = True
+    if wait:
+        waiter = get_waiter(ec2, 'snapshot_completed')
+        try:
+            waiter.wait(
+                SnapshotIds=[snapshot['SnapshotId']],
+                WaiterConfig=dict(Delay=3, MaxAttempts=wait_timeout // 3)
+            )
+        except botocore.exceptions.WaiterError as e:
+            module.fail_json_aws(e, msg='Timed out while creating snapshot')
+        except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:
+            module.fail_json_aws(
+                e, msg='Error while waiting for snapshot creation'
+            )
+
+    _tags = boto3_tag_list_to_ansible_dict(snapshot['Tags'])
+    _snapshot = camel_dict_to_snake_dict(snapshot)
+    _snapshot['tags'] = _tags
+    results = {
+        'snapshot_id': snapshot['SnapshotId'],
+        'volume_id': snapshot['VolumeId'],
+        'volume_size': snapshot['VolumeSize'],
+        'tags': _tags,
+        'snapshots': [_snapshot],
+    }
+
+    module.exit_json(changed=changed, **results)
+
+
+def delete_snapshot(module, ec2, snapshot_id):
+    try:
+        ec2.delete_snapshot(aws_retry=True, SnapshotId=snapshot_id)
+    except is_boto3_error_code('InvalidSnapshot.NotFound'):
+        module.exit_json(changed=False)
+    except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:  # pylint: disable=duplicate-except
+        module.fail_json_aws(e, msg="Failed to delete snapshot")
 
-    module.exit_json(changed=changed,
-                     snapshot_id=snapshot.id,
-                     volume_id=snapshot.volume_id,
-                     volume_size=snapshot.volume_size,
-                     tags=snapshot.tags.copy())
+    # successful delete
+    module.exit_json(changed=True)
 
 
 def create_snapshot_ansible_module():
@@ -274,21 +339,38 @@ def create_snapshot_ansible_module():
         snapshot_id=dict(),
         device_name=dict(),
         wait=dict(type='bool', default=True),
-        wait_timeout=dict(type='int', default=0),
+        wait_timeout=dict(type='int', default=600),
         last_snapshot_min_age=dict(type='int', default=0),
         snapshot_tags=dict(type='dict', default=dict()),
         state=dict(choices=['absent', 'present'], default='present'),
     )
-    module = AnsibleAWSModule(argument_spec=argument_spec, check_boto3=False)
+    mutually_exclusive = [
+        ('instance_id', 'snapshot_id', 'volume_id'),
+    ]
+    required_if = [
+        ('state', 'absent', ('snapshot_id',)),
+    ]
+    required_one_of = [
+        ('instance_id', 'snapshot_id', 'volume_id'),
+    ]
+    required_together = [
+        ('instance_id', 'device_name'),
+    ]
+
+    module = AnsibleAWSModule(
+        argument_spec=argument_spec,
+        mutually_exclusive=mutually_exclusive,
+        required_if=required_if,
+        required_one_of=required_one_of,
+        required_together=required_together,
+    )
+
     return module
 
 
 def main():
     module = create_snapshot_ansible_module()
 
-    if not HAS_BOTO:
-        module.fail_json(msg='boto required for this module')
-
     volume_id = module.params.get('volume_id')
     snapshot_id = module.params.get('snapshot_id')
     description = module.params.get('description')
@@ -300,22 +382,28 @@ def main():
     snapshot_tags = module.params.get('snapshot_tags')
     state = module.params.get('state')
 
-    ec2 = ec2_connect(module)
-
-    create_snapshot(
-        module=module,
-        state=state,
-        description=description,
-        wait=wait,
-        wait_timeout=wait_timeout,
-        ec2=ec2,
-        volume_id=volume_id,
-        instance_id=instance_id,
-        snapshot_id=snapshot_id,
-        device_name=device_name,
-        snapshot_tags=snapshot_tags,
-        last_snapshot_min_age=last_snapshot_min_age
-    )
+    ec2 = module.client('ec2', retry_decorator=AWSRetry.jittered_backoff(retries=10))
+
+    if state == 'absent':
+        delete_snapshot(
+            module=module,
+            ec2=ec2,
+            snapshot_id=snapshot_id,
+        )
+    else:
+        create_snapshot(
+            module=module,
+            description=description,
+            wait=wait,
+            wait_timeout=wait_timeout,
+            ec2=ec2,
+            volume_id=volume_id,
+            instance_id=instance_id,
+            snapshot_id=snapshot_id,
+            device_name=device_name,
+            snapshot_tags=snapshot_tags,
+            last_snapshot_min_age=last_snapshot_min_age,
+        )
 
 
 if __name__ == '__main__':
diff --git a/tests/integration/targets/ec2_snapshot/tasks/main.yml b/tests/integration/targets/ec2_snapshot/tasks/main.yml
index c8c8ed05acd..fd00216bc92 100644
--- a/tests/integration/targets/ec2_snapshot/tasks/main.yml
+++ b/tests/integration/targets/ec2_snapshot/tasks/main.yml
@@ -178,11 +178,6 @@
 #        that:
 #          - result is changed
 
-    #  Wait at least 15 seconds between concurrent volume snapshots.
-    - name: Prevent SnapshotCreationPerVolumeRateExceeded errors
-      pause:
-        seconds: 15
-
     - name: Take snapshot and tag it
       ec2_snapshot:
         volume_id: '{{ volume_id }}'
@@ -217,12 +212,28 @@
         that:
           - info_result.snapshots| length == 3
 
+    - ec2_snapshot:
+        volume_id: '{{ volume_id }}'
+        snapshot_tags:
+          ResourcePrefix: '{{ resource_prefix }}'
+      loop: '{{ range(1, 6, 1) | list }}'
+      loop_control:
+        # Anything under 15 will trigger SnapshotCreationPerVolumeRateExceeded,
+        # this should now be automatically handled, but pause a little anyway to
+        # avoid being aggressive
+        pause: 10
+        label: "Generate extra snapshots - {{ item }}"
+
+    - name: Pause to allow creation to finish
+      pause:
+        minutes: 2
+
     # check that snapshot_ids and max_results are mutually exclusive
     - name: Check that max_results and snapshot_ids are mutually exclusive
       ec2_snapshot_info:
         snapshot_ids:
           - '{{ tagged_snapshot_id }}'
-        max_results: 1
+        max_results: 5
       ignore_errors: true
       register: info_result
 
@@ -250,12 +261,12 @@
       ec2_snapshot_info:
         filters:
           "tag:Name": '{{ resource_prefix }}'
-        max_results: 1
+        max_results: 5
       register: info_result
 
     - assert:
         that:
-          - info_result.snapshots | length == 1
+          - info_result.snapshots | length == 5
           - info_result.next_token_id is defined
 
     # Pagination : 2nd request
@@ -268,8 +279,7 @@
 
     - assert:
         that:
-          - info_result.snapshots | length == 2
-          - info_result.next_token_id is defined
+          - info_result.snapshots | length == 3
 
     # delete the tagged snapshot
     - name: Delete the tagged snapshot
@@ -285,7 +295,7 @@
 
     - assert:
         that:
-          - info_result.snapshots| length == 2
+          - info_result.snapshots| length == 7
           - '"{{ tagged_snapshot_id }}" not in "{{ info_result| community.general.json_query("snapshots[].snapshot_id") }}"'
 
     - name: Delete snapshots