diff --git a/plugins/modules/dw_cluster.py b/plugins/modules/dw_cluster.py
index 0c5e377..7cf4dfa 100644
--- a/plugins/modules/dw_cluster.py
+++ b/plugins/modules/dw_cluster.py
@@ -83,7 +83,6 @@
     description:
       - Options for activating an Azure CDW Cluster
     type: dict
-    elements: dict
     required: False
     suboptions:
       subnet:
@@ -137,6 +136,26 @@
           - Only a single instance type can be listed.
         type: list
         elements: str
+  private_cloud:
+    description:
+      - Options for activating an On Premise CDW Cluster
+    type: dict
+    required: False
+    suboptions:
+      storage_class:
+        description:
+          - The storage class for the Local Storage Operator.
+        type: str
+      db_client_certificate:
+        description:
+          - TLS client certificate contents.
+          - Used for the mutual TLS connections between the Database Catalog and the metastore database.
+        type: str
+      db_client_key:
+        description:
+          - TLS client private key contents.
+          - Used for the mutual TLS connections between the Database Catalog and the metastore database.
+        type: str
   reserved_compute_nodes:
     description:
       - Set additional number of nodes to reserve for executors and coordinators to use during autoscaling.
@@ -219,6 +238,12 @@
     aws_lb_subnets: [subnet-id-1, subnet-id-2]
     aws_worker_subnets: [subnet-id-3, subnet-id-4]
 
+# Request Cloudera on premise Cluster creation
+- cloudera.cloud.dw_cluster:
+    env_crn: crn:cdp:environments...
+    private_cloud:
+      storage_class: my-storage-class
+
 # Delete a Data Warehouse Cluster
 - cloudera.cloud.dw_cluster:
     state: absent
@@ -357,6 +382,20 @@ def __init__(self, module):
         self.az_subnet = self._get_nested_param("azure", "subnet")
         self.az_managed_identity = self._get_nested_param("azure", "managed_identity")
 
+        # On Premise nested parameters
+        self.pvc_storage_class = self._get_nested_param(
+            "private_cloud",
+            "storage_class",
+        )
+        self.pvc_db_client_cert = self._get_nested_param(
+            "private_cloud",
+            "db_client_certificate",
+        )
+        self.pvc_db_client_key = self._get_nested_param(
+            "private_cloud",
+            "db_client_key",
+        )
+
         # Initialize return values
         self.cluster = {}
         self.changed = False
@@ -481,6 +520,9 @@ def process(self):
                             reserved_compute_nodes=self.reserved_compute_nodes,
                             reserved_shared_services_nodes=self.reserved_shared_services_nodes,
                             resource_pool=self.resource_pool,
+                            pvc_storage_class=self.pvc_storage_class,
+                            pvc_db_client_cert=self.pvc_db_client_cert,
+                            pvc_db_client_key=self.pvc_db_client_key,
                             lb_ip_ranges=self.lb_ip_ranges,
                             k8s_ip_ranges=self.k8s_ip_ranges,
                         )
@@ -536,6 +578,17 @@ def main():
             ),
             aws_lb_subnets=dict(type="list", aliases=["aws_public_subnets"]),
             aws_worker_subnets=dict(type="list", aliases=["aws_private_subnets"]),
+            private_cloud=dict(
+                type="dict",
+                options=dict(
+                    storage_class=dict(type="str"),
+                    db_client_certificate=dict(type="str"),
+                    db_client_key=dict(type="str"),
+                ),
+                required_together=[
+                    ["db_client_certificate", "db_client_key"],
+                ],
+            ),
             reserved_compute_nodes=dict(type="int"),
             reserved_shared_services_nodes=dict(type="int"),
             resource_pool=dict(type="str"),