diff --git a/meta/runtime.yml b/meta/runtime.yml
index 85367613ace..41cb96e652d 100644
--- a/meta/runtime.yml
+++ b/meta/runtime.yml
@@ -65,6 +65,7 @@ action_groups:
   - ec2_ami_copy
   - ec2_asg
   - ec2_asg_info
+  - ec2_asg_scheduled_action
   - ec2_asg_lifecycle_hook
   - ec2_customer_gateway
   - ec2_customer_gateway_info
diff --git a/plugins/modules/ec2_asg_scheduled_action.py b/plugins/modules/ec2_asg_scheduled_action.py
new file mode 100644
index 00000000000..5f41dc31b05
--- /dev/null
+++ b/plugins/modules/ec2_asg_scheduled_action.py
@@ -0,0 +1,320 @@
+#!/usr/bin/python
+
+# Copyright: (c) 2021, Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+# Based off of https://github.com/mmochan/ansible-aws-ec2-asg-scheduled-actions/blob/master/library/ec2_asg_scheduled_action.py
+# (c) 2016, Mike Mochan <@mmochan>
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: ec2_asg_scheduled_action
+version_added: 2.2.0
+short_description: Create, modify and delete ASG scheduled scaling actions.
+description:
+  - The module will create a new scheduled action when I(state=present) and no given action is found.
+  - The module will update a new scheduled action when I(state=present) and the given action is found.
+  - The module will delete a new scheduled action when I(state=absent) and the given action is found.
+options:
+  autoscaling_group_name:
+    description:
+      - The name of the autoscaling group to add a scheduled action to.
+    type: str
+    required: true
+  scheduled_action_name:
+    description:
+      - The name of the scheduled action.
+    type: str
+    required: true
+  start_time:
+    description:
+      - Start time for the action.
+    type: str
+  end_time:
+    description:
+      - End time for the action.
+    type: str
+  time_zone:
+    description:
+      - Time zone to run against.
+    type: str
+  recurrence:
+    description:
+      - Cron style schedule to repeat the action on.
+      - Required when I(state=present).
+    type: str
+  min_size:
+    description:
+      - ASG min capacity.
+    type: int
+  max_size:
+    description:
+      - ASG max capacity.
+    type: int
+  desired_capacity:
+    description:
+      - ASG desired capacity.
+    type: int
+  state:
+    description:
+      - Create / update or delete scheduled action.
+    type: str
+    required: false
+    default: present
+    choices: ['present', 'absent']
+author: Mark Woolley(@marknet15)
+extends_documentation_fragment:
+- amazon.aws.aws
+- amazon.aws.ec2
+'''
+
+EXAMPLES = r'''
+# Create a scheduled action for a autoscaling group.
+- name: Create a minimal scheduled action for autoscaling group
+  community.aws.ec2_asg_scheduled_action:
+    region: eu-west-1
+    autoscaling_group_name: test_asg
+    scheduled_action_name: test_scheduled_action
+    start_time: 2021 October 25 08:00 UTC
+    recurrence: 40 22 * * 1-5
+    desired_capacity: 10
+    state: present
+  register: scheduled_action
+
+- name: Create a scheduled action for autoscaling group
+  community.aws.ec2_asg_scheduled_action:
+    region: eu-west-1
+    autoscaling_group_name: test_asg
+    scheduled_action_name: test_scheduled_action
+    start_time: 2021 October 25 08:00 UTC
+    end_time: 2021 October 25 08:00 UTC
+    time_zone: Europe/London
+    recurrence: 40 22 * * 1-5
+    min_size: 10
+    max_size: 15
+    desired_capacity: 10
+    state: present
+  register: scheduled_action
+
+- name: Delete scheduled action
+  community.aws.ec2_asg_scheduled_action:
+    region: eu-west-1
+    autoscaling_group_name: test_asg
+    scheduled_action_name: test_scheduled_action
+    state: absent
+'''
+RETURN = r'''
+scheduled_action_name:
+  description: The name of the scheduled action.
+  returned: when I(state=present)
+  type: str
+  sample: test_scheduled_action
+start_time:
+  description: Start time for the action.
+  returned: when I(state=present)
+  type: str
+  sample: '2021 October 25 08:00 UTC'
+end_time:
+  description: End time for the action.
+  returned: when I(state=present)
+  type: str
+  sample: '2021 October 25 08:00 UTC'
+time_zone:
+  description: The ID of the Amazon Machine Image used by the launch configuration.
+  returned: when I(state=present)
+  type: str
+  sample: Europe/London
+recurrence:
+  description: Cron style schedule to repeat the action on.
+  returned: when I(state=present)
+  type: str
+  sample: '40 22 * * 1-5'
+min_size:
+  description: ASG min capacity.
+  returned: when I(state=present)
+  type: int
+  sample: 1
+max_size:
+  description: ASG max capacity.
+  returned: when I(state=present)
+  type: int
+  sample: 2
+desired_capacity:
+  description: ASG desired capacity.
+  returned: when I(state=present)
+  type: int
+  sample: 1
+'''
+
+try:
+    import botocore
+except ImportError:
+    pass  # caught by AnsibleAWSModule
+
+try:
+    from dateutil.parser import parse as timedate_parse
+    HAS_DATEUTIL = True
+except ImportError:
+    HAS_DATEUTIL = False
+
+from ansible_collections.amazon.aws.plugins.module_utils.core import AnsibleAWSModule
+from ansible_collections.amazon.aws.plugins.module_utils.ec2 import AWSRetry
+
+
+def format_request():
+    params = dict(
+        AutoScalingGroupName=module.params.get('autoscaling_group_name'),
+        ScheduledActionName=module.params.get('scheduled_action_name'),
+        Recurrence=module.params.get('recurrence')
+    )
+
+    # Some of these params are optional
+    if module.params.get('desired_capacity') is not None:
+        params['DesiredCapacity'] = module.params.get('desired_capacity')
+
+    if module.params.get('min_size') is not None:
+        params['MinSize'] = module.params.get('min_size')
+
+    if module.params.get('max_size') is not None:
+        params['MaxSize'] = module.params.get('max_size')
+
+    if module.params.get('time_zone') is not None:
+        params['TimeZone'] = module.params.get('time_zone')
+
+    if module.params.get('start_time') is not None:
+        params['StartTime'] = module.params.get('start_time')
+
+    if module.params.get('end_time') is not None:
+        params['EndTime'] = module.params.get('end_time')
+
+    return params
+
+
+def delete_scheduled_action(current_actions):
+    if current_actions == []:
+        return False
+
+    if module.check_mode:
+        return True
+
+    params = dict(
+        AutoScalingGroupName=module.params.get('autoscaling_group_name'),
+        ScheduledActionName=module.params.get('scheduled_action_name')
+    )
+
+    try:
+        client.delete_scheduled_action(aws_retry=True, **params)
+    except botocore.exceptions.ClientError as e:
+        module.fail_json(msg=str(e))
+
+    return True
+
+
+def get_scheduled_actions():
+    params = dict(
+        AutoScalingGroupName=module.params.get('autoscaling_group_name'),
+        ScheduledActionNames=[module.params.get('scheduled_action_name')]
+    )
+
+    try:
+        actions = client.describe_scheduled_actions(aws_retry=True, **params)
+    except botocore.exceptions.ClientError as e:
+        module.fail_json_aws(e)
+
+    current_actions = actions.get("ScheduledUpdateGroupActions")
+
+    return current_actions
+
+
+def put_scheduled_update_group_action(current_actions):
+    changed = False
+    changes = dict()
+    params = format_request()
+
+    if len(current_actions) < 1:
+        changed = True
+    else:
+        # To correctly detect changes convert the start_time & end_time to datetime object
+        if "StartTime" in params:
+            params["StartTime"] = timedate_parse(params["StartTime"])
+        if "EndTime" in params:
+            params["EndTime"] = timedate_parse(params["EndTime"])
+
+        for k, v in params.items():
+            if current_actions[0].get(k) != v:
+                changes[k] = v
+
+        if changes:
+            changed = True
+
+    if module.check_mode:
+        return changed
+
+    try:
+        client.put_scheduled_update_group_action(aws_retry=True, **params)
+    except botocore.exceptions.ClientError as e:
+        module.fail_json_aws(e)
+
+    return changed
+
+
+def main():
+    global module
+    global client
+
+    argument_spec = dict(
+        autoscaling_group_name=dict(required=True, type='str'),
+        scheduled_action_name=dict(required=True, type='str'),
+        start_time=dict(default=None, type='str'),
+        end_time=dict(default=None, type='str'),
+        time_zone=dict(default=None, type='str'),
+        recurrence=dict(type='str'),
+        min_size=dict(default=None, type='int'),
+        max_size=dict(default=None, type='int'),
+        desired_capacity=dict(default=None, type='int'),
+        state=dict(default='present', choices=['present', 'absent'])
+    )
+
+    module = AnsibleAWSModule(
+        argument_spec=argument_spec,
+        required_if=[['state', 'present', ['recurrence']]],
+        supports_check_mode=True
+    )
+
+    if not HAS_DATEUTIL:
+        module.fail_json(msg='dateutil is required for this module')
+
+    if not module.botocore_at_least("1.20.24"):
+        module.fail_json(msg='botocore version >= 1.20.24 is required for this module')
+
+    client = module.client('autoscaling', retry_decorator=AWSRetry.jittered_backoff())
+    current_actions = get_scheduled_actions()
+    state = module.params.get('state')
+    results = dict()
+
+    if state == 'present':
+        changed = put_scheduled_update_group_action(current_actions)
+        if not module.check_mode:
+            updated_action = get_scheduled_actions()[0]
+            results = dict(
+                scheduled_action_name=updated_action.get('ScheduledActionName'),
+                start_time=updated_action.get('StartTime'),
+                end_time=updated_action.get('EndTime'),
+                time_zone=updated_action.get('TimeZone'),
+                recurrence=updated_action.get('Recurrence'),
+                min_size=updated_action.get('MinSize'),
+                max_size=updated_action.get('MaxSize'),
+                desired_capacity=updated_action.get('DesiredCapacity')
+            )
+    else:
+        changed = delete_scheduled_action(current_actions)
+
+    results['changed'] = changed
+    module.exit_json(**results)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/test-requirements.txt b/test-requirements.txt
index d44dc6b2012..d809cdbfa75 100644
--- a/test-requirements.txt
+++ b/test-requirements.txt
@@ -2,6 +2,8 @@ botocore
 boto3
 boto
 
+python-dateutil # Used by ec2_asg_scheduled_action
+
 coverage==4.5.4
 placebo
 mock
diff --git a/tests/integration/requirements.txt b/tests/integration/requirements.txt
index 6e870975a35..70f48bcf09f 100644
--- a/tests/integration/requirements.txt
+++ b/tests/integration/requirements.txt
@@ -10,3 +10,5 @@ virtualenv
 awscli
 # Used for comparing SSH Public keys to the Amazon fingerprints
 pycrypto
+# Used by ec2_asg_scheduled_action
+python-dateutil
diff --git a/tests/integration/targets/ec2_asg_scheduled_action/aliases b/tests/integration/targets/ec2_asg_scheduled_action/aliases
new file mode 100644
index 00000000000..4ef4b2067d0
--- /dev/null
+++ b/tests/integration/targets/ec2_asg_scheduled_action/aliases
@@ -0,0 +1 @@
+cloud/aws
diff --git a/tests/integration/targets/ec2_asg_scheduled_action/defaults/main.yml b/tests/integration/targets/ec2_asg_scheduled_action/defaults/main.yml
new file mode 100644
index 00000000000..c1c7971b455
--- /dev/null
+++ b/tests/integration/targets/ec2_asg_scheduled_action/defaults/main.yml
@@ -0,0 +1,3 @@
+---
+# Amazon Linux 2 AMI 2.0.20211005.0 x86_64 HVM gp2
+ec2_ami_name: "amzn2-ami-hvm-2.0.20211005.0-x86_64-gp2"
diff --git a/tests/integration/targets/ec2_asg_scheduled_action/meta/main.yml b/tests/integration/targets/ec2_asg_scheduled_action/meta/main.yml
new file mode 100644
index 00000000000..f42fc39ed7a
--- /dev/null
+++ b/tests/integration/targets/ec2_asg_scheduled_action/meta/main.yml
@@ -0,0 +1,4 @@
+dependencies:
+  - role: setup_botocore_pip
+    vars:
+      botocore_version: "1.20.24"
diff --git a/tests/integration/targets/ec2_asg_scheduled_action/tasks/main.yml b/tests/integration/targets/ec2_asg_scheduled_action/tasks/main.yml
new file mode 100644
index 00000000000..45d84415d5a
--- /dev/null
+++ b/tests/integration/targets/ec2_asg_scheduled_action/tasks/main.yml
@@ -0,0 +1,351 @@
+---
+- name: "ec2_asg_scheduled_action integration tests"
+  collections:
+    - amazon.aws
+    - community.aws
+  module_defaults:
+    group/aws:
+      aws_access_key: "{{ aws_access_key }}"
+      aws_secret_key: "{{ aws_secret_key }}"
+      security_token: "{{ security_token | default(omit) }}"
+      region: "{{ aws_region }}"
+  block:
+    - name: Run tests in virtualenv
+      vars:
+        ansible_python_interpreter: "{{ botocore_virtualenv_interpreter }}"
+      block:
+        ## Set up the testing dependencies: VPC, subnet, security group, and launch configuration
+        - name: Find AMI to use
+          ec2_ami_info:
+            owners: "amazon"
+            filters:
+              name: "{{ ec2_ami_name }}"
+          register: ec2_amis
+
+        - name: Set ec2_ami_image fact
+          set_fact:
+            ec2_ami_image: "{{ ec2_amis.images[0].image_id }}"
+
+        - name: Create VPC for use in testing
+          ec2_vpc_net:
+            name: "{{ resource_prefix }}-vpc"
+            cidr_block: 10.55.77.0/24
+            tenancy: default
+          register: testing_vpc
+
+        - name: Create subnet for use in testing
+          ec2_vpc_subnet:
+            state: present
+            vpc_id: "{{ testing_vpc.vpc.id }}"
+            cidr: 10.55.77.0/24
+            az: "{{ aws_region }}a"
+            resource_tags:
+              Name: "{{ resource_prefix }}-subnet"
+          register: testing_subnet
+
+        - name: create a security group with the vpc created in the ec2_setup
+          ec2_group:
+            name: "{{ resource_prefix }}-sg"
+            description: a security group for ansible tests
+            vpc_id: "{{ testing_vpc.vpc.id }}"
+            rules:
+              - proto: tcp
+                from_port: 22
+                to_port: 22
+                cidr_ip: 0.0.0.0/0
+              - proto: tcp
+                from_port: 80
+                to_port: 80
+                cidr_ip: 0.0.0.0/0
+          register: sg
+
+        - name: ensure launch configs exist
+          ec2_lc:
+            name: "{{ resource_prefix }}-lc"
+            assign_public_ip: true
+            image_id: "{{ ec2_ami_image }}"
+            security_groups: "{{ sg.group_id }}"
+            instance_type: t3.micro
+
+        - name: Create ASG ready
+          ec2_asg:
+            name: "{{ resource_prefix }}-asg"
+            launch_config_name: "{{ resource_prefix }}-lc"
+            desired_capacity: 1
+            min_size: 1
+            max_size: 2
+            vpc_zone_identifier: "{{ testing_subnet.subnet.id }}"
+            state: present
+            wait_for_instances: yes
+          register: output
+
+        - assert:
+            that:
+              - "output.viable_instances == 1"
+
+        ## Create minimal basic scheduled action
+        - name: Create basic scheduled_action - check_mode
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test"
+            start_time: 2022 October 25 08:00 UTC
+            recurrence: 40 22 * * 1-5
+            desired_capacity: 2
+            state: present
+          register: scheduled_action
+          check_mode: True
+
+        - name: Check results - Create basic scheduled_action - check_mode
+          assert:
+            that:
+              - scheduled_action is successful
+              - scheduled_action is changed
+
+        - name: Create basic scheduled_action
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test"
+            start_time: 2022 October 25 08:00 UTC
+            recurrence: 40 22 * * 1-5
+            desired_capacity: 2
+            state: present
+          register: scheduled_action
+
+        - name: Check results - Create basic scheduled_action
+          assert:
+            that:
+              - scheduled_action is successful
+              - scheduled_action is changed
+              - scheduled_action.scheduled_action_name == "{{ resource_prefix }}-test"
+              - scheduled_action.desired_capacity == 2
+
+        - name: Create basic scheduled_action - idempotent
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test"
+            start_time: 2022 October 25 08:00 UTC
+            recurrence: 40 22 * * 1-5
+            desired_capacity: 2
+            state: present
+          register: scheduled_action
+
+        - name: Check results - Create advanced scheduled_action - idempotent
+          assert:
+            that:
+              - scheduled_action is successful
+              - scheduled_action is not changed
+
+        ## Update minimal basic scheduled action
+        - name: Update basic scheduled_action - check_mode
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test"
+            start_time: 2022 October 25 08:00 UTC
+            recurrence: 40 22 * * 1-5
+            desired_capacity: 3
+            min_size: 3
+            state: present
+          register: scheduled_action
+          check_mode: True
+
+        - name: Check results - Update basic scheduled_action - check_mode
+          assert:
+            that:
+              - scheduled_action is successful
+              - scheduled_action is changed
+
+        - name: Update basic scheduled_action
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test"
+            start_time: 2022 October 25 08:00 UTC
+            recurrence: 40 22 * * 1-5
+            desired_capacity: 3
+            min_size: 3
+            state: present
+          register: scheduled_action
+
+        - name: Check results - Update basic scheduled_action
+          assert:
+            that:
+              - scheduled_action is successful
+              - scheduled_action is changed
+              - scheduled_action.scheduled_action_name == "{{ resource_prefix }}-test"
+              - scheduled_action.desired_capacity == 3
+              - scheduled_action.min_size == 3
+
+        - name: Update basic scheduled_action - idempotent
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test"
+            start_time: 2022 October 25 08:00 UTC
+            recurrence: 40 22 * * 1-5
+            desired_capacity: 3
+            min_size: 3
+            state: present
+          register: scheduled_action
+
+        - name: Check results - Update advanced scheduled_action - idempotent
+          assert:
+            that:
+              - scheduled_action is successful
+              - scheduled_action is not changed
+
+        ## Create advanced scheduled action
+        - name: Create advanced scheduled_action - check_mode
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test"
+            start_time: 2022 October 25 09:00 UTC
+            end_time: 2022 October 25 10:00 UTC
+            time_zone: Europe/London
+            recurrence: 40 22 * * 1-5
+            min_size: 2
+            max_size: 5
+            desired_capacity: 2
+            state: present
+          register: advanced_scheduled_action
+          check_mode: True
+
+        - name: Check results - Create basic scheduled_action - check_mode
+          assert:
+            that:
+              - advanced_scheduled_action is successful
+              - advanced_scheduled_action is changed
+
+        - name: Create advanced scheduled_action
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test1"
+            start_time: 2022 October 25 09:00 UTC
+            end_time: 2022 October 25 10:00 UTC
+            time_zone: Europe/London
+            recurrence: 40 22 * * 1-5
+            min_size: 2
+            max_size: 5
+            desired_capacity: 2
+            state: present
+          register: advanced_scheduled_action
+
+        - name: Check results - Create advanced scheduled_action
+          assert:
+            that:
+              - advanced_scheduled_action is successful
+              - advanced_scheduled_action is changed
+              - advanced_scheduled_action.scheduled_action_name == "{{ resource_prefix }}-test1"
+              - advanced_scheduled_action.desired_capacity == 2
+              - advanced_scheduled_action.min_size == 2
+              - advanced_scheduled_action.max_size == 5
+              - advanced_scheduled_action.time_zone == "Europe/London"
+
+        - name: Create advanced scheduled_action - idempotent
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test1"
+            start_time: 2022 October 25 09:00 UTC
+            end_time: 2022 October 25 10:00 UTC
+            time_zone: Europe/London
+            recurrence: 40 22 * * 1-5
+            min_size: 2
+            max_size: 5
+            desired_capacity: 2
+            state: present
+          register: advanced_scheduled_action
+
+        - name: Check results - Create basic scheduled_action - idempotent
+          assert:
+            that:
+              - advanced_scheduled_action is successful
+              - advanced_scheduled_action is not changed
+
+        ## Delete scheduled action
+        - name: Delete scheduled_action - check_mode
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test1"
+            state: absent
+          register: scheduled_action_deletion
+          check_mode: True
+
+        - name: Delete scheduled_action - check_mode
+          assert:
+            that:
+              - scheduled_action_deletion is successful
+              - scheduled_action_deletion is changed
+
+        - name: Delete scheduled_action
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test1"
+            state: absent
+          register: scheduled_action_deletion
+
+        - name: Delete scheduled_action
+          assert:
+            that:
+              - scheduled_action_deletion is successful
+              - scheduled_action_deletion is changed
+
+        - name: Delete scheduled_action - idempotent
+          ec2_asg_scheduled_action:
+            autoscaling_group_name: "{{ resource_prefix }}-asg"
+            scheduled_action_name: "{{ resource_prefix }}-test1"
+            state: absent
+          register: scheduled_action_deletion
+
+        - name: Delete scheduled_action - idempotent
+          assert:
+            that:
+              - scheduled_action_deletion is successful
+              - scheduled_action_deletion is not changed
+
+  always:
+    - name: Remove ASG
+      ec2_asg:
+        name: "{{ resource_prefix }}-asg"
+        state: absent
+      register: removed
+      until: removed is not failed
+      ignore_errors: yes
+      retries: 10
+
+    # Remove the testing dependencies
+    - name: Remove launch configs
+      ec2_lc:
+        name: "{{ resource_prefix }}-lc"
+        state: absent
+      register: removed
+      until: removed is not failed
+      ignore_errors: yes
+      retries: 10
+
+    - name: Remove the security group
+      ec2_group:
+        name: "{{ resource_prefix }}-sg"
+        description: a security group for ansible tests
+        vpc_id: "{{ testing_vpc.vpc.id }}"
+        state: absent
+      register: removed
+      until: removed is not failed
+      ignore_errors: yes
+      retries: 10
+
+    - name: Remove the subnet
+      ec2_vpc_subnet:
+        state: absent
+        vpc_id: "{{ testing_vpc.vpc.id }}"
+        cidr: 10.55.77.0/24
+      register: removed
+      until: removed is not failed
+      ignore_errors: yes
+      retries: 10
+
+    - name: Remove the VPC
+      ec2_vpc_net:
+        name: "{{ resource_prefix }}-vpc"
+        cidr_block: 10.55.77.0/24
+        state: absent
+      register: removed
+      until: removed is not failed
+      ignore_errors: yes
+      retries: 10