Argocd notifications for long running sync operations

I'm running Argocd 2.7.4 and I want to get an alert for sync operations that take longer than a few minutes. I tried using the example from the documentation (https://argo-cd.readthedocs.io/en/stable/operator-manual/notifications/triggers/#functions)

  when: time.Now().Sub(time.Parse(app.status.operationState.startedAt)).Minutes() >= 5

However, that only triggers after the operation has finished which is kinda useless for my usecase. Is it actually possible to create an alert like this with argocd directly or do I need to build it myself?

I suspect this is not natively supported, which means you would need to use a combination of Argo CD's API and a monitoring tool like Prometheus. The idea would be to expose Argo CD's metrics to Prometheus and then create an alert based on the duration of the sync operation.

Here is a high-level outline of the steps you can follow:

First, Expose Argo CD's Application Controller Metrics to Prometheus by configuring the argocd-metrics service.

And set up Prometheus to scrape the metrics from Argo CD, using the Prometheus configuration.

See for example "Argo CD metrics with Prometheus operator and Datadog" by Ritesh Nanda (Oct. 2022)

Then create a custom alert in Prometheus based on the duration of the sync operation.

That would involve querying the applications' status (ApplicationService): a GET request to the /api/v1/applications/{applicationName} endpoint will include information about the application's sync status in the status.sync.status field.

To expose the sync operation durations to Prometheus, you would need to create a custom exporter. This exporter would run the script to fetch the sync operation durations and expose them as a Prometheus metric.

import requests
import time
from dateutil.parser import parse
from prometheus_client import start_http_server, Gauge

# URL of the Argo CD API server
argo_cd_api_server = "http://<argocd-server>:<port>"

# Create a Gauge metric to track sync operation durations
SYNC_DURATION = Gauge('argocd_sync_duration_seconds', 'Duration of sync operations', ['application'])

def fetch_sync_durations():
    # Get a list of all applications
    response = requests.get(f"{argo_cd_api_server}/api/v1/applications")
    applications = response.json()["items"]

    for application in applications:
        # Get the status of the application
        response = requests.get(f"{argo_cd_api_server}/api/v1/applications/{application['metadata']['name']}")
        status = response.json()["status"]

        # If the application is currently syncing, calculate the duration of the sync operation
        if status["sync"]["status"] == "Syncing":
            started_at = parse(status["operationState"]["startedAt"])
            current_duration = time.time() - started_at.timestamp()
            SYNC_DURATION.labels(application=application['metadata']['name']).set(current_duration)

if __name__ == '__main__':
    # Start up the server to expose the metrics
    start_http_server(8000)
    # Fetch sync operation durations every 60 seconds
    while True:
        fetch_sync_durations()
        time.sleep(60)

That would define a Gauge metric named argocd_sync_duration_seconds, created to track the duration of sync operations.
The fetch_sync_durations function fetches the sync operation durations and sets the value of the Gauge metric for each application.
The script then starts a HTTP server to expose the metrics and runs the fetch_sync_durations function every 60 seconds.

Finally, configure an alert manager to send notifications based on the alert. You can follow the Prometheus Alertmanager documentation to set up the alert manager and configure the notification channels (e.g., Slack, email, etc.).

A bit convoluted, but that would enable a custom alert for long-running sync operations in Argo CD.