Prevent sync operation for app in auto-sync mode

incubating/argo-cd-sync/CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## [1.4.4] - 2024-03-07
+### Fixed
+Do not sync an application in auto-sync mode
+Check for application existence before anything is done
+
 ## [1.4.3] - 2024-02-22
 ### Fixed
 intercepting application not found for better error message

incubating/argo-cd-sync/argocd_sync.py

@@ -5,6 +5,8 @@
 import logging
 import time
 import sys
+import json
+import re
 
 PAGE_SIZE   = 10
 
@@ -54,7 +56,27 @@ def main():
     export_variable(CF_OUTPUT_URL_VAR, link_to_app)
 
     ingress_host = get_runtime_ingress_host()
-    execute_argocd_sync(ingress_host)
+
+    # Does the app exist
+    # if not let's wait it has been recorded
+    # but not too long in case of simple misspelling
+    is_app_real=application_exist(ingress_host)
+    count=1
+    while count <3 and is_app_real == False:
+        logging.debug("App does not exist yet %d", count)
+        time.sleep(INTERVAL)
+        count += 1
+        is_app_real=application_exist(ingress_host)
+
+    if application_exist(ingress_host) == False:
+        print(f"ERROR application {APPLICATION} does not seem to exist")
+        sys.exit(3)
+
+    if application_autosync(ingress_host) == False:
+        execute_argocd_sync(ingress_host)
+    else:
+        logging.info("Skipping synchronization as Application is in auto-sync mode")
+
     namespace = get_runtime_ns()
     health, sync = get_app_status(ingress_host)
 
@@ -253,7 +275,7 @@ def execute_argocd_sync(ingress_host):
         retries=3,
     )
     client = Client(transport=transport, fetch_schema_from_transport=False)
-    query = get_query('argocd_sync') ## gets gql query
+    query = get_query('argocd_sync')                ## gets gql query
     variables = {
         "applicationName": APPLICATION,
         "options": {
@@ -273,15 +295,73 @@ def execute_argocd_sync(ingress_host):
         print(f"ERROR: cannot sync Application {APPLICATION}")
         logging.debug("Syncing App result: %s", err)
         sys.exit(1)
-    # finally:
-    #     print("finally block")
-    #     logging.debug("Syncing App result: %s", result)
-        # if result.errors[0].message.contains("NOT_FOUND_ERROR"):
-        #     printf("Application %s does not exit")
-        #
-        # else:
-        #     # Application sync'ed properly
-        #     logging.debug("Syncing App result: %s", result)
+
+#
+# Check for application existence
+# if it does not exist, it will return 403 error
+#
+# Return True or False
+#
+def application_exist(ingress_host):
+    runtime_api_endpoint = ingress_host + '/app-proxy/api/graphql'
+    transport = RequestsHTTPTransport(
+        url=runtime_api_endpoint,
+        headers={'authorization': CF_API_KEY},
+        verify=VERIFY,
+        retries=3,
+    )
+    client = Client(transport=transport, fetch_schema_from_transport=False)
+    query = get_query('get_app_existence')           ## gets gql query
+    variables = {
+        "applicationName": APPLICATION
+    }
+    try:
+        result = client.execute(query, variable_values=variables)
+    except TransportQueryError as err:
+        data = json.loads(re.sub('\'','\"', str(err)))
+        if (data["message"] == "Forbidden") and (data["extensions"] == 403):
+            return False
+        else:
+            print(f"ERROR: cannot test Application {APPLICATION}")
+            logging.error("Existence App result: %s", err)
+            sys.exit(1)
+    except Exception as err:
+        print(f"ERROR: cannot test Application {APPLICATION}")
+        logging.error("Existence App result: %s", err)
+        sys.exit(1)
+    return True
+
+#
+# Check if app is in auto-sync mode
+#
+# Return True or False
+#
+def application_autosync(ingress_host):
+    runtime_api_endpoint = ingress_host + '/app-proxy/api/graphql'
+    transport = RequestsHTTPTransport(
+        url=runtime_api_endpoint,
+        headers={'authorization': CF_API_KEY},
+        verify=VERIFY,
+        retries=3,
+    )
+    client = Client(transport=transport, fetch_schema_from_transport=False)
+    query = get_query('get_app_autosync')           ## gets gql query
+    variables = {
+        "applicationName": APPLICATION
+    }
+    try:
+        result = client.execute(query, variable_values=variables)
+    except Exception as err:
+        print(f"ERROR: cannot get sync policy from Application {APPLICATION}")
+        logging.debug("Application Sync policy result: %s", err)
+        sys.exit(1)
+
+    logging.debug("App sync Policy: ", result['applicationProxyQuery']['spec']['syncPolicy']['automated'])
+    if result['applicationProxyQuery']['spec']['syncPolicy']['automated'] == None:
+        return False
+    else:
+        return True
+
 
 
 def export_variable(var_name, var_value):

incubating/argo-cd-sync/queries/get_app_autosync.graphql

@@ -0,0 +1,17 @@
+query appsyncstatus ($applicationName: String!) {
+
+applicationProxyQuery(
+		name: $applicationName
+  ){
+    metadata {
+      name
+    }
+    spec {
+      syncPolicy {
+        automated {
+          prune
+        }
+      }
+    }
+  }
+}

incubating/argo-cd-sync/queries/get_app_existence.graphql

@@ -0,0 +1,9 @@
+query appexistence ($applicationName: String!) {
+  applicationProxyQuery(
+		name: $applicationName
+  ){
+    metadata {
+      name
+    }
+  }
+}

incubating/argo-cd-sync/step.yaml

@@ -1,7 +1,7 @@
 kind: step-type
 metadata:
   name: argo-cd-sync
-  version: 1.4.3
+  version: 1.4.4
   isPublic: true
   description: Syncs Argo CD apps managed by our GitOps Runtimes
   sources:
@@ -120,7 +120,7 @@ spec:
         },
         "IMAGE_TAG": {
           "type": "string",
-          "default": "1.4.3",
+          "default": "1.4.4",
           "description": "OPTIONAL - To overwrite the tag to use"
         }
       }