From 8dd6784d29330a91930c8ac57891b5a300564acf Mon Sep 17 00:00:00 2001 From: blaipr Date: Mon, 26 Oct 2026 16:33:30 +0100 Subject: [PATCH] feat: notifications to webhooks, Ascender and AWX --- CHANGELOG.md | 4 + CLAUDE.md | 6 +- README.md | 2 + docs/api.md | 2 + docs/configuration.md | 24 +- docs/notifications.md | 154 ++++ docs/observability.md | 2 + src/adapters/in/http/config.rs | 4 + src/adapters/in/http/index.rs | 4 + src/adapters/in/http/mod.rs | 1 + src/adapters/in/http/notifications.rs | 132 +++ src/adapters/in/http/openapi.rs | 8 + src/adapters/in/http/routes.rs | 8 + src/adapters/out/mod.rs | 1 + src/adapters/out/notify/http.rs | 78 ++ src/adapters/out/notify/mod.rs | 1 + src/application/config.rs | 4 + src/application/mod.rs | 1 + src/application/notify.rs | 1154 +++++++++++++++++++++++++ src/application/reconcile.rs | 22 + src/config.rs | 73 +- src/domain/mod.rs | 1 + src/domain/notification.rs | 623 +++++++++++++ src/domain/run.rs | 29 + src/lib.rs | 22 + src/main.rs | 6 +- src/ports/mod.rs | 1 + src/ports/notifier.rs | 32 + src/state.rs | 17 + tests/notifications_test.rs | 300 +++++++ 30 files changed, 2710 insertions(+), 6 deletions(-) create mode 100644 docs/notifications.md create mode 100644 src/adapters/in/http/notifications.rs create mode 100644 src/adapters/out/notify/http.rs create mode 100644 src/adapters/out/notify/mod.rs create mode 100644 src/application/notify.rs create mode 100644 src/domain/notification.rs create mode 100644 src/ports/notifier.rs create mode 100644 tests/notifications_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index bdf961f..68d00d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- Notifications: a `notifications` item sends `drift_detected`, `out_of_sync`, `sync_succeeded`, `sync_failed` and `back_in_sync` to a webhook (any receiver, with bearer, basic or header auth, a body template and optional HMAC signing) or to an Ascender or AWX job template (its webhook receiver or its launch API), one message per run, queued and retried, never with a field's value; `GET /api/v1/notifications` shows each one's last send and `POST /api/v1/notifications/{id}/test` sends a test (see `docs/notifications.md`). + ### Changed - **Breaking (who runs the image):** the image no longer bundles a demo configuration or connectors (`/app/config`, `/app/connectors`); mount your configuration at `/app/config` or set `CONFIG_DIR`, and take connectors from a project such as [unified-connectors](https://github.com/OpusProjects/unified-connectors), which now holds the SDK and the `kv` example. diff --git a/CLAUDE.md b/CLAUDE.md index 5b78a15..481a084 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -45,14 +45,17 @@ src/ │ ├── desired.rs # desired-state files: parse, load a folder, validate, order │ ├── plan.rs # THE DIFF: only declared fields compared → Operation list │ ├── run.rs # SyncStatus, RunRecord, PlanSnapshot, AppStateRegistry +│ ├── notification.rs # Notification: type (webhook/ascender/awx), events, its block │ ├── yaml.rs # YAML → JSON with exact scalars ("0644" stays a string) │ ├── credential.rs, api_key.rs, ldap.rs, project.rs, sync_health.rs ├── application/ │ ├── reconcile.rs # THE LOOP: validate → list → copy → diff → apply → re-list │ ├── projects.rs # pull a checkout (bounded by timeout_seconds) +│ ├── notify.rs # event per run (detect), message, request per type, queue + worker │ ├── scripts.rs # per-run script path resolution into checkouts │ ├── credentials.rs, login.rs, config.rs (reload) -├── ports/ # connector, state_store, git, secrets, directory, config_store, venv +├── ports/ # connector, state_store, git, secrets, directory, config_store, venv, +│ # notifier (sends one built HTTP request) └── adapters/ ├── in/http/ # applications (+ state copies), connectors, projects, │ # credentials, settings, users (auth), config, login, session, @@ -61,6 +64,7 @@ src/ └── out/ ├── connectors/process.rs # ConnectorPort → script (scrubbed env, APPLICATION_CONFIG, timeout kills) ├── state/fs.rs # StateStorePort → //.yaml + ├── notify/http.rs # HttpSenderPort → reqwest (no redirects followed) ├── git/, secrets/, directory/, config/, python.rs, process_env.rs tests/ # integration tests; tests/connectors/ holds test connector scripts # (kv.py + the SDK, copied from unified-connectors) diff --git a/README.md b/README.md index 4a36c9c..f693d49 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,7 @@ network, and a change made by hand on a system shows up as drift. - **Self-describing connectors**: a script says what it manages, each type's key and what goes first - **One application, one system**: its address and credentials live in config, never in git - **Run history**: which commit was applied, when, and the result of every operation +- **Notifications**: drift, applied commits and failures to a webhook or an Ascender or AWX job - **Config as code**: one item per file, any folders; `name` is the identity - **Strict validation**: a typo in desired state or config fails before any system is touched - **Web UI and LDAP sign-in**: the same UI family and sign-in as unified-api, on its own address @@ -41,6 +42,7 @@ network, and a change made by hand on a system shows up as drift. | [How it works](docs/how-it-works.md) | The reconcile loop: pull, list, diff, apply, and what each status means | | [Connectors](docs/connectors.md) | The script contract, the Python SDK, and writing a connector | | [Configuration](docs/configuration.md) | Every YAML file field by field, and desired-state files | +| [Notifications](docs/notifications.md) | Drift, syncs and failures to a webhook or an Ascender or AWX job template | | [API](docs/api.md) | All routes with authentication and examples | | [Architecture](docs/architecture.md) | The layers, the dependency rule, and what is shared with unified-api | | [Configuration API](docs/config-api.md) | Pushing the config directory over HTTP and reloading it live | diff --git a/docs/api.md b/docs/api.md index 5513354..5574284 100644 --- a/docs/api.md +++ b/docs/api.md @@ -18,6 +18,8 @@ session also need its `X-CSRF-Token`). Every failure is `{"error": "..."}`. | `/api/v1/applications/{id}/state/{kind}` | GET | key granted the app | The system's current settings, as of its last check | | `/api/v1/search?q=` | GET | any key (its apps) | Find an object by key or value: its application, kind, whether git declares it | | `/api/v1/connectors` | GET | admin | Connectors: script, project, timeout, who uses them | +| `/api/v1/notifications` | GET | admin | Notifications: events, destination, last send | +| `/api/v1/notifications/{id}/test` | POST | admin | Send a test message now, once; answers the outcome | | `/api/v1/projects` | GET | admin | Git checkouts, their health, what uses them | | `/api/v1/projects/{id}/sync` | POST | admin | Pull a project now | | `/api/v1/credentials` | GET | admin | Credentials: where from, whether they resolve | diff --git a/docs/configuration.md b/docs/configuration.md index a5b30b3..0b9ac0b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -20,8 +20,10 @@ as startup does, and exits. │ └── desired-state.yaml ├── credentials/ │ └── cred-isilon.yaml -└── api_keys/ - └── key-pipeline.yaml +├── api_keys/ +│ └── key-pipeline.yaml +└── notifications/ + └── ntf-failures.yaml ``` | Top-level key | Holds | @@ -31,6 +33,7 @@ as startup does, and exits. | `projects` | Git repositories: desired state, and connector scripts | | `credentials` | Where each credential's secrets come from (never the values) | | `api_keys` | API keys by name and role (the values come from env vars) | +| `notifications` | What runs report, and where: webhooks, Ascender and AWX job templates | - **The key says what an item is, never the folder.** A file in `connectors/` holding `applications:` is read as an application. @@ -169,6 +172,23 @@ api_keys: `UNIFIED_GITOPS_KEY`, if set, is one more admin key. With no keys at all the API is open, which is logged loudly at startup. +## notifications + +Which events, from which applications, go where. The full reference is [notifications.md](notifications.md): + +```yaml +# notifications/ntf-ascender-drift.yaml +notifications: + - name: ntf-ascender-drift + type: ascender # webhook | ascender | awx + events: [drift_detected] + applications: [isilon-prod] # empty = every application + ascender: + url: "https://ascender.example.com" + job_template: 123 + credentials: [cred-ascender-jt123] # its token is the template's webhook key +``` + ## Desired-state files In the application's folder, one kind per file: diff --git a/docs/notifications.md b/docs/notifications.md new file mode 100644 index 0000000..53002b8 --- /dev/null +++ b/docs/notifications.md @@ -0,0 +1,154 @@ +# Notifications + +A notification tells another system what happened to an application: drift found, a commit applied, +a sync that failed. It is a configuration item like the others, under `notifications:`. Every send +goes out from this instance, so nothing has to reach in. + +- [Events](#events) +- [Types](#types) +- [The message](#the-message) +- [Ascender and AWX](#ascender-and-awx) +- [Sending, retries and what is kept](#sending-retries-and-what-is-kept) + +--- + +## Events + +Each run produces at most one event, so one change never arrives twice under two names. + +| Event | When | Fires | +| --- | --- | --- | +| `drift_detected` | Differences at a commit already applied: someone changed the system outside git | On entering drift; with `auto_sync`, every revert | +| `out_of_sync` | A new commit waits to be applied (`auto_sync` off) | On entering the state | +| `sync_succeeded` | A sync applied at least one change from git: a commit or a rollback | Every such sync | +| `sync_failed` | A check or sync could not finish, or some operations failed | On entering the state | +| `back_in_sync` | In sync again after a failure or a difference | On entering the state | + +- **Drift and commits are told apart by commit:** differences found at the commit the application + was last in sync at were made on the system; at a newer commit, they come from git. +- **With `auto_sync`, a reverted drift is one `drift_detected`** with `reverted: true`, not a + `drift_detected` plus a `sync_succeeded`. +- **History is in memory:** after a restart, the first run fires only failures and applied + changes, so a restart does not report every pending difference again. + +## Types + +The `type` names the receiver, and its settings live in the block of the same name. + +```yaml +# notifications/ntf-failures.yaml +notifications: + - name: ntf-failures + description: "Any application that fails" + type: webhook + events: [sync_failed, back_in_sync] + applications: [] # empty = every application + webhook: + url: "https://alerts.example.com/hooks/unified-gitops" + method: POST # POST (default), PUT or PATCH + credentials: [cred-alerts] + auth: bearer # bearer (default) | basic | none | {header: X-Name} + headers: + X-Source: unified-gitops + # body: '{"text": "{{application}}: {{event}}"}' # optional; default is the message + # sign: {} # optional HMAC-SHA256 of the body + timeout_seconds: 30 + retries: 3 +``` + +| `auth` | What is sent | Credential field | +| --- | --- | --- | +| `bearer`, or left out with a credential | `Authorization: Bearer ` | `token` | +| `basic` | `Authorization: Basic …` | `username`, `password` | +| `{header: X-Name}` | `X-Name: ` | `token` | +| `none` | nothing | — | + +- **`body`** is a template: `{{field}}` takes any field of [the message](#the-message), plus + `{{revision_short}}`. Text is written escaped for a JSON string (write the quotes yourself); lists, + numbers and booleans as JSON. A name the message does not have fails the configuration. +- **`sign`** adds `X-Hub-Signature-256: sha256=` over the body, from the credential field + `secret`. `header`, `prefix` and `secret_field` change those. +- **Never in the YAML:** a token, a password, or a URL that carries one. They belong in + credentials, which come from env vars or Vault. + +The `ascender` and `awx` types start a job template; see [Ascender and AWX](#ascender-and-awx). + +## The message + +The default body, and the fields a `body` template can use: + +```json +{ + "event": "drift_detected", + "notification": "ntf-ascender-drift", + "application": "portainer", + "description": "Portainer settings, environments and users", + "status": "synced", + "previous_status": "synced", + "cause": "drift", + "reverted": true, + "revision": "39d6efb30883db193fcda905e0aa3def7b5d1f30", + "applied": 1, + "changes": [ + {"kind": "user", "key": "someone.else", "op": "delete", "fields": [], "ok": true} + ], + "errors": [], + "at": "2026-10-26T15:42:10Z", + "url": "https://gitops.example.com/applications/portainer" +} +``` + +- **`changes`** names each object, its operation and the names of the fields that changed, never + their values: a value can be a secret, and the message leaves this instance. +- **`cause`** is `commit`, `drift` or `rollback`; **`reverted`** is set on `drift_detected` only. +- **`errors`**: the first five lines of a failure, for `sync_failed`. +- **`url`** links the application in the UI when `server.url` is set. + +## Ascender and AWX + +Both start a job template (or a workflow job template) in two ways. Their routes are the same today; +they are separate types so each follows its own product. + +```yaml +# notifications/ntf-ascender-drift.yaml +notifications: + - name: ntf-ascender-drift + description: "Drift report in Ascender" + type: ascender # or awx, with an awx: block + events: [drift_detected] + applications: [portainer] + ascender: + url: "https://ascender.example.com" # the controller, without /api + job_template: 123 # or workflow_job_template: 45 + trigger: webhook # webhook (default) | launch + credentials: [cred-ascender-jt123] + # extra_vars: {report: drift} # launch only +``` + +| `trigger` | Route | The credential's `token` | The playbook reads the message as | +| --- | --- | --- | --- | +| `webhook` | `/api/v2/job_templates/{id}/gitlab/` | The template's webhook key | `ascender_webhook_payload` (Ascender), `tower_webhook_payload` (both) | +| `launch` | `/api/v2/job_templates/{id}/launch/` | A user's OAuth token | `unified_gitops`, beside `extra_vars` | + +- **`webhook`** needs the template's **Webhooks** option on, with service **GitLab**: the + controller then gives it a key, compared as is, and no user is needed. Any event is accepted. +- **`launch`** needs a user allowed to run the template, and **Prompt on launch** for its variables. + Without it the controller drops `extra_vars`; the send reports that as a failure instead of + starting a job without the event. +- The last send records the job a `launch` started. + +## Sending, retries and what is kept + +- **A queue, not the run:** a run queues its event and moves on; a worker sends, eight at a time, + so a slow receiver never holds up a check or a sync. A full queue (1000) drops the newest message, + counted in the metrics. +- **Retries:** an unanswered send, a `429` or a `5xx` is retried `retries` times, waiting 2, 4, 8 … + up to 60 seconds. A `4xx` is not: a wrong key or template does not fix itself. +- **Redirects are not followed:** the body and its credential go only where the configuration says. +- **What is kept:** each notification's last send (event, application, result, status, job, + error), in memory, on `GET /api/v1/notifications`. `POST /api/v1/notifications/{id}/test` + sends a test message now, once, and answers the outcome. +- **Metrics:** `unified_gitops_notifications_total{notification,event,outcome}` (`success`, + `error`, `dropped`) and `unified_gitops_notification_duration_seconds`. +- **Reload:** notifications reload like applications. A message keeps the definition it was queued + with. diff --git a/docs/observability.md b/docs/observability.md index ce98f36..3a3f095 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -27,6 +27,8 @@ stops running pushes nothing, so only a scrape-time gauge shows the silence. | `unified_gitops_project_sync_last_success_age_seconds` | project | Seconds since the last good pull | | `unified_gitops_connector_calls_total` | application, connector, command, outcome | Calls to the systems: `list` reads, `apply` writes, `describe` starts a run | | `unified_gitops_connector_call_duration_seconds` | application, connector, command | How long each call to a system took (histogram) | +| `unified_gitops_notifications_total` | notification, event, outcome | Messages sent (`success`), failed after retries (`error`), or dropped by a full queue | +| `unified_gitops_notification_duration_seconds` | notification | How long a send took, retries included (histogram) | | `unified_gitops_http_requests_total` | method, path, status | Requests, by matched route | | `unified_gitops_http_request_duration_seconds` | method, path | Latency histogram | | `unified_gitops_config_generation` | | Bumped by every applied reload | diff --git a/src/adapters/in/http/config.rs b/src/adapters/in/http/config.rs index d9ca2bd..41406f0 100644 --- a/src/adapters/in/http/config.rs +++ b/src/adapters/in/http/config.rs @@ -53,6 +53,7 @@ pub struct ConfigSummary { pub credentials: usize, pub projects: usize, pub api_keys: usize, + pub notifications: usize, } #[derive(Serialize, ToSchema)] @@ -116,6 +117,7 @@ pub struct ReloadInfo { pub connectors: DeltaInfo, pub projects: DeltaInfo, pub credentials: DeltaInfo, + pub notifications: DeltaInfo, /// How many API keys are in force after the reload pub api_keys: usize, } @@ -251,6 +253,7 @@ fn summarize(cfg: &AppConfig) -> ConfigSummary { credentials: cfg.credentials.len(), projects: cfg.projects.len(), api_keys: cfg.api_keys.len(), + notifications: cfg.notifications.len(), } } @@ -263,6 +266,7 @@ fn reload_info(report: &reload_use_case::ReloadReport, api_keys: usize) -> Reloa connectors: (&report.connectors).into(), projects: (&report.projects).into(), credentials: (&report.credentials).into(), + notifications: (&report.notifications).into(), api_keys, } } diff --git a/src/adapters/in/http/index.rs b/src/adapters/in/http/index.rs index 10be83f..18d3576 100644 --- a/src/adapters/in/http/index.rs +++ b/src/adapters/in/http/index.rs @@ -30,6 +30,9 @@ pub struct ApiIndex { /// Admin-only; absent for a restricted key #[serde(skip_serializing_if = "Option::is_none")] pub connectors: Option<&'static str>, + /// Admin-only; absent for a restricted key + #[serde(skip_serializing_if = "Option::is_none")] + pub notifications: Option<&'static str>, /// Admin-only route; absent for a restricted key #[serde(skip_serializing_if = "Option::is_none")] pub projects: Option<&'static str>, @@ -71,6 +74,7 @@ pub async fn api_index( applications: "/api/v1/applications", search: "/api/v1/search", connectors: admin.then_some("/api/v1/connectors"), + notifications: admin.then_some("/api/v1/notifications"), projects: admin.then_some("/api/v1/projects"), // Same condition the config handlers enforce: the routes exist in the // router either way, and answer 403 when no store is configured. diff --git a/src/adapters/in/http/mod.rs b/src/adapters/in/http/mod.rs index 03e6ea9..1fd4394 100644 --- a/src/adapters/in/http/mod.rs +++ b/src/adapters/in/http/mod.rs @@ -12,6 +12,7 @@ pub mod html; pub mod index; pub mod login; pub mod metrics; +pub mod notifications; pub mod openapi; pub mod projects; pub mod routes; diff --git a/src/adapters/in/http/notifications.rs b/src/adapters/in/http/notifications.rs new file mode 100644 index 0000000..14425e1 --- /dev/null +++ b/src/adapters/in/http/notifications.rs @@ -0,0 +1,132 @@ +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::HeaderMap; +use serde::Serialize; +use std::sync::Arc; +use utoipa::ToSchema; + +use crate::AppState; +use crate::adapters::r#in::http::auth::AuthContext; +use crate::adapters::r#in::http::error::{ApiError, ErrorBody}; +use crate::application::notify::{self, Delivery, Job}; +use crate::domain::notification::{Event, JobTrigger, NotificationType}; + +// Notifications: which events go where, and how the last send went. +// Admin-only, like connectors: a notification can name any application, and +// its destination says where this instance reaches out to. + +#[derive(Serialize, ToSchema)] +pub struct NotificationInfo { + /// The item's identity, as in the configuration + pub name: String, + pub description: Option, + #[serde(rename = "type")] + pub notification_type: NotificationType, + pub events: Vec, + /// Applications it listens to; empty = every application + pub applications: Vec, + /// Where it sends: scheme and host only, never a path or query, which + /// can carry a receiver's secret + pub destination: String, + /// ascender and awx: the job template (or workflow job template) it starts + pub job_template: Option, + pub workflow_job_template: Option, + /// ascender and awx: through the template's webhook receiver or its launch API + pub trigger: Option, + pub credentials: Vec, + /// The last send since this process started, if any + pub last: Option, +} + +#[utoipa::path( + get, + path = "/api/v1/notifications", + tag = "Notifications", + responses( + (status = 200, description = "Every notification: its events, applications and destination, and its last send", body = Vec), + (status = 403, description = "API key is not admin", body = ErrorBody) + ) +)] +pub async fn list_notifications( + State(state): State>, + axum::Extension(auth): axum::Extension, +) -> Result>, ApiError> { + if !auth.permissions.is_admin() { + return Err(ApiError::admin_only()); + } + let config = state.config(); + let mut out: Vec = config + .notifications + .iter() + .map(|(id, n)| { + let target = n.ascender.as_ref().or(n.awx.as_ref()); + let mut applications = n.applications.clone(); + applications.sort(); + NotificationInfo { + name: id.clone(), + description: n.description.clone(), + notification_type: n.notification_type, + events: n.events.clone(), + applications, + destination: n.destination(), + job_template: target.and_then(|t| t.job_template), + workflow_job_template: target.and_then(|t| t.workflow_job_template), + trigger: target.map(|t| t.trigger), + credentials: n.credentials().to_vec(), + last: state.notifier.log.last(id), + } + }) + .collect(); + out.sort_by(|a, b| a.name.cmp(&b.name)); + Ok(Json(out)) +} + +#[utoipa::path( + post, + path = "/api/v1/notifications/{id}/test", + tag = "Notifications", + params(("id" = String, Path, description = "Notification name")), + responses( + (status = 200, description = "Sent once, now, without retrying: the outcome, also kept as the notification's last send", body = Delivery), + (status = 403, description = "API key is not admin", body = ErrorBody), + (status = 404, description = "No such notification", body = ErrorBody) + ) +)] +pub async fn test_notification( + State(state): State>, + axum::Extension(auth): axum::Extension, + headers: HeaderMap, + Path(id): Path, +) -> Result, ApiError> { + if !auth.permissions.is_admin() { + return Err(ApiError::admin_only()); + } + let config = state.config(); + let Some(notification) = config.notifications.get(&id) else { + return Err(ApiError::not_found(format!( + "notification '{id}' is not configured" + ))); + }; + let mut message = notify::test_message(&id, config.url.as_deref()); + message.insert("notification".into(), serde_json::json!(id)); + let delivery = notify::deliver( + &*state.secrets, + &*state.http, + &state.notifier.log, + Job { + notification_id: id.clone(), + notification: notification.clone(), + message, + }, + false, + ) + .await; + crate::adapters::r#in::http::audit::record( + &auth, + &headers, + "notification_test", + &id, + if delivery.ok { "success" } else { "error" }, + ); + Ok(Json(delivery)) +} diff --git a/src/adapters/in/http/openapi.rs b/src/adapters/in/http/openapi.rs index 4c80046..244fd57 100644 --- a/src/adapters/in/http/openapi.rs +++ b/src/adapters/in/http/openapi.rs @@ -40,6 +40,8 @@ impl Modify for SecurityAddon { http::applications::get_application_state, http::search::search_objects, http::connectors::list_connectors, + http::notifications::list_notifications, + http::notifications::test_notification, http::projects::list_projects, http::projects::sync_project_now, http::credentials::list_credentials, @@ -75,6 +77,11 @@ impl Modify for SecurityAddon { crate::domain::run::RunMode, crate::domain::run::RunResult, http::connectors::ConnectorInfo, + http::notifications::NotificationInfo, + crate::application::notify::Delivery, + crate::domain::notification::NotificationType, + crate::domain::notification::Event, + crate::domain::notification::JobTrigger, crate::domain::connector::ConnectorType, crate::ports::state_store::Snapshot, crate::application::search::FoundObject, @@ -113,6 +120,7 @@ impl Modify for SecurityAddon { (name = "Health", description = "Liveness and readiness probes"), (name = "Applications", description = "Each live system and its desired state in git: status, the pending plan, run history and the local copy of the system's current settings; refresh compares, sync applies"), (name = "Connectors", description = "The scripts that talk to each kind of system, and the applications using them (admin-only)"), + (name = "Notifications", description = "What runs report and where it goes: webhooks and Ascender or AWX job templates, with each one's last send (admin-only)"), (name = "Projects", description = "Git project checkouts (admin-only operational routes)"), (name = "Credentials", description = "Configured credentials: where each one's secrets come from, whether they resolve now, and what uses it (admin-only; never a secret value)"), (name = "Auth", description = "Who the caller is, and the ways in: LDAP login and API keys (read-only; declared in git)"), diff --git a/src/adapters/in/http/routes.rs b/src/adapters/in/http/routes.rs index 1003517..47c2a97 100644 --- a/src/adapters/in/http/routes.rs +++ b/src/adapters/in/http/routes.rs @@ -67,6 +67,14 @@ pub fn create_router(state: Arc, api_keys: Arc) -> Rou ) .route("/api/v1/search", get(http::search::search_objects)) .route("/api/v1/connectors", get(http::connectors::list_connectors)) + .route( + "/api/v1/notifications", + get(http::notifications::list_notifications), + ) + .route( + "/api/v1/notifications/{id}/test", + post(http::notifications::test_notification), + ) .route("/api/v1/projects", get(http::projects::list_projects)) .route( "/api/v1/projects/{id}/sync", diff --git a/src/adapters/out/mod.rs b/src/adapters/out/mod.rs index 016fc9f..e2e172a 100644 --- a/src/adapters/out/mod.rs +++ b/src/adapters/out/mod.rs @@ -4,6 +4,7 @@ pub mod config; pub mod connectors; pub mod directory; pub mod git; +pub mod notify; // Not a port implementation: the scrubbed-environment Command builder shared // by every adapter that spawns a local script. pub(crate) mod process_env; diff --git a/src/adapters/out/notify/http.rs b/src/adapters/out/notify/http.rs new file mode 100644 index 0000000..2c580b8 --- /dev/null +++ b/src/adapters/out/notify/http.rs @@ -0,0 +1,78 @@ +use crate::ports::notifier::{HttpSenderPort, OutboundRequest, OutboundResponse, SendFuture}; + +// How much of an answer is kept: a receiver's error message or a job id, not +// whatever page a misconfigured URL returns. +const KEPT_BODY_BYTES: usize = 4096; + +// HttpSenderPort over reqwest (rustls), the client Vault already uses. +pub struct ReqwestSender { + client: reqwest::Client, +} + +impl ReqwestSender { + pub fn new() -> Self { + Self { + // A redirect would resend the body, credentials included, to + // wherever the receiver points: never followed. + client: reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap_or_else(|_| reqwest::Client::new()), + } + } +} + +impl Default for ReqwestSender { + fn default() -> Self { + Self::new() + } +} + +impl HttpSenderPort for ReqwestSender { + fn send(&self, request: OutboundRequest) -> SendFuture<'_> { + Box::pin(async move { + let method = reqwest::Method::from_bytes(request.method.as_bytes()) + .map_err(|e| format!("method {}: {e}", request.method))?; + let mut builder = self + .client + .request(method, &request.url) + .timeout(request.timeout) + .body(request.body); + for (name, value) in &request.headers { + builder = builder.header(name, value); + } + // The error's text, never the request: its headers carry the credential. + let response = builder.send().await.map_err(|e| without_url(&e))?; + let status = response.status().as_u16(); + let bytes = response.bytes().await.unwrap_or_default(); + let kept = &bytes[..bytes.len().min(KEPT_BODY_BYTES)]; + Ok(OutboundResponse { + status, + body: String::from_utf8_lossy(kept).into_owned(), + }) + }) + } +} + +// reqwest's error, without the URL it carries: a webhook URL can hold its +// secret in its path or query. +fn without_url(error: &reqwest::Error) -> String { + let kind = if error.is_timeout() { + "timed out" + } else if error.is_connect() { + "could not connect" + } else { + "request failed" + }; + let mut cause = String::new(); + let mut source = std::error::Error::source(error); + while let Some(inner) = source { + cause = inner.to_string(); + source = inner.source(); + } + if cause.is_empty() { + kind.to_string() + } else { + format!("{kind}: {cause}") + } +} diff --git a/src/adapters/out/notify/mod.rs b/src/adapters/out/notify/mod.rs new file mode 100644 index 0000000..3883215 --- /dev/null +++ b/src/adapters/out/notify/mod.rs @@ -0,0 +1 @@ +pub mod http; diff --git a/src/application/config.rs b/src/application/config.rs index 01fffa9..3840981 100644 --- a/src/application/config.rs +++ b/src/application/config.rs @@ -69,6 +69,7 @@ pub struct ReloadReport { pub connectors: SectionDelta, pub projects: SectionDelta, pub credentials: SectionDelta, + pub notifications: SectionDelta, } impl ReloadReport { @@ -147,6 +148,7 @@ fn runtime_config(cfg: &AppConfig) -> RuntimeConfig { applications: cfg.applications.clone(), connectors: cfg.connectors.clone(), credentials: cfg.credentials.clone(), + notifications: cfg.notifications.clone(), projects: cfg.projects.clone(), secrets: cfg.secrets_config.clone(), metrics_require_auth: cfg.server.metrics_require_auth, @@ -176,6 +178,7 @@ fn diff( connectors: SectionDelta::between(&previous.connectors, &next.connectors), projects: SectionDelta::between(&previous.projects, &next.projects), credentials: SectionDelta::between(&previous.credentials, &next.credentials), + notifications: SectionDelta::between(&previous.notifications, &next.notifications), ..ReloadReport::default() }; @@ -185,6 +188,7 @@ fn diff( ("connectors", &report.connectors), ("projects", &report.projects), ("credentials", &report.credentials), + ("notifications", &report.notifications), ] { if !delta.is_empty() { applied.push(name.to_string()); diff --git a/src/application/mod.rs b/src/application/mod.rs index 5bc909f..f85b330 100644 --- a/src/application/mod.rs +++ b/src/application/mod.rs @@ -6,6 +6,7 @@ pub mod config; pub mod credentials; pub mod login; +pub mod notify; pub mod projects; pub mod reconcile; pub mod scripts; diff --git a/src/application/notify.rs b/src/application/notify.rs new file mode 100644 index 0000000..cde716c --- /dev/null +++ b/src/application/notify.rs @@ -0,0 +1,1154 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use base64::Engine; +use dashmap::DashMap; +use serde::Serialize; +use serde_json::{Map, Value, json}; +use tokio::sync::{Semaphore, mpsc}; +use utoipa::ToSchema; + +use crate::domain::notification::{ + Auth, AuthScheme, Event, JobTemplateTarget, JobTrigger, Notification, NotificationType, + WebhookTarget, +}; +use crate::domain::run::{Baseline, RunMode, RunRecord, RunResult, SyncStatus, Trigger, now}; +use crate::ports::notifier::{HttpSenderPort, OutboundRequest, OutboundResponse}; +use crate::ports::secrets::SecretsPort; + +// The use case "tell someone what happened to an application". +// +// After every run, `detect` compares where the application stood before it +// (its Baseline) with what the run found and did, and names at most one +// event: one message per run, never "drift" and "synced" for the same +// change. Notifications that want that event are queued; a worker sends +// them, so a slow receiver never holds up a run. + +// Why a sync changed something. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Cause { + // A commit the system did not have yet. + Commit, + // Something changed on the system outside git, at a commit already applied. + Drift, + // An older commit, applied on purpose. + Rollback, +} + +impl Cause { + pub fn as_str(self) -> &'static str { + match self { + Cause::Commit => "commit", + Cause::Drift => "drift", + Cause::Rollback => "rollback", + } + } +} + +#[derive(Debug, Clone, PartialEq)] +pub struct Detected { + pub event: Event, + pub cause: Option, + // drift_detected only: whether this same run put the system back. + pub reverted: Option, +} + +impl Detected { + fn new(event: Event, cause: Option, reverted: Option) -> Self { + Self { + event, + cause, + reverted, + } + } +} + +// The event one run produced, if any. Events that describe a state +// (out_of_sync, sync_failed, back_in_sync, an unreverted drift) fire when the +// state is entered, not on every check that finds it still there; events that +// describe something done (sync_succeeded, a reverted drift) fire every time. +// +// A baseline of Unknown (first run since the process started: history is in +// memory) fires only failures and applied changes, so a restart does not +// report every pending difference again. +pub fn detect(before: &Baseline, record: &RunRecord, status: SyncStatus) -> Option { + let at_synced_commit = record.revision.is_some() && record.revision == before.synced_revision; + + if matches!(record.result, RunResult::Failed | RunResult::Partial) + || status == SyncStatus::Failed + { + return (before.status != SyncStatus::Failed) + .then(|| Detected::new(Event::SyncFailed, None, None)); + } + + let applied = record.applied.iter().filter(|a| a.ok).count(); + if record.mode == RunMode::Apply && applied > 0 { + let cause = if record.trigger == Trigger::Rollback { + Cause::Rollback + } else if at_synced_commit { + Cause::Drift + } else { + Cause::Commit + }; + return Some(match cause { + Cause::Drift => Detected::new(Event::DriftDetected, Some(cause), Some(true)), + _ => Detected::new(Event::SyncSucceeded, Some(cause), None), + }); + } + + match status { + SyncStatus::OutOfSync if at_synced_commit => (before.status == SyncStatus::Synced) + .then(|| Detected::new(Event::DriftDetected, Some(Cause::Drift), Some(false))), + SyncStatus::OutOfSync => matches!(before.status, SyncStatus::Synced | SyncStatus::Failed) + .then(|| Detected::new(Event::OutOfSync, Some(Cause::Commit), None)), + SyncStatus::Synced => matches!(before.status, SyncStatus::Failed | SyncStatus::OutOfSync) + .then(|| Detected::new(Event::BackInSync, None, None)), + _ => None, + } +} + +// How many error lines a message carries, and how long each may be: enough +// to say why, not a script's whole stderr. +const MESSAGE_ERRORS: usize = 5; +const MESSAGE_ERROR_CHARS: usize = 500; + +pub struct RunOutcome<'a> { + pub application: &'a str, + pub description: Option<&'a str>, + // server.url, for a link to the application in the UI. + pub base_url: Option<&'a str>, + pub before: &'a Baseline, + pub status: SyncStatus, + pub record: &'a RunRecord, + pub detected: &'a Detected, +} + +// The default message: what happened, to which application, at which commit, +// and which objects changed (kind, key, operation and the NAMES of changed +// fields). Never a field's value: a value can be a secret, and the message +// leaves this process. +pub fn message(o: &RunOutcome<'_>) -> Map { + let record = o.record; + let fields_of = |kind: &str, key: &str| -> Vec { + record + .operations + .iter() + .find(|op| op.kind == kind && op.key == key) + .map(|op| op.changes.iter().map(|c| c.field.clone()).collect()) + .unwrap_or_default() + }; + let changes: Vec = if record.mode == RunMode::Apply && !record.applied.is_empty() { + record + .applied + .iter() + .map(|a| { + json!({ + "kind": a.kind, + "key": a.key, + "op": a.op.as_str(), + "fields": fields_of(&a.kind, &a.key), + "ok": a.ok, + }) + }) + .collect() + } else { + record + .operations + .iter() + .map(|op| { + json!({ + "kind": op.kind, + "key": op.key, + "op": op.op.as_str(), + "fields": op.changes.iter().map(|c| c.field.clone()).collect::>(), + }) + }) + .collect() + }; + let errors: Vec = record + .errors + .iter() + .take(MESSAGE_ERRORS) + .map(|e| e.chars().take(MESSAGE_ERROR_CHARS).collect()) + .collect(); + let mut m = Map::new(); + m.insert("event".into(), json!(o.detected.event.as_str())); + m.insert("application".into(), json!(o.application)); + m.insert("description".into(), json!(o.description)); + m.insert("status".into(), json!(o.status)); + m.insert("previous_status".into(), json!(o.before.status)); + m.insert("cause".into(), json!(o.detected.cause.map(Cause::as_str))); + m.insert("reverted".into(), json!(o.detected.reverted)); + m.insert("revision".into(), json!(record.revision)); + m.insert( + "applied".into(), + json!(record.applied.iter().filter(|a| a.ok).count()), + ); + m.insert("changes".into(), Value::Array(changes)); + m.insert("errors".into(), json!(errors)); + m.insert("at".into(), json!(rfc3339(record.finished_at))); + m.insert( + "url".into(), + json!( + o.base_url + .map(|base| format!("{base}/applications/{}", o.application)) + ), + ); + m +} + +// The message a "Send test" sends: the real shape, marked as a test. +pub fn test_message(notification_id: &str, base_url: Option<&str>) -> Map { + let mut m = Map::new(); + m.insert("event".into(), json!("test")); + m.insert("application".into(), Value::Null); + m.insert( + "description".into(), + json!(format!("Test of notification {notification_id}")), + ); + for key in [ + "status", + "previous_status", + "cause", + "reverted", + "revision", + "url", + ] { + m.insert(key.into(), Value::Null); + } + m.insert("applied".into(), json!(0)); + m.insert("changes".into(), json!([])); + m.insert("errors".into(), json!([])); + m.insert("at".into(), json!(rfc3339(now()))); + if let Some(base) = base_url { + m.insert("url".into(), json!(base)); + } + m +} + +fn rfc3339(unix_seconds: u64) -> String { + chrono::DateTime::from_timestamp(unix_seconds as i64, 0) + .map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)) + .unwrap_or_default() +} + +// Fills {{field}} placeholders from the message. Text is written escaped for +// a JSON string (the template writes the quotes); lists, numbers and +// booleans as JSON; an absent value as nothing. +pub fn render(template: &str, message: &Map) -> String { + let mut out = String::with_capacity(template.len()); + let mut rest = template; + while let Some(start) = rest.find("{{") { + let Some(len) = rest[start + 2..].find("}}") else { + break; + }; + out.push_str(&rest[..start]); + let name = rest[start + 2..start + 2 + len].trim(); + out.push_str(&placeholder_value(name, message)); + rest = &rest[start + 2 + len + 2..]; + } + out.push_str(rest); + out +} + +fn placeholder_value(name: &str, message: &Map) -> String { + let value = if name == "revision_short" { + message + .get("revision") + .and_then(Value::as_str) + .map(|r| Value::String(r.chars().take(7).collect())) + .unwrap_or(Value::Null) + } else { + message.get(name).cloned().unwrap_or(Value::Null) + }; + match value { + Value::Null => String::new(), + Value::String(s) => { + let quoted = Value::String(s).to_string(); + quoted[1..quoted.len() - 1].to_string() + } + other => other.to_string(), + } +} + +// The request one notification sends for one message. Pure: the credentials +// arrive resolved, and nothing here touches the network. +pub fn build_request( + notification: &Notification, + message: &Map, + credentials: &HashMap, +) -> Result { + match notification.notification_type { + NotificationType::Webhook => { + let target = notification + .webhook + .as_ref() + .ok_or("type webhook needs a 'webhook:' block")?; + webhook_request(target, message, credentials) + } + NotificationType::Ascender | NotificationType::Awx => { + let target = notification + .ascender + .as_ref() + .or(notification.awx.as_ref()) + .ok_or("a job template notification needs its block")?; + job_template_request(target, message, credentials) + } + } +} + +fn base_headers(event: &str) -> Vec<(String, String)> { + vec![ + ("Content-Type".into(), "application/json".into()), + ( + "User-Agent".into(), + format!("unified-gitops/{}", env!("CARGO_PKG_VERSION")), + ), + ("X-Unified-Gitops-Event".into(), event.to_string()), + ] +} + +fn field<'a>(credentials: &'a HashMap, name: &str) -> Result<&'a str, String> { + credentials + .get(name) + .map(String::as_str) + .filter(|v| !v.is_empty()) + .ok_or_else(|| format!("the credential has no '{name}' field")) +} + +fn webhook_request( + target: &WebhookTarget, + message: &Map, + credentials: &HashMap, +) -> Result { + let body = match &target.body { + Some(template) => render(template, message).into_bytes(), + None => serde_json::to_vec(message).map_err(|e| e.to_string())?, + }; + let event = message.get("event").and_then(Value::as_str).unwrap_or(""); + let mut headers = base_headers(event); + for (name, value) in &target.headers { + headers.retain(|(n, _)| !n.eq_ignore_ascii_case(name)); + headers.push((name.clone(), value.clone())); + } + match &target.auth { + Auth::Scheme(AuthScheme::Bearer) => headers.push(( + "Authorization".into(), + format!("Bearer {}", field(credentials, "token")?), + )), + Auth::Default => { + if let Some(token) = credentials.get("token").filter(|t| !t.is_empty()) { + headers.push(("Authorization".into(), format!("Bearer {token}"))); + } + } + Auth::Scheme(AuthScheme::Basic) => { + let pair = format!( + "{}:{}", + field(credentials, "username")?, + field(credentials, "password")? + ); + headers.push(( + "Authorization".into(), + format!( + "Basic {}", + base64::engine::general_purpose::STANDARD.encode(pair) + ), + )); + } + Auth::Header { header } => { + headers.push((header.clone(), field(credentials, "token")?.to_string())); + } + Auth::Scheme(AuthScheme::None) => {} + } + if let Some(sign) = &target.sign { + let secret = field(credentials, &sign.secret_field)?; + let digest = hmac_sha256(secret.as_bytes(), &body); + headers.push(( + sign.header.clone(), + format!("{}{}", sign.prefix, hex(&digest)), + )); + } + Ok(OutboundRequest { + method: target.method.clone(), + url: target.url.clone(), + headers, + body, + timeout: Duration::from_secs(target.timeout_seconds), + }) +} + +// Ascender and AWX speak the same routes. `webhook` posts the message to the +// template's own receiver in GitLab mode (its key, compared as is, and any +// event name; the playbook reads the message as _webhook_payload). +// `launch` starts it as a user, the message under extra_vars.unified_gitops. +fn job_template_request( + target: &JobTemplateTarget, + message: &Map, + credentials: &HashMap, +) -> Result { + let token = field(credentials, "token")?; + let event = message.get("event").and_then(Value::as_str).unwrap_or(""); + let mut headers = base_headers(event); + let body = match target.trigger { + JobTrigger::Webhook => { + headers.push(("X-Gitlab-Token".into(), token.to_string())); + headers.push(("X-Gitlab-Event".into(), format!("unified-gitops {event}"))); + serde_json::to_vec(message).map_err(|e| e.to_string())? + } + JobTrigger::Launch => { + headers.push(("Authorization".into(), format!("Bearer {token}"))); + let mut extra_vars = target.extra_vars.clone(); + extra_vars.insert("unified_gitops".into(), Value::Object(message.clone())); + serde_json::to_vec(&json!({ "extra_vars": extra_vars })).map_err(|e| e.to_string())? + } + }; + Ok(OutboundRequest { + method: "POST".into(), + url: target.endpoint(), + headers, + body, + timeout: Duration::from_secs(target.timeout_seconds), + }) +} + +// What an answer means for this receiver: Ok with the job it started, if it +// says; Err with why it did not do what was asked. +pub fn interpret( + notification: &Notification, + response: &OutboundResponse, +) -> Result, String> { + if !(200..300).contains(&response.status) { + let excerpt: String = response.body.chars().take(300).collect(); + return Err(format!("HTTP {}: {}", response.status, excerpt.trim())); + } + let target = notification.ascender.as_ref().or(notification.awx.as_ref()); + let Some(target) = target else { + return Ok(None); + }; + let answer: Value = serde_json::from_str(&response.body).unwrap_or(Value::Null); + match target.trigger { + JobTrigger::Launch => { + // Without "Prompt on launch" for variables (or a survey), the + // controller drops extra_vars and says so only here: the job + // would run without the event. + if answer + .get("ignored_fields") + .and_then(|f| f.get("extra_vars")) + .is_some() + { + return Err( + "the job template ignored extra_vars: tick \"Prompt on launch\" \ + for its variables" + .to_string(), + ); + } + Ok(answer + .get("job") + .or_else(|| answer.get("workflow_job")) + .or_else(|| answer.get("id")) + .map(|id| id.to_string())) + } + JobTrigger::Webhook => Ok(None), + } +} + +// Worth sending again: no answer at all, or a receiver that is overloaded or +// failing. A 4xx is this side's mistake (a wrong key, a wrong template) and +// will not fix itself. +fn retryable(result: &Result) -> bool { + match result { + Err(_) => true, + Ok(response) => response.status == 429 || response.status >= 500, + } +} + +// HMAC-SHA256 (RFC 2104) over sha2, already in the tree for config ETags. +pub fn hmac_sha256(key: &[u8], message: &[u8]) -> [u8; 32] { + use sha2::{Digest, Sha256}; + const BLOCK: usize = 64; + let mut padded = [0u8; BLOCK]; + if key.len() > BLOCK { + let digest = Sha256::digest(key); + padded[..digest.len()].copy_from_slice(&digest); + } else { + padded[..key.len()].copy_from_slice(key); + } + let inner_key: Vec = padded.iter().map(|b| b ^ 0x36).collect(); + let outer_key: Vec = padded.iter().map(|b| b ^ 0x5c).collect(); + let inner = Sha256::new() + .chain_update(&inner_key) + .chain_update(message) + .finalize(); + let outer = Sha256::new() + .chain_update(&outer_key) + .chain_update(inner) + .finalize(); + let mut out = [0u8; 32]; + out.copy_from_slice(&outer); + out +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +// The last send of each notification, for the API and the UI. In memory, +// like run history. +#[derive(Debug, Clone, Serialize, ToSchema)] +pub struct Delivery { + pub event: String, + pub application: Option, + // Unix seconds. + pub at: u64, + pub ok: bool, + pub attempts: u32, + // The receiver's HTTP status, when it answered. + pub status: Option, + // The job the controller started, when it says (launch). + pub job: Option, + pub error: Option, +} + +#[derive(Default)] +pub struct DeliveryLog { + last: DashMap, +} + +impl DeliveryLog { + pub fn last(&self, notification_id: &str) -> Option { + self.last.get(notification_id).map(|d| d.clone()) + } + + fn record(&self, notification_id: &str, delivery: Delivery) { + self.last.insert(notification_id.to_string(), delivery); + } +} + +// One message for one notification, its definition as of the run that made +// it: a reload while it waits does not change where it goes. +pub struct Job { + pub notification_id: String, + pub notification: Notification, + pub message: Map, +} + +// Bounded: a receiver that is down for hours must not grow memory without +// limit. A full queue drops the new message, counted and logged. +const QUEUE: usize = 1000; +// Sends in flight at once, so one slow receiver does not hold up the others. +const CONCURRENT: usize = 8; +// Longest wait between two attempts. +const MAX_BACKOFF_SECONDS: u64 = 60; + +pub struct Dispatcher { + tx: mpsc::Sender, + // Taken by start(); None once the worker runs. + rx: std::sync::Mutex>>, + pub log: Arc, +} + +impl Default for Dispatcher { + fn default() -> Self { + Self::new() + } +} + +impl Dispatcher { + pub fn new() -> Self { + let (tx, rx) = mpsc::channel(QUEUE); + Self { + tx, + rx: std::sync::Mutex::new(Some(rx)), + log: Arc::new(DeliveryLog::default()), + } + } + + pub fn enqueue(&self, job: Job) { + if let Err(e) = self.tx.try_send(job) { + let job = match e { + mpsc::error::TrySendError::Full(job) | mpsc::error::TrySendError::Closed(job) => { + job + } + }; + let event = job + .message + .get("event") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + tracing::warn!(notification = %job.notification_id, event = %event, "notification queue full; message dropped"); + count(&job.notification_id, &event, "dropped"); + } + } + + // Starts the worker that sends what is queued. Once: later calls do + // nothing. Separate from construction so building the app needs no + // runtime; main starts it beside the scheduler, and tests that send do. + pub fn start(&self, secrets: Arc, http: Arc) { + let Some(mut rx) = self.rx.lock().unwrap_or_else(|e| e.into_inner()).take() else { + return; + }; + let log = Arc::clone(&self.log); + tokio::spawn(async move { + let slots = Arc::new(Semaphore::new(CONCURRENT)); + while let Some(job) = rx.recv().await { + let Ok(slot) = Arc::clone(&slots).acquire_owned().await else { + return; + }; + let (secrets, http, log) = + (Arc::clone(&secrets), Arc::clone(&http), Arc::clone(&log)); + tokio::spawn(async move { + let _slot = slot; + deliver(&*secrets, &*http, &log, job, true).await; + }); + } + }); + } +} + +// Sends one message, with retries when `retry` is set, records the outcome +// and returns it. The worker's path, and the "Send test" button's (which +// answers at once, without retrying). +pub async fn deliver( + secrets: &dyn SecretsPort, + http: &dyn HttpSenderPort, + log: &DeliveryLog, + job: Job, + retry: bool, +) -> Delivery { + let started = Instant::now(); + let event = job + .message + .get("event") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + let application = job + .message + .get("application") + .and_then(Value::as_str) + .map(str::to_string); + let mut delivery = Delivery { + event: event.clone(), + application, + at: now(), + ok: false, + attempts: 0, + status: None, + job: None, + error: None, + }; + + let request = match resolve(secrets, job.notification.credentials()).await { + Ok(credentials) => build_request(&job.notification, &job.message, &credentials), + Err(e) => Err(e), + }; + match request { + Err(e) => delivery.error = Some(e), + Ok(request) => { + let retries = if retry { + retries_of(&job.notification) + } else { + 0 + }; + loop { + delivery.attempts += 1; + let result = http.send(request.clone()).await; + delivery.status = result.as_ref().ok().map(|r| r.status); + let again = retryable(&result); + match result.and_then(|response| interpret(&job.notification, &response)) { + Ok(started_job) => { + delivery.ok = true; + delivery.job = started_job; + delivery.error = None; + break; + } + Err(e) => delivery.error = Some(e), + } + if !again || delivery.attempts > retries { + break; + } + let wait = (1u64 << delivery.attempts.min(6)).min(MAX_BACKOFF_SECONDS); + tokio::time::sleep(Duration::from_secs(wait)).await; + } + } + } + + let outcome = if delivery.ok { "success" } else { "error" }; + count(&job.notification_id, &event, outcome); + metrics::histogram!( + "unified_gitops_notification_duration_seconds", + "notification" => job.notification_id.clone() + ) + .record(started.elapsed().as_secs_f64()); + if delivery.ok { + tracing::info!(notification = %job.notification_id, event = %event, attempts = delivery.attempts, "notification sent"); + } else { + tracing::warn!( + notification = %job.notification_id, + event = %event, + attempts = delivery.attempts, + error = delivery.error.as_deref().unwrap_or(""), + "notification failed" + ); + } + log.record(&job.notification_id, delivery.clone()); + delivery +} + +fn retries_of(notification: &Notification) -> u32 { + notification + .webhook + .as_ref() + .map(|w| w.retries) + .or_else(|| notification.ascender.as_ref().map(|t| t.retries)) + .or_else(|| notification.awx.as_ref().map(|t| t.retries)) + .unwrap_or(0) +} + +async fn resolve( + secrets: &dyn SecretsPort, + ids: &[String], +) -> Result, String> { + crate::application::credentials::resolve_credentials(secrets, ids) + .await + .map_err(|e| e.message) +} + +fn count(notification_id: &str, event: &str, outcome: &'static str) { + metrics::counter!( + "unified_gitops_notifications_total", + "notification" => notification_id.to_string(), + "event" => event.to_string(), + "outcome" => outcome + ) + .increment(1); +} + +// After a run: names its event, if any, and queues it for every notification +// that wants it from this application. +pub fn after_run( + dispatcher: &Dispatcher, + notifications: &HashMap, + outcome: &RunOutcome<'_>, +) { + if notifications.is_empty() { + return; + } + let event = outcome.detected.event; + let mut wanted: Vec<(&String, &Notification)> = notifications + .iter() + .filter(|(_, n)| n.wants(event, outcome.application)) + .collect(); + if wanted.is_empty() { + return; + } + wanted.sort_by(|a, b| a.0.cmp(b.0)); + let base = message(outcome); + for (id, notification) in wanted { + let mut m = base.clone(); + m.insert("notification".into(), json!(id)); + dispatcher.enqueue(Job { + notification_id: id.clone(), + notification: notification.clone(), + message: m, + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::domain::plan::{Change, OpKind, Operation}; + use crate::domain::run::AppliedOperation; + + fn baseline(status: SyncStatus, synced: Option<&str>) -> Baseline { + Baseline { + status, + synced_revision: synced.map(str::to_string), + } + } + + fn record(mode: RunMode, trigger: Trigger, result: RunResult, revision: &str) -> RunRecord { + RunRecord { + id: 1, + trigger, + mode, + result, + started_at: 0, + finished_at: 1_700_000_000, + revision: Some(revision.to_string()), + planned: 0, + operations: Vec::new(), + applied: Vec::new(), + errors: Vec::new(), + } + } + + fn applied(ok: bool) -> AppliedOperation { + AppliedOperation { + kind: "user".into(), + key: "jane".into(), + op: OpKind::Update, + ok, + error: None, + } + } + + fn event(d: Option) -> Option<(Event, Option, Option)> { + d.map(|d| (d.event, d.cause, d.reverted)) + } + + #[test] + fn a_change_from_git_applied_is_sync_succeeded_with_cause_commit() { + let mut r = record(RunMode::Apply, Trigger::Schedule, RunResult::Succeeded, "b"); + r.applied.push(applied(true)); + let d = detect( + &baseline(SyncStatus::Synced, Some("a")), + &r, + SyncStatus::Synced, + ); + assert_eq!( + event(d), + Some((Event::SyncSucceeded, Some(Cause::Commit), None)) + ); + } + + #[test] + fn a_change_applied_at_the_synced_commit_is_reverted_drift_only() { + let mut r = record(RunMode::Apply, Trigger::Schedule, RunResult::Succeeded, "a"); + r.applied.push(applied(true)); + let d = detect( + &baseline(SyncStatus::Synced, Some("a")), + &r, + SyncStatus::Synced, + ); + assert_eq!( + event(d), + Some((Event::DriftDetected, Some(Cause::Drift), Some(true))) + ); + } + + #[test] + fn drift_found_without_auto_sync_fires_once() { + let r = record(RunMode::Plan, Trigger::Schedule, RunResult::OutOfSync, "a"); + let first = detect( + &baseline(SyncStatus::Synced, Some("a")), + &r, + SyncStatus::OutOfSync, + ); + assert_eq!( + event(first), + Some((Event::DriftDetected, Some(Cause::Drift), Some(false))) + ); + let again = detect( + &baseline(SyncStatus::OutOfSync, Some("a")), + &r, + SyncStatus::OutOfSync, + ); + assert_eq!(again, None, "still drifted: not a new event"); + } + + #[test] + fn a_new_commit_waiting_is_out_of_sync_once() { + let r = record(RunMode::Plan, Trigger::Schedule, RunResult::OutOfSync, "b"); + let first = detect( + &baseline(SyncStatus::Synced, Some("a")), + &r, + SyncStatus::OutOfSync, + ); + assert_eq!( + event(first), + Some((Event::OutOfSync, Some(Cause::Commit), None)) + ); + let again = detect( + &baseline(SyncStatus::OutOfSync, Some("a")), + &r, + SyncStatus::OutOfSync, + ); + assert_eq!(again, None); + } + + #[test] + fn failures_fire_on_entering_the_failed_state() { + let r = record(RunMode::Apply, Trigger::Schedule, RunResult::Failed, "a"); + let first = detect( + &baseline(SyncStatus::Synced, Some("a")), + &r, + SyncStatus::Failed, + ); + assert_eq!(event(first), Some((Event::SyncFailed, None, None))); + let again = detect( + &baseline(SyncStatus::Failed, Some("a")), + &r, + SyncStatus::Failed, + ); + assert_eq!(again, None); + let mut partial = record(RunMode::Apply, Trigger::Schedule, RunResult::Partial, "b"); + partial.applied.push(applied(true)); + partial.applied.push(applied(false)); + let d = detect( + &baseline(SyncStatus::Synced, Some("a")), + &partial, + SyncStatus::Failed, + ); + assert_eq!( + event(d), + Some((Event::SyncFailed, None, None)), + "partial is a failure, once" + ); + } + + #[test] + fn recovering_is_back_in_sync() { + let r = record(RunMode::Plan, Trigger::Schedule, RunResult::Succeeded, "a"); + let d = detect( + &baseline(SyncStatus::Failed, Some("a")), + &r, + SyncStatus::Synced, + ); + assert_eq!(event(d), Some((Event::BackInSync, None, None))); + let steady = detect( + &baseline(SyncStatus::Synced, Some("a")), + &r, + SyncStatus::Synced, + ); + assert_eq!(steady, None); + } + + #[test] + fn after_a_restart_only_failures_and_applied_changes_fire() { + let unknown = baseline(SyncStatus::Unknown, None); + let pending = record(RunMode::Plan, Trigger::Schedule, RunResult::OutOfSync, "a"); + assert_eq!(detect(&unknown, &pending, SyncStatus::OutOfSync), None); + let synced = record(RunMode::Plan, Trigger::Schedule, RunResult::Succeeded, "a"); + assert_eq!(detect(&unknown, &synced, SyncStatus::Synced), None); + let failed = record(RunMode::Plan, Trigger::Schedule, RunResult::Failed, "a"); + assert!(detect(&unknown, &failed, SyncStatus::Failed).is_some()); + } + + #[test] + fn a_rollback_applied_is_sync_succeeded_with_cause_rollback() { + let mut r = record( + RunMode::Apply, + Trigger::Rollback, + RunResult::Succeeded, + "old", + ); + r.applied.push(applied(true)); + let d = detect( + &baseline(SyncStatus::Synced, Some("new")), + &r, + SyncStatus::Synced, + ); + assert_eq!( + event(d), + Some((Event::SyncSucceeded, Some(Cause::Rollback), None)) + ); + } + + fn outcome_message() -> Map { + let mut r = record( + RunMode::Apply, + Trigger::Schedule, + RunResult::Succeeded, + "0123456789abcdef", + ); + r.operations.push(Operation { + kind: "user".into(), + key: "jane".into(), + op: OpKind::Update, + object: json!({"Username": "jane", "Password": "hunter2"}), + changes: vec![Change { + field: "Password".into(), + from: json!("old-secret"), + to: json!("hunter2"), + }], + file: None, + }); + r.applied.push(applied(true)); + let before = baseline(SyncStatus::Synced, Some("a")); + let detected = Detected::new(Event::SyncSucceeded, Some(Cause::Commit), None); + message(&RunOutcome { + application: "portainer", + description: Some("Portainer"), + base_url: Some("https://gitops.example.com"), + before: &before, + status: SyncStatus::Synced, + record: &r, + detected: &detected, + }) + } + + #[test] + fn a_message_names_changed_fields_but_never_their_values() { + let m = outcome_message(); + let text = serde_json::to_string(&m).unwrap(); + assert!( + !text.contains("hunter2") && !text.contains("old-secret"), + "{text}" + ); + assert_eq!(m["changes"][0]["fields"], json!(["Password"])); + assert_eq!(m["cause"], json!("commit")); + assert_eq!(m["applied"], json!(1)); + assert_eq!(m["at"], json!("2023-11-14T22:13:20Z")); + assert_eq!( + m["url"], + json!("https://gitops.example.com/applications/portainer") + ); + } + + #[test] + fn render_escapes_text_and_writes_json_for_lists() { + let mut m = outcome_message(); + m.insert("description".into(), json!("say \"hi\"")); + let body = render( + r#"{"d": "{{description}}", "c": {{changes}}, "r": "{{revision_short}}", "x": "{{reverted}}"}"#, + &m, + ); + let parsed: Value = serde_json::from_str(&body).expect(&body); + assert_eq!(parsed["d"], json!("say \"hi\"")); + assert_eq!(parsed["c"][0]["key"], json!("jane")); + assert_eq!(parsed["r"], json!("0123456")); + assert_eq!(parsed["x"], json!("")); + } + + fn notification(yaml: &str) -> Notification { + serde_saphyr::from_str(yaml).unwrap() + } + + fn header<'a>(request: &'a OutboundRequest, name: &str) -> Option<&'a str> { + request + .headers + .iter() + .find(|(n, _)| n.eq_ignore_ascii_case(name)) + .map(|(_, v)| v.as_str()) + } + + #[test] + fn a_webhook_puts_the_credential_where_auth_says() { + let creds = HashMap::from([("token".to_string(), "t0k".to_string())]); + let m = outcome_message(); + let bearer = notification( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com/h\n credentials: [c]\n", + ); + let request = build_request(&bearer, &m, &creds).unwrap(); + assert_eq!(header(&request, "Authorization"), Some("Bearer t0k")); + assert_eq!(header(&request, "Content-Type"), Some("application/json")); + let parsed: Value = serde_json::from_slice(&request.body).unwrap(); + assert_eq!(parsed["application"], json!("portainer")); + + let named = notification( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com/h\n credentials: [c]\n auth: {header: X-Gitlab-Token}\n headers: {X-Gitlab-Event: unified-gitops}\n", + ); + let request = build_request(&named, &m, &creds).unwrap(); + assert_eq!(header(&request, "X-Gitlab-Token"), Some("t0k")); + assert_eq!(header(&request, "Authorization"), None); + assert_eq!(header(&request, "X-Gitlab-Event"), Some("unified-gitops")); + + let basic = notification( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com/h\n credentials: [c]\n auth: basic\n", + ); + let err = build_request(&basic, &m, &creds).unwrap_err(); + assert!(err.contains("'username'"), "{err}"); + } + + #[test] + fn a_signed_webhook_carries_the_hmac_of_its_body() { + let creds = HashMap::from([("secret".to_string(), "Jefe".to_string())]); + let signed = notification( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com/h\n credentials: [c]\n auth: none\n body: 'what do ya want for nothing?'\n sign: {}\n", + ); + let request = build_request(&signed, &outcome_message(), &creds).unwrap(); + // RFC 4231, test case 2. + assert_eq!( + header(&request, "X-Hub-Signature-256"), + Some("sha256=5bdcc146bf60754e6a042426089575c75a003f089d2739839dec58b964ec3843") + ); + } + + #[test] + fn a_job_template_webhook_uses_the_gitlab_receiver_and_its_key() { + let creds = HashMap::from([("token".to_string(), "jt-key".to_string())]); + let n = notification( + "type: ascender\nevents: [drift_detected]\nascender:\n url: https://ascender.example.com\n job_template: 12\n credentials: [k]\n", + ); + let request = build_request(&n, &outcome_message(), &creds).unwrap(); + assert_eq!( + request.url, + "https://ascender.example.com/api/v2/job_templates/12/gitlab/" + ); + assert_eq!(header(&request, "X-Gitlab-Token"), Some("jt-key")); + assert_eq!( + header(&request, "X-Gitlab-Event"), + Some("unified-gitops sync_succeeded") + ); + assert_eq!(header(&request, "Authorization"), None); + let parsed: Value = serde_json::from_slice(&request.body).unwrap(); + assert_eq!(parsed["event"], json!("sync_succeeded")); + } + + #[test] + fn a_launch_sends_the_message_under_extra_vars() { + let creds = HashMap::from([("token".to_string(), "user-token".to_string())]); + let n = notification( + "type: awx\nevents: [drift_detected]\nawx:\n url: https://awx.example.com\n job_template: 5\n trigger: launch\n credentials: [t]\n extra_vars: {report: drift}\n", + ); + let request = build_request(&n, &outcome_message(), &creds).unwrap(); + assert_eq!( + request.url, + "https://awx.example.com/api/v2/job_templates/5/launch/" + ); + assert_eq!(header(&request, "Authorization"), Some("Bearer user-token")); + let parsed: Value = serde_json::from_slice(&request.body).unwrap(); + assert_eq!(parsed["extra_vars"]["report"], json!("drift")); + assert_eq!( + parsed["extra_vars"]["unified_gitops"]["application"], + json!("portainer") + ); + } + + #[test] + fn a_launch_whose_variables_were_ignored_is_a_failure() { + let n = notification( + "type: ascender\nevents: [drift_detected]\nascender:\n url: https://a.example.com\n job_template: 5\n trigger: launch\n credentials: [t]\n", + ); + let ignored = OutboundResponse { + status: 201, + body: r#"{"job": 77, "ignored_fields": {"extra_vars": {"unified_gitops": {}}}}"#.into(), + }; + assert!( + interpret(&n, &ignored) + .unwrap_err() + .contains("Prompt on launch") + ); + let started = OutboundResponse { + status: 201, + body: r#"{"job": 77, "id": 77}"#.into(), + }; + assert_eq!(interpret(&n, &started).unwrap(), Some("77".to_string())); + let refused = OutboundResponse { + status: 403, + body: r#"{"detail": "You do not have permission"}"#.into(), + }; + assert!(interpret(&n, &refused).unwrap_err().starts_with("HTTP 403")); + } + + #[test] + fn only_unanswered_or_server_failures_are_retried() { + assert!(retryable(&Err("timed out".into()))); + assert!(retryable(&Ok(OutboundResponse { + status: 503, + body: String::new() + }))); + assert!(retryable(&Ok(OutboundResponse { + status: 429, + body: String::new() + }))); + assert!(!retryable(&Ok(OutboundResponse { + status: 401, + body: String::new() + }))); + } +} diff --git a/src/application/reconcile.rs b/src/application/reconcile.rs index a7097ae..0177bf9 100644 --- a/src/application/reconcile.rs +++ b/src/application/reconcile.rs @@ -28,6 +28,9 @@ pub struct Catalog<'a> { pub applications: &'a HashMap, pub connectors: &'a HashMap, pub projects: &'a HashMap, + pub notifications: &'a HashMap, + // server.url, for links in notifications. + pub url: Option<&'a str>, } // The ports and registries a run uses. @@ -45,6 +48,8 @@ pub struct Reconciler<'a> { pub project_health: &'a SyncHealthRegistry, // What each connector last described, for search. pub described: &'a crate::domain::connector::DescribedKinds, + // Queues what a run has to tell (application::notify). + pub notifier: &'a crate::application::notify::Dispatcher, } // Pull the application's project, then reconcile. What every run should do — @@ -384,7 +389,24 @@ async fn reconcile_tree( } _ => r.health.record_success(app_id), } + // Where it stood before this run, read before the run is recorded over it. + let before = r.states.baseline(app_id); r.states.record(app_id, status, snapshot, record.clone()); + if let Some(detected) = crate::application::notify::detect(&before, &record, status) { + crate::application::notify::after_run( + r.notifier, + catalog.notifications, + &crate::application::notify::RunOutcome { + application: app_id, + description: app.description.as_deref(), + base_url: catalog.url, + before: &before, + status, + record: &record, + detected: &detected, + }, + ); + } tracing::info!( application = %app_id, mode = ?mode, diff --git a/src/config.rs b/src/config.rs index 51aa305..df34b29 100644 --- a/src/config.rs +++ b/src/config.rs @@ -7,6 +7,7 @@ use crate::domain::api_key::{ApiKeyDef, ApiKeyRole}; use crate::domain::app::Application; use crate::domain::connector::Connector; use crate::domain::credential::Credential; +use crate::domain::notification::Notification; use crate::domain::project::GitProject; pub struct AppConfig { @@ -20,6 +21,7 @@ pub struct AppConfig { pub connectors: HashMap, pub applications: HashMap, pub api_keys: HashMap, + pub notifications: HashMap, } // HTTP server configuration — config.yaml @@ -182,12 +184,13 @@ pub struct ConfigApiConfig { pub const REQUIRED_CONFIG_FILE: &str = "config.yaml"; // The top-level keys an item file may use. -pub const SECTIONS: [&str; 5] = [ +pub const SECTIONS: [&str; 6] = [ "applications", "connectors", "projects", "credentials", "api_keys", + "notifications", ]; // Whether a path, relative to the configuration directory, can be a @@ -535,6 +538,26 @@ impl AppConfig { } } + // Notifications: their own settings, then the applications they listen + // to and the credentials they send with. + for (id, notification) in &self.notifications { + errors.extend(notification.errors(id)); + for app_id in ¬ification.applications { + if !self.applications.contains_key(app_id) { + errors.push(format!( + "Notification '{id}' references unknown application '{app_id}'" + )); + } + } + for cred in notification.credentials() { + if !self.credentials.contains_key(cred) { + errors.push(format!( + "Notification '{id}' references unknown credential '{cred}'" + )); + } + } + } + // A vault_path needs a Vault to read it from. if self.secrets_config.vault.is_none() { for (id, credential) in &self.credentials { @@ -720,18 +743,20 @@ pub fn load_config_detailed(dir: &Path) -> Result { } } let mut section = |name: &str| entries.remove(name).unwrap_or_default(); - let (c, pr, pl, ap, ak) = ( + let (c, pr, pl, ap, ak, nt) = ( section("credentials"), section("projects"), section("connectors"), section("applications"), section("api_keys"), + section("notifications"), ); let credentials = take("credential", c, &mut errors); let projects = take("project", pr, &mut errors); let connectors = take("connector", pl, &mut errors); let applications = take("application", ap, &mut errors); let api_keys = take("API key", ak, &mut errors); + let notifications = take("notification", nt, &mut errors); if !errors.is_empty() { // Cross-references are only worth checking once every item parsed: // otherwise one typo'd connector also reports every application using it. @@ -749,6 +774,7 @@ pub fn load_config_detailed(dir: &Path) -> Result { connectors, applications, api_keys, + notifications, }; let errors = config.validate_errors(); @@ -923,6 +949,49 @@ mod tests { const PROJECT: &str = "projects:\n - name: prj-state\n git_url: https://example.com/state.git\n"; + const APPLICATION: &str = + "applications:\n - name: portainer\n connector: kv\n project: prj-state\n"; + const CREDENTIAL: &str = "credentials:\n - name: cred-key\n type: token\n env_prefix: JT\n secret_keys:\n token: KEY\n"; + + #[test] + fn notifications_load_as_items() { + let dir = dir_with(&[ + ("connectors/kv.yaml", CONNECTOR), + ("projects.yaml", PROJECT), + ("applications/portainer.yaml", APPLICATION), + ("credentials/cred-key.yaml", CREDENTIAL), + ( + "notifications/ntf-ascender-drift.yaml", + "notifications:\n - name: ntf-ascender-drift\n type: ascender\n events: [drift_detected]\n applications: [portainer]\n ascender:\n url: https://ascender.example.com\n job_template: 12\n credentials: [cred-key]\n", + ), + ]); + let cfg = load_config(dir.path().to_str().unwrap()).expect("loads"); + let n = &cfg.notifications["ntf-ascender-drift"]; + assert_eq!(n.applications, vec!["portainer".to_string()]); + } + + #[test] + fn a_notification_must_name_known_applications_and_credentials() { + let dir = dir_with(&[ + ("connectors/kv.yaml", CONNECTOR), + ("projects.yaml", PROJECT), + ("applications/portainer.yaml", APPLICATION), + ( + "notifications/n.yaml", + "notifications:\n - name: n\n type: webhook\n events: [sync_failed]\n applications: [portainr]\n webhook:\n url: https://hooks.example.com/x\n credentials: [cred-nope]\n", + ), + ]); + let err = load_err(&dir); + assert!( + err.contains("Notification 'n' references unknown application 'portainr'"), + "{err}" + ); + assert!( + err.contains("Notification 'n' references unknown credential 'cred-nope'"), + "{err}" + ); + } + #[test] fn items_load_from_any_depth_by_their_top_level_key() { let dir = dir_with(&[ diff --git a/src/domain/mod.rs b/src/domain/mod.rs index 50cb943..ddcb157 100644 --- a/src/domain/mod.rs +++ b/src/domain/mod.rs @@ -4,6 +4,7 @@ pub mod connector; pub mod credential; pub mod desired; pub mod ldap; +pub mod notification; pub mod plan; pub mod project; pub mod run; diff --git a/src/domain/notification.rs b/src/domain/notification.rs new file mode 100644 index 0000000..0a3e65e --- /dev/null +++ b/src/domain/notification.rs @@ -0,0 +1,623 @@ +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +// A notification: which events, from which applications, and where they go. +// The `type` names the receiver; its settings live in the block of the same +// name (`webhook:`, `ascender:`, `awx:`), so a block that belongs to another +// type is a load error rather than settings silently ignored, and adding a +// type later is a new block, with nothing to rename. +#[derive(Debug, Clone, PartialEq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Notification { + #[serde(default)] + pub description: Option, + #[serde(rename = "type")] + pub notification_type: NotificationType, + pub events: Vec, + // Empty = every application. + #[serde(default)] + pub applications: Vec, + #[serde(default)] + pub webhook: Option, + #[serde(default)] + pub ascender: Option, + #[serde(default)] + pub awx: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum NotificationType { + // Any HTTP receiver: its URL, method, headers, body and authentication. + Webhook, + // A job template in Ascender, through its webhook receiver or its launch API. + Ascender, + // The same in AWX. The routes are Ascender's today; they are separate + // types so neither carries the other's name, and either can follow its + // own product when they part. + Awx, +} + +impl NotificationType { + pub fn as_str(self) -> &'static str { + match self { + NotificationType::Webhook => "webhook", + NotificationType::Ascender => "ascender", + NotificationType::Awx => "awx", + } + } +} + +// What happened to an application, as a notification names it. One run +// produces at most one event (application::notify::detect). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Deserialize, Serialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum Event { + // The system differs from a commit already applied: someone changed it + // outside git. With auto_sync it is reverted in the same run. + DriftDetected, + // A new commit is waiting to be applied (auto_sync off). + OutOfSync, + // A sync applied at least one change from git (a commit or a rollback). + SyncSucceeded, + // A check or a sync could not finish, or some of its operations failed. + SyncFailed, + // In sync again after a failure or a difference. + BackInSync, +} + +impl Event { + pub fn as_str(self) -> &'static str { + match self { + Event::DriftDetected => "drift_detected", + Event::OutOfSync => "out_of_sync", + Event::SyncSucceeded => "sync_succeeded", + Event::SyncFailed => "sync_failed", + Event::BackInSync => "back_in_sync", + } + } +} + +#[derive(Debug, Clone, PartialEq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct WebhookTarget { + pub url: String, + #[serde(default = "default_method")] + pub method: String, + // Resolved per send; `auth` says which of their fields go where. + #[serde(default)] + pub credentials: Vec, + #[serde(default)] + pub auth: Auth, + // Fixed headers. Not secret: anything secret belongs in a credential. + #[serde(default)] + pub headers: BTreeMap, + // A template with {{field}} placeholders; None sends the default message. + #[serde(default)] + pub body: Option, + // An HMAC-SHA256 of the body, for receivers that check signatures. + #[serde(default)] + pub sign: Option, + #[serde(default = "default_timeout_seconds")] + pub timeout_seconds: u64, + #[serde(default = "default_retries")] + pub retries: u32, +} + +// How a credential reaches the receiver. Written as one word (`bearer`, +// `basic`, `none`) or as `{header: Name}`. +#[derive(Debug, Clone, PartialEq, Deserialize, Default)] +#[serde(untagged)] +pub enum Auth { + Scheme(AuthScheme), + Header { + header: String, + }, + // `auth:` left out: a bearer token when there is a credential, nothing + // when there is none. + #[default] + #[serde(skip)] + Default, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AuthScheme { + // Authorization: Bearer + Bearer, + // Authorization: Basic base64(:) + Basic, + // Nothing: the credentials are only there for the body or a signature. + None, +} + +#[derive(Debug, Clone, PartialEq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Signing { + #[serde(default = "default_signature_header")] + pub header: String, + // Written before the hex digest, as GitHub writes `sha256=`. + #[serde(default = "default_signature_prefix")] + pub prefix: String, + // The credential field holding the shared secret. + #[serde(default = "default_signature_field")] + pub secret_field: String, +} + +// A job template (or workflow job template) to start, in Ascender or AWX. +#[derive(Debug, Clone, PartialEq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct JobTemplateTarget { + // The controller's address, without /api. + pub url: String, + #[serde(default)] + pub job_template: Option, + #[serde(default)] + pub workflow_job_template: Option, + #[serde(default)] + pub trigger: JobTrigger, + // `webhook`: one credential whose `token` is the template's webhook key. + // `launch`: one credential whose `token` is a user's OAuth token. + #[serde(default)] + pub credentials: Vec, + // `launch` only: variables sent beside the event, which arrives as + // `unified_gitops`. The template must prompt for variables on launch. + #[serde(default)] + pub extra_vars: serde_json::Map, + #[serde(default = "default_timeout_seconds")] + pub timeout_seconds: u64, + #[serde(default = "default_retries")] + pub retries: u32, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, ToSchema, Default)] +#[serde(rename_all = "snake_case")] +pub enum JobTrigger { + // The template's own webhook receiver, in GitLab mode: its key in + // X-Gitlab-Token, no user needed. The playbook reads the event as + // `_webhook_payload`. + #[default] + Webhook, + // POST .../launch/ as a user, with extra_vars. + Launch, +} + +impl JobTrigger { + pub fn as_str(self) -> &'static str { + match self { + JobTrigger::Webhook => "webhook", + JobTrigger::Launch => "launch", + } + } +} + +fn default_method() -> String { + "POST".to_string() +} + +fn default_timeout_seconds() -> u64 { + 30 +} + +fn default_retries() -> u32 { + 3 +} + +fn default_signature_header() -> String { + "X-Hub-Signature-256".to_string() +} + +fn default_signature_prefix() -> String { + "sha256=".to_string() +} + +fn default_signature_field() -> String { + "secret".to_string() +} + +// The fields a body template may name: the default message's own keys, plus +// a short revision. Checked at load, so a typo fails the push, not a send. +pub const TEMPLATE_FIELDS: [&str; 15] = [ + "event", + "application", + "description", + "status", + "previous_status", + "cause", + "reverted", + "revision", + "revision_short", + "applied", + "changes", + "errors", + "at", + "url", + "notification", +]; + +const METHODS: [&str; 3] = ["POST", "PUT", "PATCH"]; + +// Longest a send may take: a receiver that never answers must not hold the +// sender (and every notification queued behind it) for long. +const MAX_TIMEOUT_SECONDS: u64 = 300; +const MAX_RETRIES: u32 = 10; + +impl Notification { + // Everything wrong with this item on its own; references to applications + // and credentials are checked by the loader, which knows them. + pub fn errors(&self, id: &str) -> Vec { + let mut errors = Vec::new(); + let at = |what: String| format!("Notification '{id}': {what}"); + if self.events.is_empty() { + errors.push(at("needs at least one event".to_string())); + } + let blocks = [ + ("webhook", self.webhook.is_some()), + ("ascender", self.ascender.is_some()), + ("awx", self.awx.is_some()), + ]; + let wanted = self.notification_type.as_str(); + for (block, present) in blocks { + if block == wanted && !present { + errors.push(at(format!("type {wanted} needs a '{wanted}:' block"))); + } + if block != wanted && present { + errors.push(at(format!( + "a '{block}:' block does not belong to type {wanted}" + ))); + } + } + if let Some(webhook) = &self.webhook { + errors.extend(webhook.errors().into_iter().map(at)); + } + for target in [&self.ascender, &self.awx].into_iter().flatten() { + errors.extend(target.errors().into_iter().map(at)); + } + errors + } + + // The credentials this notification names, whichever its type. + pub fn credentials(&self) -> &[String] { + if let Some(webhook) = &self.webhook { + return &webhook.credentials; + } + match (&self.ascender, &self.awx) { + (Some(target), _) | (None, Some(target)) => &target.credentials, + _ => &[], + } + } + + // Whether this notification wants `event` from `application`. + pub fn wants(&self, event: Event, application: &str) -> bool { + self.events.contains(&event) + && (self.applications.is_empty() || self.applications.iter().any(|a| a == application)) + } + + // Where it sends, without anything a URL can carry that is secret: a + // chat webhook's path or query is often the credential itself. + pub fn destination(&self) -> String { + let url = match (&self.webhook, &self.ascender, &self.awx) { + (Some(w), _, _) => &w.url, + (_, Some(t), _) | (_, _, Some(t)) => &t.url, + _ => return String::new(), + }; + origin(url) + } +} + +impl WebhookTarget { + fn errors(&self) -> Vec { + let mut errors = Vec::new(); + errors.extend(url_errors("webhook.url", &self.url)); + if !METHODS.contains(&self.method.as_str()) { + errors.push(format!( + "webhook.method '{}' must be one of {}", + self.method, + METHODS.join(", ") + )); + } + if let Some(body) = &self.body { + errors.extend(template_errors(body)); + } + let needs_credential = match &self.auth { + Auth::Scheme(AuthScheme::Bearer | AuthScheme::Basic) | Auth::Header { .. } => true, + Auth::Scheme(AuthScheme::None) | Auth::Default => false, + }; + if needs_credential && self.credentials.is_empty() { + errors.push("webhook.auth needs a credential in webhook.credentials".to_string()); + } + if let Auth::Header { header } = &self.auth + && !is_header_name(header) + { + errors.push(format!( + "webhook.auth header '{header}' is not a header name" + )); + } + for name in self.headers.keys() { + if !is_header_name(name) { + errors.push(format!("webhook.headers '{name}' is not a header name")); + } + } + if let Some(sign) = &self.sign { + if !is_header_name(&sign.header) { + errors.push(format!( + "webhook.sign header '{}' is not a header name", + sign.header + )); + } + if self.credentials.is_empty() { + errors.push("webhook.sign needs a credential holding its secret".to_string()); + } + } + errors.extend(bounds_errors("webhook", self.timeout_seconds, self.retries)); + errors + } +} + +impl JobTemplateTarget { + fn errors(&self) -> Vec { + let mut errors = Vec::new(); + errors.extend(url_errors("url", &self.url)); + if self.url.trim_end_matches('/').ends_with("/api") || self.url.contains("/api/v2") { + errors.push(format!( + "url '{}' is the controller's address, without /api", + self.url + )); + } + match (self.job_template, self.workflow_job_template) { + (Some(_), Some(_)) => { + errors.push("set job_template or workflow_job_template, not both".to_string()) + } + (None, None) => { + errors.push("needs a job_template or a workflow_job_template".to_string()) + } + _ => {} + } + if self.credentials.len() != 1 { + errors.push(format!( + "trigger {} needs exactly one credential, whose token is {}", + self.trigger.as_str(), + match self.trigger { + JobTrigger::Webhook => "the template's webhook key", + JobTrigger::Launch => "a user's token", + } + )); + } + if self.trigger == JobTrigger::Webhook && !self.extra_vars.is_empty() { + errors.push( + "extra_vars only go with trigger launch: a webhook sends the event itself" + .to_string(), + ); + } + errors.extend(bounds_errors( + "job template", + self.timeout_seconds, + self.retries, + )); + errors + } + + // The route a send goes to. + pub fn endpoint(&self) -> String { + let (collection, id) = match (self.job_template, self.workflow_job_template) { + (Some(id), _) => ("job_templates", id), + (None, Some(id)) => ("workflow_job_templates", id), + (None, None) => ("job_templates", 0), + }; + let action = match self.trigger { + JobTrigger::Webhook => "gitlab", + JobTrigger::Launch => "launch", + }; + format!( + "{}/api/v2/{collection}/{id}/{action}/", + self.url.trim_end_matches('/') + ) + } +} + +fn url_errors(field: &str, url: &str) -> Vec { + if url.starts_with("https://") || url.starts_with("http://") { + Vec::new() + } else { + vec![format!( + "{field} '{}' must be an http:// or https:// address", + origin(url) + )] + } +} + +fn bounds_errors(what: &str, timeout_seconds: u64, retries: u32) -> Vec { + let mut errors = Vec::new(); + if timeout_seconds == 0 || timeout_seconds > MAX_TIMEOUT_SECONDS { + errors.push(format!( + "{what} timeout_seconds must be between 1 and {MAX_TIMEOUT_SECONDS}" + )); + } + if retries > MAX_RETRIES { + errors.push(format!("{what} retries must be at most {MAX_RETRIES}")); + } + errors +} + +// Placeholders a body names that no message has. +fn template_errors(body: &str) -> Vec { + placeholders(body) + .filter(|name| !TEMPLATE_FIELDS.contains(name)) + .map(|name| { + format!( + "webhook.body names '{{{{{name}}}}}', which no message has (one of: {})", + TEMPLATE_FIELDS.join(", ") + ) + }) + .collect() +} + +// Every {{name}} in a template, in order. +pub fn placeholders(template: &str) -> impl Iterator { + template.split("{{").skip(1).filter_map(|rest| { + rest.split_once("}}") + .map(|(name, _)| name.trim()) + .filter(|name| !name.is_empty()) + }) +} + +fn is_header_name(name: &str) -> bool { + !name.is_empty() + && name + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') +} + +// scheme://host[:port] of a URL, or the text up to its first '/' after the +// scheme; never its user info, path or query. +pub fn origin(url: &str) -> String { + let (scheme, rest) = url.split_once("://").unwrap_or(("", url)); + let host = rest.split(['/', '?', '#']).next().unwrap_or(""); + let host = host.rsplit_once('@').map_or(host, |(_, h)| h); + if scheme.is_empty() { + host.to_string() + } else { + format!("{scheme}://{host}") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn parse(yaml: &str) -> Notification { + serde_saphyr::from_str(yaml).unwrap() + } + + #[test] + fn a_webhook_with_its_block_is_valid() { + let n = parse( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://alerts.example.com/x\n", + ); + assert!(n.errors("n").is_empty(), "{:?}", n.errors("n")); + assert_eq!(n.webhook.unwrap().method, "POST"); + } + + #[test] + fn the_type_and_its_block_must_agree() { + let n = parse( + "type: ascender\nevents: [sync_failed]\nwebhook:\n url: https://alerts.example.com/x\n", + ); + let errors = n.errors("n").join("\n"); + assert!( + errors.contains("type ascender needs a 'ascender:' block"), + "{errors}" + ); + assert!( + errors.contains("'webhook:' block does not belong to type ascender"), + "{errors}" + ); + } + + #[test] + fn an_unknown_event_or_key_fails_to_parse() { + assert!( + serde_saphyr::from_str::( + "type: webhook\nevents: [drifted]\nwebhook:\n url: https://x.example.com\n" + ) + .is_err() + ); + assert!(serde_saphyr::from_str::( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com\n token: x\n" + ) + .is_err()); + } + + #[test] + fn a_body_may_only_name_message_fields() { + let n = parse( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com\n body: '{\"a\": \"{{application}}\", \"b\": {{chnages}}}'\n", + ); + let errors = n.errors("n").join("\n"); + assert!(errors.contains("'{{chnages}}'"), "{errors}"); + assert!(!errors.contains("{{application}}"), "{errors}"); + assert_eq!(n.errors("n").len(), 1, "{errors}"); + } + + #[test] + fn auth_takes_a_scheme_or_a_header() { + let bearer = parse( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com\n credentials: [c]\n auth: basic\n", + ); + assert_eq!( + bearer.webhook.unwrap().auth, + Auth::Scheme(AuthScheme::Basic) + ); + let header = parse( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com\n credentials: [c]\n auth: {header: X-Gitlab-Token}\n", + ); + assert_eq!( + header.webhook.unwrap().auth, + Auth::Header { + header: "X-Gitlab-Token".into() + } + ); + let missing = parse( + "type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com\n auth: bearer\n", + ); + assert!(missing.errors("n").join("").contains("needs a credential")); + } + + #[test] + fn a_job_template_target_names_one_template_and_one_credential() { + let n = parse( + "type: ascender\nevents: [drift_detected]\nascender:\n url: https://ascender.example.com\n job_template: 7\n credentials: [k]\n", + ); + assert!(n.errors("n").is_empty(), "{:?}", n.errors("n")); + let target = n.ascender.unwrap(); + assert_eq!(target.trigger, JobTrigger::Webhook); + assert_eq!( + target.endpoint(), + "https://ascender.example.com/api/v2/job_templates/7/gitlab/" + ); + + let launch = parse( + "type: awx\nevents: [drift_detected]\nawx:\n url: https://awx.example.com/\n workflow_job_template: 9\n trigger: launch\n credentials: [t]\n", + ); + assert_eq!( + launch.awx.unwrap().endpoint(), + "https://awx.example.com/api/v2/workflow_job_templates/9/launch/" + ); + + let wrong = parse( + "type: ascender\nevents: [drift_detected]\nascender:\n url: https://ascender.example.com/api/v2\n job_template: 7\n workflow_job_template: 8\n extra_vars: {a: 1}\n", + ); + let errors = wrong.errors("n").join("\n"); + assert!(errors.contains("without /api"), "{errors}"); + assert!(errors.contains("not both"), "{errors}"); + assert!(errors.contains("exactly one credential"), "{errors}"); + assert!( + errors.contains("extra_vars only go with trigger launch"), + "{errors}" + ); + } + + #[test] + fn wants_filters_on_event_and_application() { + let n = parse( + "type: webhook\nevents: [sync_failed]\napplications: [a]\nwebhook:\n url: https://x.example.com\n", + ); + assert!(n.wants(Event::SyncFailed, "a")); + assert!(!n.wants(Event::SyncFailed, "b")); + assert!(!n.wants(Event::DriftDetected, "a")); + let all = + parse("type: webhook\nevents: [sync_failed]\nwebhook:\n url: https://x.example.com\n"); + assert!(all.wants(Event::SyncFailed, "anything")); + } + + #[test] + fn the_destination_never_shows_a_path_or_user_info() { + assert_eq!( + origin("https://user:pw@hooks.example.com/services/T0/B0/secret?x=1"), + "https://hooks.example.com" + ); + assert_eq!(origin("http://127.0.0.1:9000/x"), "http://127.0.0.1:9000"); + } +} diff --git a/src/domain/run.rs b/src/domain/run.rs index 9585673..6800f8f 100644 --- a/src/domain/run.rs +++ b/src/domain/run.rs @@ -171,10 +171,20 @@ pub const RECORDED_OPERATIONS: usize = 200; // would otherwise keep ten thousand of them, times HISTORY. pub const RECORDED_ERRORS: usize = 50; +// An application's standing before a run (AppStateRegistry::baseline). +#[derive(Debug, Clone, PartialEq)] +pub struct Baseline { + pub status: SyncStatus, + pub synced_revision: Option, +} + #[derive(Default)] struct Entry { status: Option, revision: Option, + // The commit the application was last found in sync at. Differences found + // at that same commit were made outside git: drift (application::notify). + synced_revision: Option, plan: Option, history: VecDeque, } @@ -211,6 +221,9 @@ impl AppStateRegistry { if run.revision.is_some() { entry.revision = run.revision.clone(); } + if status == SyncStatus::Synced && run.revision.is_some() { + entry.synced_revision = run.revision.clone(); + } if plan.is_some() { entry.plan = plan; } @@ -236,6 +249,22 @@ impl AppStateRegistry { } } + // Where the application stood before a run: what notifications compare + // the run's outcome with. + pub fn baseline(&self, app_id: &str) -> Baseline { + let entries = self.entries.read().unwrap_or_else(|e| e.into_inner()); + match entries.get(app_id) { + Some(e) => Baseline { + status: e.status.unwrap_or(SyncStatus::Unknown), + synced_revision: e.synced_revision.clone(), + }, + None => Baseline { + status: SyncStatus::Unknown, + synced_revision: None, + }, + } + } + pub fn status(&self, app_id: &str) -> SyncStatus { let entries = self.entries.read().unwrap_or_else(|e| e.into_inner()); entries diff --git a/src/lib.rs b/src/lib.rs index 6f436dd..2ac1f83 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -35,6 +35,8 @@ pub struct AppBuilder { connectors: HashMap, projects: HashMap, credentials: HashMap, + notifications: HashMap, + http: Arc, secrets_settings: config::SecretsConfig, projects_dir: std::path::PathBuf, ui: crate::config::UiConfig, @@ -64,6 +66,8 @@ impl AppBuilder { connectors: HashMap::new(), projects: HashMap::new(), credentials: HashMap::new(), + notifications: HashMap::new(), + http: Arc::new(adapters::out::notify::http::ReqwestSender::new()), secrets_settings: config::SecretsConfig::default(), projects_dir: std::path::PathBuf::from("projects"), ui: crate::config::UiConfig::default(), @@ -146,6 +150,20 @@ impl AppBuilder { // The credential DEFINITIONS (not their values), and the settings the // resolver chain is built from. Both only matter to a reload: it rebuilds // the chain from the new ones and reports which credential changed. + pub fn notifications( + mut self, + notifications: HashMap, + ) -> Self { + self.notifications = notifications; + self + } + + // What notifications are sent through: a recorder in tests. + pub fn http(mut self, http: Arc) -> Self { + self.http = http; + self + } + pub fn credentials( mut self, credentials: HashMap, @@ -168,6 +186,7 @@ impl AppBuilder { self.connectors = cfg.connectors.clone(); self.projects = cfg.projects.clone(); self.credentials = cfg.credentials.clone(); + self.notifications = cfg.notifications.clone(); self.secrets_settings = cfg.secrets_config.clone(); self.projects_dir = std::path::PathBuf::from(&cfg.projects_config.dir); self.state_dir = std::path::PathBuf::from(&cfg.state.dir); @@ -274,6 +293,7 @@ impl AppBuilder { connectors: self.connectors, projects: self.projects, credentials: self.credentials, + notifications: self.notifications, secrets: self.secrets_settings, metrics_require_auth: self.metrics_require_auth, cors_allowed_origins: self.cors_allowed_origins, @@ -298,6 +318,8 @@ impl AppBuilder { described: Arc::new(domain::connector::DescribedKinds::new()), started: std::time::Instant::now(), login_throttle: Arc::new(crate::application::login::LoginThrottle::default()), + notifier: Arc::new(application::notify::Dispatcher::new()), + http: self.http, }); let router = adapters::r#in::http::routes::create_router( Arc::clone(&state), diff --git a/src/main.rs b/src/main.rs index f077925..ecdc12e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,12 +16,13 @@ async fn main() { Ok(cfg) => { println!( "configuration OK ({}): {} applications, {} connectors, \ - {} credentials, {} projects", + {} credentials, {} projects, {} notifications", config_dir, cfg.applications.len(), cfg.connectors.len(), cfg.credentials.len(), cfg.projects.len(), + cfg.notifications.len(), ); return; } @@ -125,6 +126,9 @@ async fn main() { } let (app, state) = builder.build_with_state(); + // Sends what runs have to tell (notifications), from the first run on: + // a failure of the very first check is worth knowing. + state.start_notifier(); // One switch for every background task. Flipped after the HTTP server has // drained; each task returns at its next wait point (a running apply is diff --git a/src/ports/mod.rs b/src/ports/mod.rs index 4a615a1..2338236 100644 --- a/src/ports/mod.rs +++ b/src/ports/mod.rs @@ -2,6 +2,7 @@ pub mod config_store; pub mod connector; pub mod directory; pub mod git; +pub mod notifier; pub mod secrets; pub mod state_store; pub mod venv; diff --git a/src/ports/notifier.rs b/src/ports/notifier.rs new file mode 100644 index 0000000..9bd4b24 --- /dev/null +++ b/src/ports/notifier.rs @@ -0,0 +1,32 @@ +use std::future::Future; +use std::pin::Pin; +use std::time::Duration; + +// One HTTP request, built whole by application::notify: the adapter only +// carries it, so what a notification sends is decided (and tested) in one +// place, without a network. +#[derive(Debug, Clone, PartialEq)] +pub struct OutboundRequest { + pub method: String, + pub url: String, + pub headers: Vec<(String, String)>, + pub body: Vec, + pub timeout: Duration, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct OutboundResponse { + pub status: u16, + // At most a few KB: enough to read an error or a job id. + pub body: String, +} + +pub type SendFuture<'a> = + Pin> + Send + 'a>>; + +// Sends notifications. Err is a request that got no answer (refused, timed +// out, TLS); any answer, error statuses included, is Ok, and the caller +// decides what it means for its receiver. +pub trait HttpSenderPort: Send + Sync { + fn send(&self, request: OutboundRequest) -> SendFuture<'_>; +} diff --git a/src/state.rs b/src/state.rs index caa3c70..cd1e24b 100644 --- a/src/state.rs +++ b/src/state.rs @@ -5,6 +5,7 @@ use std::sync::{Arc, RwLock}; use crate::domain::app::Application; use crate::domain::connector::Connector; use crate::domain::credential::Credential; +use crate::domain::notification::Notification; use crate::domain::project::GitProject; use crate::domain::run::AppStateRegistry; use crate::domain::sync_health::SyncHealthRegistry; @@ -27,6 +28,8 @@ pub struct RuntimeConfig { // copy. It lives in the snapshot so a reload can say WHICH credential // changed. pub credentials: HashMap, + // Read after every run, so they reload like applications. + pub notifications: HashMap, // Carried for the same reason as `credentials`: the chain is rebuilt from // it on every reload. pub secrets: crate::config::SecretsConfig, @@ -54,6 +57,8 @@ impl RuntimeConfig { applications: &self.applications, connectors: &self.connectors, projects: &self.projects, + notifications: &self.notifications, + url: self.url.as_deref(), } } } @@ -171,6 +176,10 @@ pub struct AppState { // Recent failed LDAP logins and the cap on concurrent ones (see // application::login::LoginThrottle). pub login_throttle: Arc, + // Queues what runs have to tell, and sends it (application::notify). + pub notifier: Arc, + // What sends it: an HTTP client in production, a recorder in tests. + pub http: Arc, } impl AppState { @@ -224,6 +233,13 @@ impl AppState { *self.restart_pending.write().expect("restart pending lock") = keys; } + // Starts sending queued notifications. Once; main calls it beside the + // scheduler, and so does a test that wants to see what was sent. + pub fn start_notifier(&self) { + self.notifier + .start(Arc::clone(&self.secrets), Arc::clone(&self.http)); + } + // The ports a reconcile run uses, borrowed from the state that owns them. pub fn reconciler(&self) -> crate::application::reconcile::Reconciler<'_> { crate::application::reconcile::Reconciler { @@ -238,6 +254,7 @@ impl AppState { venv: &*self.venv, project_health: &self.project_health, described: &self.described, + notifier: &self.notifier, } } } diff --git a/tests/notifications_test.rs b/tests/notifications_test.rs new file mode 100644 index 0000000..9bbe743 --- /dev/null +++ b/tests/notifications_test.rs @@ -0,0 +1,300 @@ +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use serde_json::Value; +use std::collections::HashMap; +use std::path::Path; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tower::ServiceExt; + +use unified_gitops::adapters::out::secrets::mock::MockSecrets; +use unified_gitops::ports::notifier::{ + HttpSenderPort, OutboundRequest, OutboundResponse, SendFuture, +}; + +// Notifications through the real router and the example connector: a sync +// that applies a commit sends sync_succeeded to the receiver, with the +// credential where the notification says; a failing check sends sync_failed; +// the API lists the last send and sends a test. + +#[derive(Default)] +struct Recorder { + sent: Mutex>, +} + +impl HttpSenderPort for Recorder { + fn send(&self, request: OutboundRequest) -> SendFuture<'_> { + self.sent.lock().unwrap().push(request); + Box::pin(async { + Ok(OutboundResponse { + status: 202, + body: r#"{"message": "Job queued."}"#.into(), + }) + }) + } +} + +impl Recorder { + async fn wait_for(&self, count: usize) -> Vec { + for _ in 0..100 { + let sent = self.sent.lock().unwrap().clone(); + if sent.len() >= count { + return sent; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!( + "expected {count} sends, got {}", + self.sent.lock().unwrap().len() + ); + } +} + +fn write(path: &Path, body: &str) { + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, body).unwrap(); +} + +fn header<'a>(request: &'a OutboundRequest, name: &str) -> Option<&'a str> { + request + .headers + .iter() + .find(|(n, _)| n.eq_ignore_ascii_case(name)) + .map(|(_, v)| v.as_str()) +} + +struct Fixture { + app: axum::Router, + recorder: Arc, + _dirs: Vec, +} + +fn fixture() -> Fixture { + let projects = tempfile::tempdir().unwrap(); + let store = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + write( + &projects.path().join("prj-state/apps/demo/namespaces.yaml"), + "kind: namespace\nitems:\n - name: team-a\n description: Team A\n", + ); + + let connector = |script: &str| -> unified_gitops::domain::connector::Connector { + serde_saphyr::from_str(&format!("path: {}/{script}\n", env!("CARGO_MANIFEST_DIR"))).unwrap() + }; + let app_def = |connector: &str| -> unified_gitops::domain::app::Application { + serde_saphyr::from_str(&format!( + "description: Demo\nconnector: {connector}\nconfig:\n dir: {}\nproject: prj-state\npath: apps/demo\n", + store.path().display() + )) + .unwrap() + }; + let project: unified_gitops::domain::project::GitProject = + serde_saphyr::from_str("git_url: https://example.com/state.git\n").unwrap(); + let hook: unified_gitops::domain::notification::Notification = serde_saphyr::from_str( + "type: webhook\nevents: [sync_succeeded, sync_failed]\nwebhook:\n url: https://hooks.example.com/services/T0/secret-path\n credentials: [cred-hook]\n retries: 0\n", + ) + .unwrap(); + + let recorder = Arc::new(Recorder::default()); + let (app, app_state) = unified_gitops::AppBuilder::new() + .applications(HashMap::from([ + ("app-demo".to_string(), app_def("kv")), + ("app-broken".to_string(), app_def("missing")), + ])) + .connectors(HashMap::from([ + ("kv".to_string(), connector("tests/connectors/kv.py")), + ( + "missing".to_string(), + connector("tests/connectors/does-not-exist.py"), + ), + ])) + .projects( + HashMap::from([("prj-state".to_string(), project)]), + projects.path().to_path_buf(), + ) + .notifications(HashMap::from([("hook".to_string(), hook)])) + .secrets(Arc::new( + MockSecrets::new().with("cred-hook", [("token", "hook-token")]), + )) + .http(recorder.clone()) + .state_dir(state.path().to_path_buf()) + .api_key(Some("admin-secret".into())) + .build_with_state(); + app_state.start_notifier(); + + Fixture { + app, + recorder, + _dirs: vec![projects, store, state], + } +} + +async fn call(app: &axum::Router, method: &str, uri: &str) -> (StatusCode, Value) { + let resp = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(uri) + .header("x-api-key", "admin-secret") + .header("accept", "application/json") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let status = resp.status(); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + ( + status, + serde_json::from_slice(&bytes).unwrap_or(Value::Null), + ) +} + +#[tokio::test] +async fn a_sync_that_applies_a_change_sends_sync_succeeded() { + let f = fixture(); + let (status, run) = call(&f.app, "POST", "/api/v1/applications/app-demo/sync").await; + assert_eq!(status, StatusCode::OK, "{run}"); + assert_eq!(run["result"], "succeeded"); + + let sent = f.recorder.wait_for(1).await; + let request = &sent[0]; + assert_eq!(request.method, "POST"); + assert_eq!( + request.url, + "https://hooks.example.com/services/T0/secret-path" + ); + assert_eq!(header(request, "Authorization"), Some("Bearer hook-token")); + assert_eq!( + header(request, "X-Unified-Gitops-Event"), + Some("sync_succeeded") + ); + let body: Value = serde_json::from_slice(&request.body).unwrap(); + assert_eq!(body["event"], "sync_succeeded"); + assert_eq!(body["application"], "app-demo"); + assert_eq!(body["cause"], "commit"); + assert_eq!(body["applied"], 1); + assert_eq!(body["changes"][0]["kind"], "namespace"); + assert_eq!(body["notification"], "hook"); + + // A second sync finds nothing to do: nothing more is sent. + let (_, again) = call(&f.app, "POST", "/api/v1/applications/app-demo/sync").await; + assert_eq!(again["result"], "succeeded"); + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(f.recorder.sent.lock().unwrap().len(), 1); + + // The list shows the last send, and where it went without its path. + let (status, list) = call(&f.app, "GET", "/api/v1/notifications").await; + assert_eq!(status, StatusCode::OK, "{list}"); + assert_eq!(list[0]["name"], "hook"); + assert_eq!(list[0]["destination"], "https://hooks.example.com"); + assert_eq!(list[0]["last"]["ok"], true); + assert_eq!(list[0]["last"]["event"], "sync_succeeded"); + assert_eq!(list[0]["last"]["status"], 202); +} + +#[tokio::test] +async fn a_failing_check_sends_sync_failed_once() { + let f = fixture(); + let (_, run) = call(&f.app, "POST", "/api/v1/applications/app-broken/refresh").await; + assert_eq!(run["result"], "failed", "{run}"); + let sent = f.recorder.wait_for(1).await; + let body: Value = serde_json::from_slice(&sent[0].body).unwrap(); + assert_eq!(body["event"], "sync_failed"); + assert_eq!(body["application"], "app-broken"); + assert!(!body["errors"].as_array().unwrap().is_empty()); + + // Still failing: not a new event. + call(&f.app, "POST", "/api/v1/applications/app-broken/refresh").await; + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(f.recorder.sent.lock().unwrap().len(), 1); +} + +#[tokio::test] +async fn send_test_sends_now_and_answers_the_outcome() { + let f = fixture(); + let (status, delivery) = call(&f.app, "POST", "/api/v1/notifications/hook/test").await; + assert_eq!(status, StatusCode::OK, "{delivery}"); + assert_eq!(delivery["ok"], true); + assert_eq!(delivery["event"], "test"); + let sent = f.recorder.sent.lock().unwrap().clone(); + let body: Value = serde_json::from_slice(&sent[0].body).unwrap(); + assert_eq!(body["event"], "test"); + + let (status, _) = call(&f.app, "POST", "/api/v1/notifications/nope/test").await; + assert_eq!(status, StatusCode::NOT_FOUND); +} + +// The real client against a local receiver: headers and body arrive as +// built, and a redirect is reported, never followed with the credential. +#[tokio::test] +async fn the_http_sender_delivers_and_does_not_follow_redirects() { + use axum::routing::post; + let seen: Arc>> = Arc::default(); + let seen_in = Arc::clone(&seen); + let receiver = axum::Router::new() + .route( + "/hook", + post(move |headers: axum::http::HeaderMap, body: String| { + let seen = Arc::clone(&seen_in); + async move { + let token = headers + .get("x-gitlab-token") + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + seen.lock().unwrap().push((token, body)); + (StatusCode::ACCEPTED, "{\"message\": \"Job queued.\"}") + } + }), + ) + .route( + "/moved", + post(|| async { axum::response::Redirect::temporary("/hook") }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, receiver).await.unwrap() }); + + let sender = unified_gitops::adapters::out::notify::http::ReqwestSender::new(); + let request = |path: &str| OutboundRequest { + method: "POST".into(), + url: format!("http://{addr}{path}"), + headers: vec![ + ("X-Gitlab-Token".into(), "jt-key".into()), + ("Content-Type".into(), "application/json".into()), + ], + body: br#"{"event":"drift_detected"}"#.to_vec(), + timeout: Duration::from_secs(5), + }; + let answer = sender.send(request("/hook")).await.unwrap(); + assert_eq!(answer.status, 202); + assert!(answer.body.contains("Job queued.")); + assert_eq!( + seen.lock().unwrap()[0], + ( + "jt-key".to_string(), + r#"{"event":"drift_detected"}"#.to_string() + ) + ); + + let moved = sender.send(request("/moved")).await.unwrap(); + assert_eq!(moved.status, 307); + assert_eq!( + seen.lock().unwrap().len(), + 1, + "the redirect was not followed" + ); + + let unreachable = sender + .send(OutboundRequest { + url: "http://127.0.0.1:1/hook?token=secret".into(), + ..request("/hook") + }) + .await + .unwrap_err(); + assert!(!unreachable.contains("secret"), "{unreachable}"); +}