diff --git a/changelogs/fragments/71-migrate-cluster-dpm.yml b/changelogs/fragments/71-migrate-cluster-dpm.yml
new file mode 100644
index 00000000..d2b4c77b
--- /dev/null
+++ b/changelogs/fragments/71-migrate-cluster-dpm.yml
@@ -0,0 +1,3 @@
+---
+minor_changes:
+  - cluster_dpm - Migrated module from community.vmware to configure DPM in a vCenter cluster
diff --git a/meta/runtime.yml b/meta/runtime.yml
index 8bb5f1ee..5d578569 100644
--- a/meta/runtime.yml
+++ b/meta/runtime.yml
@@ -4,6 +4,7 @@ action_groups:
     vmware:
         - appliance_info
         - cluster
+        - cluster_dpm
         - cluster_drs
         - cluster_vcls
         - content_template
diff --git a/plugins/modules/cluster_dpm.py b/plugins/modules/cluster_dpm.py
new file mode 100644
index 00000000..baaf17b1
--- /dev/null
+++ b/plugins/modules/cluster_dpm.py
@@ -0,0 +1,228 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2023, Ansible Cloud Team (@ansible-collections)
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = r'''
+---
+module: cluster_dpm
+short_description: Manage Distributed Power Management (DPM) on VMware vSphere clusters
+description:
+    - Manages DPM on VMware vSphere clusters.
+author:
+    - Ansible Cloud Team (@ansible-collections)
+
+options:
+    cluster:
+        description:
+            - The name of the cluster to be managed.
+        type: str
+        required: true
+        aliases: [ cluster_name ]
+    datacenter:
+        description:
+            - The name of the datacenter.
+        type: str
+        required: true
+        aliases: [ datacenter_name ]
+    enable:
+        description:
+            - Whether to enable DPM.
+        type: bool
+        default: true
+    automation_level:
+        description:
+            - Determines whether the host power state and migration recommendations generated by vSphere DPM are run
+                automatically or manually.
+            - If set to V(manual), then vCenter generates host power operation and related virtual machine
+                migration recommendations are made, but they are not automatically run.
+            - If set to V(automated), then vCenter host power operations are automatically run if
+                related virtual machine migrations can all be run automatically.
+        type: str
+        default: automated
+        choices: [ automated, manual ]
+    recommendation_priority_threshold:
+        description:
+            - Threshold for generated host power recommendations ranging from V(1) (most conservative) to V(5) (most aggressive).
+            - The power state (host power on or off) recommendations generated by the vSphere DPM feature
+                are assigned priorities that range from priority V(1) recommendations to priority V(5) recommendations.
+            - A priority V(1) recommendation is mandatory, while a priority V(5) recommendation brings only slight improvement
+        type: int
+        default: 3
+        choices: [ 1, 2, 3, 4, 5 ]
+
+extends_documentation_fragment:
+    - vmware.vmware.vmware.documentation
+'''
+
+EXAMPLES = r'''
+- name: Enable DPM
+  vmware.vmware.cluster_dpm:
+    hostname: '{{ vcenter_hostname }}'
+    username: '{{ vcenter_username }}'
+    password: '{{ vcenter_password }}'
+    datacenter_name: datacenter
+    cluster_name: cluster
+    enable: true
+  delegate_to: localhost
+
+- name: Enable DPM and generate but don't apply all recommendations
+  vmware.vmware.cluster_dpm:
+    hostname: '{{ vcenter_hostname }}'
+    username: '{{ vcenter_username }}'
+    password: '{{ vcenter_password }}'
+    datacenter_name: datacenter
+    cluster_name: cluster
+    enable: true
+    automation_level: manual
+    recommendation_priority_threshold: 5
+  delegate_to: localhost
+'''
+
+RETURN = r'''
+result:
+    description:
+        - Information about the DPM config update task, if something changed
+        - If nothing changed, an empty dictionary is returned
+    returned: On success
+    type: dict
+    sample: {
+        "result": {
+            "completion_time": "2024-07-29T15:27:37.041577+00:00",
+            "entity_name": "test-5fb1_cluster_dpm_test",
+            "error": null,
+            "result": null,
+            "state": "success"
+        }
+    }
+'''
+
+try:
+    from pyVmomi import vim, vmodl
+except ImportError:
+    pass
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.vmware.vmware.plugins.module_utils._vmware import (
+    PyVmomi,
+    vmware_argument_spec
+)
+from ansible_collections.vmware.vmware.plugins.module_utils._vmware_tasks import (
+    TaskError,
+    RunningTaskMonitor
+)
+
+from ansible.module_utils._text import to_native
+
+
+class VMwareCluster(PyVmomi):
+    def __init__(self, module):
+        super(VMwareCluster, self).__init__(module)
+
+        datacenter = self.get_datacenter_by_name(self.params.get('datacenter'), fail_on_missing=True)
+        self.cluster = self.get_cluster_by_name(self.params.get('cluster'), fail_on_missing=True, datacenter=datacenter)
+
+    @property
+    def recommendation_priority_threshold(self):
+        """
+        When applying or reading this threshold from the vCenter config, the values are reversed. So
+        for example, vCenter thinks 1 is the most aggressive when docs/UI say 5 is most aggressive.
+        We present the scale seen in the docs/UI to the user and then adjust the value here to ensure
+        vCenter behaves as intended.
+        """
+        return 6 - self.params['recommendation_priority_threshold']
+
+    def check_dpm_config_diff(self):
+        """
+        Check the active DPM configuration and determine if desired configuration is different.
+        If the current DPM configuration is undefined for some reason, the error is caught
+        and the function returns True.
+        Returns:
+            True if there is difference, else False
+        """
+        try:
+            dpm_config = self.cluster.configurationEx.dpmConfigInfo
+
+            if (dpm_config.enabled != self.params['enable'] or
+                    dpm_config.defaultDpmBehavior != self.params['automation_level'] or
+                    dpm_config.hostPowerActionRate != self.recommendation_priority_threshold):
+                return True
+
+        except AttributeError:
+            return True
+
+        return False
+
+    def __create_dpm_config_spec(self):
+        """
+        Uses the class's attributes to create a new cluster DPM config spec
+        """
+        cluster_config_spec = vim.cluster.ConfigSpecEx()
+        cluster_config_spec.dpmConfig = vim.cluster.DpmConfigInfo()
+        cluster_config_spec.dpmConfig.enabled = self.params['enable']
+        cluster_config_spec.dpmConfig.defaultDpmBehavior = self.params['automation_level']
+        cluster_config_spec.dpmConfig.hostPowerActionRate = self.recommendation_priority_threshold
+
+        return cluster_config_spec
+
+    def apply_dpm_configuration(self):
+        """
+        Apply the class's attributes as a DPM config to the cluster
+        """
+        cluster_config_spec = self.__create_dpm_config_spec()
+
+        try:
+            task = self.cluster.ReconfigureComputeResource_Task(cluster_config_spec, True)
+            _, task_result = RunningTaskMonitor(task).wait_for_completion()   # pylint: disable=disallowed-name
+        except (vmodl.RuntimeFault, vmodl.MethodFault)as vmodl_fault:
+            self.module.fail_json(msg=to_native(vmodl_fault.msg))
+        except TaskError as task_e:
+            self.module.fail_json(msg=to_native(task_e))
+        except Exception as generic_exc:
+            self.module.fail_json(msg="Failed to update cluster due to exception %s" % to_native(generic_exc))
+
+        return task_result
+
+
+def main():
+    module = AnsibleModule(
+        argument_spec={
+            **vmware_argument_spec(), **dict(
+                cluster=dict(type='str', required=True, aliases=['cluster_name']),
+                datacenter=dict(type='str', required=True, aliases=['datacenter_name']),
+                enable=dict(type='bool', default=True),
+                automation_level=dict(
+                    type='str',
+                    choices=['automated', 'manual'],
+                    default='automated'
+                ),
+                recommendation_priority_threshold=dict(type='int', choices=[1, 2, 3, 4, 5], default=3)
+            )
+        },
+        supports_check_mode=True,
+    )
+
+    result = dict(
+        changed=False,
+        result={}
+    )
+
+    cluster_dpm = VMwareCluster(module)
+
+    config_is_different = cluster_dpm.check_dpm_config_diff()
+    if config_is_different:
+        result['changed'] = True
+        if not module.check_mode:
+            result['result'] = cluster_dpm.apply_dpm_configuration()
+
+    module.exit_json(**result)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/integration/targets/vmware_cluster_dpm/defaults/main.yml b/tests/integration/targets/vmware_cluster_dpm/defaults/main.yml
new file mode 100644
index 00000000..1a8b3db5
--- /dev/null
+++ b/tests/integration/targets/vmware_cluster_dpm/defaults/main.yml
@@ -0,0 +1,7 @@
+---
+test_cluster: "{{ tiny_prefix }}_cluster_dpm_test"
+run_on_simulator: false
+
+dpm_enable: true
+dpm_automation_level: manual
+dpm_recommendation_priority_threshold: 2
diff --git a/tests/integration/targets/vmware_cluster_dpm/run.yml b/tests/integration/targets/vmware_cluster_dpm/run.yml
new file mode 100644
index 00000000..63a3f523
--- /dev/null
+++ b/tests/integration/targets/vmware_cluster_dpm/run.yml
@@ -0,0 +1,27 @@
+- hosts: localhost
+  gather_facts: no
+  collections:
+    - community.general
+
+  tasks:
+    - name: Import eco-vcenter credentials
+      ansible.builtin.include_vars:
+        file: ../../integration_config.yml
+      tags: eco-vcenter-ci
+
+    - name: Import simulator vars
+      ansible.builtin.include_vars:
+        file: vars.yml
+      tags: integration-ci
+
+    - name: Vcsim
+      ansible.builtin.import_role:
+        name: prepare_vcsim
+      tags: integration-ci
+
+    - name: Import vmware_cluster_dpm role
+      ansible.builtin.import_role:
+        name: vmware_cluster_dpm
+      tags:
+        - integration-ci
+        - eco-vcenter-ci
diff --git a/tests/integration/targets/vmware_cluster_dpm/tasks/main.yml b/tests/integration/targets/vmware_cluster_dpm/tasks/main.yml
new file mode 100644
index 00000000..8126e7b3
--- /dev/null
+++ b/tests/integration/targets/vmware_cluster_dpm/tasks/main.yml
@@ -0,0 +1,86 @@
+---
+- name: Test On Simulator
+  when: run_on_simulator
+  block:
+    - name: Set DPM Settings In Cluster
+      vmware.vmware.cluster_dpm:
+        validate_certs: false
+        hostname: "{{ vcenter_hostname }}"
+        username: "{{ vcenter_username }}"
+        password: "{{ vcenter_password }}"
+        datacenter: "{{ vcenter_datacenter }}"
+        cluster: "{{ test_cluster }}"
+        port: "{{ vcenter_port }}"
+        enable: "{{ dpm_enable }}"
+        automation_level: "{{ dpm_automation_level }}"
+        recommendation_priority_threshold: "{{ dpm_recommendation_priority_threshold }}"
+    # The simulator never seems to update its DPM settings, so there's nothing to validate here
+
+- name: Test On VCenter
+  when: not run_on_simulator
+  block:
+    - name: Import common vars
+      ansible.builtin.include_vars:
+        file: ../group_vars.yml
+    - name: Create Test Cluster
+      vmware.vmware.cluster:
+        hostname: "{{ vcenter_hostname }}"
+        username: "{{ vcenter_username }}"
+        password: "{{ vcenter_password }}"
+        datacenter: "{{ vcenter_datacenter }}"
+        validate_certs: false
+        port: "{{ vcenter_port }}"
+        cluster_name: "{{ test_cluster }}"
+    - name: Set DPM Settings In Test Cluster
+      vmware.vmware.cluster_dpm:
+        validate_certs: false
+        hostname: "{{ vcenter_hostname }}"
+        username: "{{ vcenter_username }}"
+        password: "{{ vcenter_password }}"
+        datacenter: "{{ vcenter_datacenter }}"
+        cluster: "{{ test_cluster }}"
+        port: "{{ vcenter_port }}"
+        enable: "{{ dpm_enable }}"
+        automation_level: "{{ dpm_automation_level }}"
+        recommendation_priority_threshold: "{{ dpm_recommendation_priority_threshold }}"
+      register: _out
+    # testing for idempotence because recommendation_preiority_threshold is a little counter intuitive
+    - name: Set DPM Settings In Test Cluster Again - Idempotence
+      vmware.vmware.cluster_dpm:
+        validate_certs: false
+        hostname: "{{ vcenter_hostname }}"
+        username: "{{ vcenter_username }}"
+        password: "{{ vcenter_password }}"
+        datacenter: "{{ vcenter_datacenter }}"
+        cluster: "{{ test_cluster }}"
+        port: "{{ vcenter_port }}"
+        enable: "{{ dpm_enable }}"
+        automation_level: "{{ dpm_automation_level }}"
+        recommendation_priority_threshold: "{{ dpm_recommendation_priority_threshold }}"
+      register: _out
+    - name: Check Task Result
+      ansible.builtin.assert:
+        that: _out is not changed
+    - name: Gather Cluster Settings
+      community.vmware.vmware_cluster_info:
+        validate_certs: false
+        hostname: "{{ vcenter_hostname }}"
+        username: "{{ vcenter_username }}"
+        password: "{{ vcenter_password }}"
+        datacenter: "{{ vcenter_datacenter }}"
+        cluster_name: "{{ test_cluster }}"
+        port: "{{ vcenter_port }}"
+      register: _cluster_info
+    # cluster_info doesn't return DPM config right now so we can't validate this
+
+  always:
+    - name: Destroy Test Cluster
+      vmware.vmware.cluster:
+        hostname: "{{ vcenter_hostname }}"
+        username: "{{ vcenter_username }}"
+        password: "{{ vcenter_password }}"
+        datacenter: "{{ vcenter_datacenter }}"
+        port: "{{ vcenter_port }}"
+        validate_certs: false
+        cluster_name: "{{ test_cluster }}"
+        state: absent
diff --git a/tests/integration/targets/vmware_cluster_dpm/vars.yml b/tests/integration/targets/vmware_cluster_dpm/vars.yml
new file mode 100644
index 00000000..1f06fb7b
--- /dev/null
+++ b/tests/integration/targets/vmware_cluster_dpm/vars.yml
@@ -0,0 +1,8 @@
+vcenter_hostname: "127.0.0.1"
+vcenter_username: "user"
+vcenter_password: "pass"
+vcenter_port: 8989
+vcenter_datacenter: DC0
+test_cluster: DC0_C0
+
+run_on_simulator: true