Avoid conflicts when deployed by ArgoCD

This change introduces a black list for labels like the one already
existing for metadata.

The list contains one entry for the prefix "app.kubernetes.io". The
label "app.kubernetes.io/instance" is per default used by ArgoCD to
track resources, which causes copied Secrets to be potentially deleted
again by ArgoCD. Also labels with prefix "app.kubernetes.io" are in
general very specific to the resources in their respective namespace and
therefore shouldn't probably be automatically copied to resources in
other namespaces anyway.

In order to avoid code duplication the filtering is delegated to an
embedded function filter_dict.

Signed-off-by: Max Harmathy <[email protected]>

src/consts.py

@@ -9,4 +9,5 @@
 
 CLUSTER_SECRET_LABEL = "clustersecret.io"
 
-BLACK_LISTED_ANNOTATIONS = ["kopf.zalando.org", "kubectl.kubernetes.io"]
+BLACK_LISTED_ANNOTATIONS = ["kopf.zalando.org", "kubectl.kubernetes.io"]
+BLACK_LISTED_LABELS = ["app.kubernetes.io", ]

src/kubernetes_utils.py

@@ -8,7 +8,7 @@
 
 from os_utils import get_replace_existing, get_version
 from consts import CREATE_BY_ANNOTATION, LAST_SYNC_ANNOTATION, VERSION_ANNOTATION, BLACK_LISTED_ANNOTATIONS, \
-    CREATE_BY_AUTHOR, CLUSTER_SECRET_LABEL
+    BLACK_LISTED_LABELS, CREATE_BY_AUTHOR, CLUSTER_SECRET_LABEL
 
 
 def patch_clustersecret_status(
@@ -286,27 +286,37 @@ def create_secret_metadata(
         Kubernetes metadata object with ClusterSecret annotations.
     """
 
-    _labels = {
+    def filter_dict(
+            prefixes: List[str],
+            base: Dict[str, str],
+            source: Optional[Dict[str, str]] = None
+    ) -> Dict[str, str]:
+        """ Remove potential useless / dangerous annotations and labels"""
+        for item in base.items():
+            yield item
+        if source is not None:
+            for item in source.items():
+                key, _ = item
+                if not any(key.startswith(prefix) for prefix in prefixes):
+                    yield item
+
+    base_labels = {
         CLUSTER_SECRET_LABEL: 'true'
     }
-    _labels.update(labels or {})
-
-    _annotations = {
+    base_annotations = {
         CREATE_BY_ANNOTATION: CREATE_BY_AUTHOR,
         VERSION_ANNOTATION: get_version(),
         LAST_SYNC_ANNOTATION: datetime.now().isoformat(),
     }
-    _annotations.update(annotations or {})
-
-    # Remove potential useless / dangerous annotations
-    _annotations = {key: value for key, value in _annotations.items() if
-                    not any(key.startswith(prefix) for prefix in BLACK_LISTED_ANNOTATIONS)}
 
+    _annotations = filter_dict(BLACK_LISTED_ANNOTATIONS, base_annotations,
+                               annotations)
+    _labels = filter_dict(BLACK_LISTED_LABELS, base_labels, labels)
     return V1ObjectMeta(
         name=name,
         namespace=namespace,
-        annotations=_annotations,
-        labels=_labels,
+        annotations=dict(_annotations),
+        labels=dict(_labels),
     )