Added create_only param to ignore existing objects

plugins/module_utils/k8s/runner.py

@@ -175,6 +175,10 @@ def perform_action(svc, definition: Dict, params: Dict) -> Dict:
         if params.get("apply"):
             instance, warnings = svc.apply(resource, definition, existing)
             result["method"] = "apply"
+        elif existing and params.get("create_only", False):
+            instance = existing.to_dict()
+            result["changed"] = False
+            result["method"] = "create"
         elif not existing:
             if state == "patched":
                 result.setdefault("warnings", []).append(

plugins/modules/k8s.py

@@ -52,6 +52,11 @@
     - If set to C(yes), and I(state) is C(present), an existing object will be replaced.
     type: bool
     default: no
+  create_only:
+    description:
+    - If set to C(yes), an object will be created if it doesn't exist, no objects will be updated.
+    type: bool
+    default: no
   merge_type:
     description:
     - Whether to override the default patch merge approach with a specific type. By default, the strategic
@@ -230,6 +235,19 @@
           name: port-8000-tcp
           port: 8000
 
+- name: Create a Secret but do not update existing
+  kubernetes.core.k8s:
+    state: present
+    create_only: yes
+    definition:
+      kind: Secret
+      metadata:
+        name: passwords
+        namespace: testing
+      data:
+        admin: "{{ lookup('community.general.random_string', length=12, base64=true) }}"
+        database: "{{ lookup('community.general.random_string', length=12, base64=true) }}"
+
 - name: Remove an existing Service object
   kubernetes.core.k8s:
     state: absent
@@ -473,6 +491,7 @@ def argspec():
         default="present", choices=["present", "absent", "patched"]
     )
     argument_spec["force"] = dict(type="bool", default=False)
+    argument_spec["create_only"] = dict(type="bool", default=False)
     argument_spec["label_selectors"] = dict(type="list", elements="str")
     argument_spec["generate_name"] = dict()
     argument_spec["server_side_apply"] = dict(

tests/unit/module_utils/test_runner.py

@@ -167,6 +167,20 @@
             (definition, []),
             {"changed": True, "method": "create", "result": definition},
         ),
+        (
+            "create",
+            {"create_only": True},
+            modified_def,
+            (modified_def, []),
+            {"changed": False, "method": "create", "result": modified_def},
+        ),
+        (
+            "create",
+            {"create_only": True},
+            {},
+            (definition, []),
+            {"changed": True, "method": "create", "result": definition},
+        ),
     ],
 )
 def test_perform_action(action, params, existing, instance_warnings, expected):