diff --git a/lib/ansible/config/module_defaults.yml b/lib/ansible/config/module_defaults.yml
index 953b6e08c5c..a47a967c840 100644
--- a/lib/ansible/config/module_defaults.yml
+++ b/lib/ansible/config/module_defaults.yml
@@ -389,6 +389,8 @@ groupings:
   - aws
   ecs_service_info:
   - aws
+  ecs_tag:
+  - aws
   ecs_task:
   - aws
   ecs_taskdefinition:
diff --git a/lib/ansible/modules/cloud/amazon/ecs_tag.py b/lib/ansible/modules/cloud/amazon/ecs_tag.py
new file mode 100644
index 00000000000..99b9b925f7e
--- /dev/null
+++ b/lib/ansible/modules/cloud/amazon/ecs_tag.py
@@ -0,0 +1,235 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# Copyright: (c) 2019, Michael Pechner <mikey@mikey.com>
+# 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
+ANSIBLE_METADATA = {'metadata_version': '1.1',
+                    'status': ['preview'],
+                    'supported_by': 'community'}
+
+
+DOCUMENTATION = r'''
+---
+module: ecs_tag
+short_description: create and remove tags on Amazon ECS resources
+notes:
+    - none
+description:
+    - Creates, removes and lists tags for Amazon ECS resources.
+    - Resources are referenced by their cluster name.
+version_added: '2.10'
+author:
+  - Michael Pechner (@mpechner)
+requirements: [ boto3, botocore ]
+options:
+  cluster_name:
+    description:
+      - The name of the cluster whose resources we are tagging.
+    required: true
+    type: str
+  resource:
+    description:
+      - The ECS resource name.
+      - Required unless I(resource_type=cluster).
+    type: str
+  resource_type:
+    description:
+      - The type of resource.
+    default: cluster
+    choices: ['cluster', 'task', 'service', 'task_definition', 'container']
+    type: str
+  state:
+    description:
+      - Whether the tags should be present or absent on the resource.
+      - Use C(list) to interrogate the tags of an ECS resource.
+    default: present
+    choices: ['present', 'absent', 'list']
+    type: str
+  tags:
+    description:
+      - A dictionary of tags to add or remove from the resource.
+      - If the value provided for a tag is null and I(state=absent), the tag will be removed regardless of its current value.
+    type: dict
+  purge_tags:
+    description:
+      - Whether unspecified tags should be removed from the resource.
+      - Note that when combined with I(state=absent), specified tags with non-matching values are not purged.
+    type: bool
+    default: false
+extends_documentation_fragment:
+    - aws
+    - ec2
+'''
+
+EXAMPLES = r'''
+- name: Ensure tags are present on a resource
+  ecs_tag:
+    cluster_name: mycluster
+    resource_type: cluster
+    state: present
+    tags:
+      Name: ubervol
+      env: prod
+
+- name: Retrieve all tags on a cluster
+  ecs_tag:
+    cluster_name: mycluster
+    resource: http_task
+    resource_type: task
+    state: list
+
+- name: Remove the Env tag
+  ecs_tag:
+    cluster_name: mycluster
+    resource_type: cluster
+    tags:
+      Env:
+    state: absent
+
+- name: Remove the Env tag if it's currently 'development'
+  ecs_tag:
+    cluster_name: mycluster
+    resource_type: cluster
+    tags:
+      Env: development
+    state: absent
+
+- name: Remove all tags except for Name from a cluster
+  ecs_tag:
+    cluster_name: mycluster
+    resource_type: cluster
+    tags:
+        Name: foo
+    state: absent
+    purge_tags: true
+'''
+
+RETURN = r'''
+tags:
+  description: A dict containing the tags on the resource
+  returned: always
+  type: dict
+added_tags:
+  description: A dict of tags that were added to the resource
+  returned: If tags were added
+  type: dict
+removed_tags:
+  description: A dict of tags that were removed from the resource
+  returned: If tags were removed
+  type: dict
+'''
+
+from ansible.module_utils.aws.core import AnsibleAWSModule
+from ansible.module_utils.ec2 import boto3_tag_list_to_ansible_dict, ansible_dict_to_boto3_tag_list, compare_aws_tags
+
+try:
+    from botocore.exceptions import BotoCoreError, ClientError
+except ImportError:
+    pass    # Handled by AnsibleAWSModule
+__metaclass__ = type
+
+
+def get_tags(ecs, module, resource):
+    try:
+        return boto3_tag_list_to_ansible_dict(ecs.list_tags_for_resource(resourceArn=resource)['tags'])
+    except (BotoCoreError, ClientError) as e:
+        module.fail_json_aws(e, msg='Failed to fetch tags for resource {0}'.format(resource))
+
+
+def get_arn(ecs, module, cluster_name, resource_type, resource):
+
+    try:
+        if resource_type == 'cluster':
+            description = ecs.describe_clusters(clusters=[resource])
+            resource_arn = description['clusters'][0]['clusterArn']
+        elif resource_type == 'task':
+            description = ecs.describe_tasks(cluster=cluster_name, tasks=[resource])
+            resource_arn = description['tasks'][0]['taskArn']
+        elif resource_type == 'service':
+            description = ecs.describe_services(cluster=cluster_name, services=[resource])
+            resource_arn = description['services'][0]['serviceArn']
+        elif resource_type == 'task_definition':
+            description = ecs.describe_task_definition(taskDefinition=resource)
+            resource_arn = description['taskDefinition']['taskDefinitionArn']
+        elif resource_type == 'container':
+            description = ecs.describe_container_instances(clusters=[resource])
+            resource_arn = description['containerInstances'][0]['containerInstanceArn']
+    except (IndexError, KeyError):
+        module.fail_json(msg='Failed to find {0} {1}'.format(resource_type, resource))
+    except (BotoCoreError, ClientError) as e:
+        module.fail_json_aws(e, msg='Failed to find {0} {1}'.format(resource_type, resource))
+
+    return resource_arn
+
+
+def main():
+    argument_spec = dict(
+        cluster_name=dict(required=True),
+        resource=dict(required=False),
+        tags=dict(type='dict'),
+        purge_tags=dict(type='bool', default=False),
+        state=dict(default='present', choices=['present', 'absent', 'list']),
+        resource_type=dict(default='cluster', choices=['cluster', 'task', 'service', 'task_definition', 'container'])
+    )
+    required_if = [('state', 'present', ['tags']), ('state', 'absent', ['tags'])]
+
+    module = AnsibleAWSModule(argument_spec=argument_spec, required_if=required_if, supports_check_mode=True)
+
+    resource_type = module.params['resource_type']
+    cluster_name = module.params['cluster_name']
+    if resource_type == 'cluster':
+        resource = cluster_name
+    else:
+        resource = module.params['resource']
+    tags = module.params['tags']
+    state = module.params['state']
+    purge_tags = module.params['purge_tags']
+
+    result = {'changed': False}
+
+    ecs = module.client('ecs')
+
+    resource_arn = get_arn(ecs, module, cluster_name, resource_type, resource)
+
+    current_tags = get_tags(ecs, module, resource_arn)
+
+    if state == 'list':
+        module.exit_json(changed=False, tags=current_tags)
+
+    add_tags, remove = compare_aws_tags(current_tags, tags, purge_tags=purge_tags)
+
+    remove_tags = {}
+    if state == 'absent':
+        for key in tags:
+            if key in current_tags and (tags[key] is None or current_tags[key] == tags[key]):
+                remove_tags[key] = current_tags[key]
+
+    for key in remove:
+        remove_tags[key] = current_tags[key]
+
+    if remove_tags:
+        result['changed'] = True
+        result['removed_tags'] = remove_tags
+        if not module.check_mode:
+            try:
+                ecs.untag_resource(resourceArn=resource_arn, tagKeys=list(remove_tags.keys()))
+            except (BotoCoreError, ClientError) as e:
+                module.fail_json_aws(e, msg='Failed to remove tags {0} from resource {1}'.format(remove_tags, resource))
+
+    if state == 'present' and add_tags:
+        result['changed'] = True
+        result['added_tags'] = add_tags
+        current_tags.update(add_tags)
+        if not module.check_mode:
+            try:
+                tags = ansible_dict_to_boto3_tag_list(add_tags, tag_name_key_name='key', tag_value_key_name='value')
+                ecs.tag_resource(resourceArn=resource_arn, tags=tags)
+            except (BotoCoreError, ClientError) as e:
+                module.fail_json_aws(e, msg='Failed to set tags {0} on resource {1}'.format(add_tags, resource))
+
+    result['tags'] = get_tags(ecs, module, resource_arn)
+    module.exit_json(**result)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/test/integration/targets/ecs_tag/aliases b/test/integration/targets/ecs_tag/aliases
new file mode 100644
index 00000000000..fe51f28bd2e
--- /dev/null
+++ b/test/integration/targets/ecs_tag/aliases
@@ -0,0 +1,3 @@
+cloud/aws
+ecs_tag
+unsupported
diff --git a/test/integration/targets/ecs_tag/tasks/main.yml b/test/integration/targets/ecs_tag/tasks/main.yml
new file mode 100644
index 00000000000..54c601fb53c
--- /dev/null
+++ b/test/integration/targets/ecs_tag/tasks/main.yml
@@ -0,0 +1,350 @@
+- module_defaults:
+    group/aws:
+      aws_access_key: '{{ aws_access_key | default(omit) }}'
+      aws_secret_key: '{{ aws_secret_key | default(omit) }}'
+      security_token: '{{ security_token | default(omit) }}'
+      region: '{{ aws_region | default(omit) }}'
+  block:
+    - name: create ecs cluster
+      ecs_cluster:
+        name: "{{ resource_prefix }}"
+        state: present
+      register: cluster_info
+
+    - name: create ecs_taskdefinition
+      ecs_taskdefinition:
+        containers:
+          - name: my_container
+            image: ubuntu
+            memory: 128
+        family: "{{ resource_prefix }}"
+        state: present
+      register: ecs_taskdefinition_creation
+
+    # even after deleting the cluster and recreating with a different name
+    # the previous service can prevent the current service from starting
+    # while it's in a draining state. Check the service info and sleep
+    # if the service does not report as inactive.
+
+    - name: check if service is still running from a previous task
+      ecs_service_info:
+        service: "{{ resource_prefix }}"
+        cluster: "{{ resource_prefix }}"
+        details: yes
+      register: ecs_service_info_results
+
+    - name: delay if the service was not inactive
+      pause:
+        seconds: 30
+      when:
+        - ecs_service_info_results.services|length >0
+        - ecs_service_info_results.services[0]['status'] != 'INACTIVE'
+
+    - name: create ecs_service
+      ecs_service:
+        name: "{{ resource_prefix }}"
+        cluster: "{{ resource_prefix }}"
+        task_definition: "{{ resource_prefix }}"
+        desired_count: 1
+        state: present
+      register: ecs_service_creation
+
+    - name: ecs_service up
+      assert:
+        that:
+          - ecs_service_creation.changed
+
+    # Test tagging cluster resource
+
+    - name: cluster tags - list when there are none
+      ecs_tag:
+        cluster_name: "{{ resource_prefix}}"
+        resource: "{{ resource_prefix}}"
+        resource_type: cluster
+        state: list
+      register: taglist
+
+    - name: cluster tags - Should be an empty list
+      assert:
+        that:
+          - taglist.tags|list|length == 0
+          - taglist.failed == false
+          - taglist.changed == false
+
+    - name: cluster tags - Add tags to cluster
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{resource_prefix}}"
+        resource_type: cluster
+        state: present
+        tags:
+          Name: "{{ resource_prefix }}"
+          another: foobar
+      register: taglist
+
+    - name: cluster tags - tags should be there
+      assert:
+        that:
+          - taglist.changed == true
+          - taglist.added_tags.Name == "{{ resource_prefix }}"
+          - taglist.added_tags.another == "foobar"
+
+    - name: cluster tags - Add tags to cluster again
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{resource_prefix}}"
+        resource_type: cluster
+        state: present
+        tags:
+          Name: "{{ resource_prefix }}"
+          another: foobar
+      register: taglist
+
+    - name: cluster tags - No change after adding again
+      assert:
+        that:
+          - taglist.changed == false
+
+    - name: cluster tags - List tags
+      ecs_tag:
+        cluster_name: "{{ resource_prefix}}"
+        resource: "{{ resource_prefix}}"
+        resource_type: cluster
+        state: list
+      register: taglist
+
+    - name: cluster tags - should have 2 tags
+      assert:
+        that:
+          - taglist.tags|list|length == 2
+          - taglist.failed == false
+          - taglist.changed == false
+
+    - name: cluster tags - remove tag another
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{resource_prefix}}"
+        resource_type: cluster
+        state: absent
+        tags:
+          another:
+      register: taglist
+
+    - name: cluster tags - tag another should be gone
+      assert:
+        that:
+          - taglist.changed == true
+          - '"another" not in taglist.tags'
+
+    - name: cluster tags - remove tag when not present
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{resource_prefix}}"
+        resource_type: cluster
+        state: absent
+        tags:
+          temp:
+          temp_two:
+      register: taglist
+      ignore_errors: yes
+
+    - name: cluster tags - check that there was no fail, but changed is false
+      assert:
+        that:
+          - taglist.failed == false
+          - taglist.changed == false
+
+
+    - name: cluster tags - invalid cluster name
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}-foo"
+        resource: "{{resource_prefix}}-foo"
+        resource_type: cluster
+        state: absent
+        tags:
+          temp:
+          temp_two:
+      register: taglist
+      ignore_errors: yes
+
+    - name: cluster tags - Make sure invalid clustername is handled
+      assert:
+        that:
+          - taglist.failed == true
+          - taglist.changed == false
+          - 'taglist.msg is regex("Failed to find cluster ansible-test-.*-foo")'
+
+    # Test tagging service resource
+
+    - name: services tags - Add name tag
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{ecs_service_creation.service.serviceName}}"
+        resource_type: service
+        state: present
+        tags:
+          Name: "service-{{resource_prefix}}"
+      register: taglist
+
+    - name: service tag - name tags should be there
+      assert:
+        that:
+          - taglist.changed == true
+          - taglist.added_tags.Name == "service-{{ resource_prefix }}"
+          - taglist.tags.Name == "service-{{ resource_prefix }}"
+
+    - name: services tags - Add name tag again - see no change
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{ecs_service_creation.service.serviceName}}"
+        resource_type: service
+        state: present
+        tags:
+          Name: "service-{{resource_prefix}}"
+      register: taglist
+
+    - name: service tag - test adding tag twice has no effect
+      assert:
+        that:
+          - taglist.changed == false
+          - taglist.tags.Name == "service-{{ resource_prefix }}"
+
+    - name: service tags - retrieve all tags on a service
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{ecs_service_creation.service.serviceName}}"
+        resource_type: service
+        state: list
+      register: taglist
+
+    - name: services tags - should have 1 tag
+      assert:
+        that:
+          - taglist.tags|list|length == 1
+          - taglist.failed == false
+          - taglist.changed == false
+
+    - name: service tags - remove service tags
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{ecs_service_creation.service.serviceName}}"
+        resource_type: service
+        state: absent
+        tags:
+          Name:
+      register: taglist
+
+    - name: service tags - all tags gone
+      assert:
+        that:
+          - taglist.tags|list|length == 0
+          - taglist.changed == true
+          - '"Name" not in taglist.tags'
+
+
+    # Test tagging task_definition resource
+
+    - name: task_definition tags - Add name tag
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{ecs_taskdefinition_creation.taskdefinition.family}}"
+        resource_type: task_definition
+        state: present
+        tags:
+          Name: "task_definition-{{resource_prefix}}"
+      register: taglist
+
+    - name: task_definition tag - name tags should be there
+      assert:
+        that:
+          - taglist.changed == true
+          - taglist.added_tags.Name == "task_definition-{{ resource_prefix }}"
+          - taglist.tags.Name == "task_definition-{{ resource_prefix }}"
+
+    - name: task_definition tags - Add name tag again - see no change
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{ecs_taskdefinition_creation.taskdefinition.family}}"
+        resource_type: task_definition
+        state: present
+        tags:
+          Name: "task_definition-{{resource_prefix}}"
+      register: taglist
+
+    - name: task_definition tag - test adding tag twice has no effect
+      assert:
+        that:
+          - taglist.changed == false
+          - taglist.tags.Name == "task_definition-{{ resource_prefix }}"
+
+    - name: task_definition tags - retrieve all tags on a task_definition
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{ecs_taskdefinition_creation.taskdefinition.family}}"
+        resource_type: task_definition
+        state: list
+      register: taglist
+
+    - name: task_definition tags - should have 1 tag
+      assert:
+        that:
+          - taglist.tags|list|length == 1
+          - taglist.failed == false
+          - taglist.changed == false
+
+    - name: task_definition tags - remove task_definition tags
+      ecs_tag:
+        cluster_name: "{{resource_prefix}}"
+        resource: "{{ecs_taskdefinition_creation.taskdefinition.family}}"
+        resource_type: task_definition
+        state: absent
+        tags:
+          Name:
+      register: taglist
+
+    - name: task_definition tags - all tags gone
+      assert:
+        that:
+          - taglist.tags|list|length == 0
+          - taglist.changed == true
+          - '"Name" not in taglist.tags'
+
+  always:
+    - name: scale down ecs service
+      ecs_service:
+        name: "{{ resource_prefix }}"
+        cluster: "{{ resource_prefix }}"
+        task_definition: "{{ resource_prefix }}"
+        desired_count: 0
+        state: present
+      ignore_errors: yes
+
+    - name: pause to wait for scale down
+      pause:
+        seconds: 30
+
+    - name: remove ecs service
+      ecs_service:
+        name: "{{ resource_prefix }}"
+        cluster: "{{ resource_prefix }}"
+        task_definition: "{{ resource_prefix }}"
+        desired_count: 1
+        state: absent
+      ignore_errors: yes
+
+    - name: remove ecs task definition
+      ecs_taskdefinition:
+        containers:
+          - name: my_container
+            image: ubuntu
+            memory: 128
+        family: "{{ resource_prefix }}"
+        revision: "{{ ecs_taskdefinition_creation.taskdefinition.revision }}"
+        state: absent
+      ignore_errors: yes
+
+    - name: remove ecs cluster
+      ecs_cluster:
+        name: "{{ resource_prefix }}"
+        state: absent
+      ignore_errors: yes