diff --git a/README.md b/README.md index bf7da245..01e43420 100644 --- a/README.md +++ b/README.md @@ -260,6 +260,10 @@ Run with config file: | `POSTHOG_HOST` | PostHog ingest host | `us.i.posthog.com` | | `ADDITIONAL_POSTHOG_API_KEYS` | **(Experimental)** Comma-separated list of additional PostHog API keys to publish logs to. Requires `POSTHOG_API_KEY` to be set. | - | | `DUCKGRES_IDENTIFIER` | Suffix appended to the OTel `service.name` in PostHog logs (e.g., `duckgres-acme`); only used when `POSTHOG_API_KEY` is set | - | +| `DUCKGRES_SLACK_WEBHOOK_URL` | Slack incoming webhook URL for operational notifications. Env-only because it is secret-bearing. | - | +| `DUCKGRES_NOTIFICATION_HASH_KEY` | Optional secret key used to include a stable HMAC org reference in Slack notifications. Env-only because it is secret-bearing. | - | +| `DUCKGRES_NOTIFICATION_QUEUE_SIZE` | In-process notification queue size before events are dropped instead of blocking user requests | `100` | +| `DUCKGRES_NOTIFICATION_TIMEOUT` | Per-notification delivery timeout | `2s` | ### PostHog Logging @@ -329,6 +333,29 @@ teardown), so you can build a provisioning funnel and alert on failures. > scales 1:1 with query throughput. Capture is asynchronous and batched, so it > stays off the query latency path. +### Slack Operational Notifications + +Set `DUCKGRES_SLACK_WEBHOOK_URL` to enable fire-and-forget Slack notifications +for org and managed-warehouse lifecycle events. Delivery is asynchronous: +Duckgres queues events in memory, applies `DUCKGRES_NOTIFICATION_TIMEOUT` +(default `2s`) per webhook call, and drops notifications when the bounded queue +is full (default size `100`) rather than blocking signup, provisioning, or +admin requests. Shutdown does not drain the in-memory notification queue; this +preserves the best-effort contract during process exit. + +Slack messages intentionally do **not** include raw org IDs, customer names, +database names, endpoints, SQL text, credentials, secret names, or bucket +names. They include the event name and a small allowlist of low-risk properties +such as `metadata_store`, `ducklake_enabled`, `source`, and stable failure +`reason`. Set `DUCKGRES_NOTIFICATION_HASH_KEY` to add a stable +`hmac-sha256` org reference for Slack-side correlation. + +See [`docs/runbooks/slack-notifications.md`](docs/runbooks/slack-notifications.md) +for local testing and failure recovery. + +Notification health is exposed through `duckgres_notifications_dropped_total` +and `duckgres_notification_delivery_failures_total`. + ### CLI Flags ```bash diff --git a/cmd/duckgres-controlplane/main.go b/cmd/duckgres-controlplane/main.go index 5aea959b..98b16da9 100644 --- a/cmd/duckgres-controlplane/main.go +++ b/cmd/duckgres-controlplane/main.go @@ -145,6 +145,8 @@ func main() { defer loggingShutdown() analyticsShutdown := cliboot.InitAnalytics() defer analyticsShutdown() + notificationsShutdown := cliboot.InitNotifications() + defer notificationsShutdown() tracingShutdown := cliboot.InitTracing() defer tracingShutdown() diff --git a/controlplane/admin/api.go b/controlplane/admin/api.go index c0c43ed7..a662c6ac 100644 --- a/controlplane/admin/api.go +++ b/controlplane/admin/api.go @@ -15,6 +15,7 @@ import ( "github.com/gin-gonic/gin" "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/internal/notifications" "gorm.io/gorm" "gorm.io/gorm/clause" "k8s.io/apimachinery/pkg/api/resource" @@ -554,6 +555,11 @@ func (h *apiHandler) createOrg(c *gin.Context) { c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) return } + notifications.Default().Notify(notifications.Event{ + Name: "org_created", + OrgID: org.Name, + Props: map[string]any{"source": "admin_api"}, + }) c.JSON(http.StatusCreated, org) } @@ -681,6 +687,11 @@ func (h *apiHandler) deleteOrg(c *gin.Context) { c.JSON(http.StatusNotFound, gin.H{"error": "org not found"}) return } + notifications.Default().Notify(notifications.Event{ + Name: "org_deleted", + OrgID: name, + Props: map[string]any{"source": "admin_api"}, + }) c.JSON(http.StatusOK, gin.H{"deleted": name}) } diff --git a/controlplane/admin/api_test.go b/controlplane/admin/api_test.go index e810f29f..9d39a61b 100644 --- a/controlplane/admin/api_test.go +++ b/controlplane/admin/api_test.go @@ -15,6 +15,7 @@ import ( "github.com/gin-gonic/gin" "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/internal/notifications" "gorm.io/gorm" ) @@ -24,6 +25,39 @@ type fakeAPIStore struct { warehouses map[string]*configstore.ManagedWarehouse } +type adminCapturedNotification struct { + name string + orgID string + props map[string]any +} + +type adminFakeNotifier struct { + events []adminCapturedNotification +} + +func (f *adminFakeNotifier) Notify(event notifications.Event) { + f.events = append(f.events, adminCapturedNotification{name: event.Name, orgID: event.OrgID, props: event.Props}) +} +func (f *adminFakeNotifier) Close() {} + +func installAdminFakeNotifier(t *testing.T) *adminFakeNotifier { + t.Helper() + fake := &adminFakeNotifier{} + notifications.SetDefault(fake) + t.Cleanup(func() { notifications.SetDefault(nil) }) + return fake +} + +func (f *adminFakeNotifier) count(name string) int { + n := 0 + for _, event := range f.events { + if event.name == name { + n++ + } + } + return n +} + func newFakeAPIStore() *fakeAPIStore { return &fakeAPIStore{ orgs: make(map[string]*configstore.Org), @@ -1139,6 +1173,7 @@ func TestDeleteOrgRejectsWhenWarehouseStillExists(t *testing.T) { } func TestDeleteOrgSucceedsAfterWarehouseRemoved(t *testing.T) { + notifier := installAdminFakeNotifier(t) store := newFakeAPIStore() seedOrgWithWarehouse(store, "analytics") delete(store.warehouses, "analytics") @@ -1154,9 +1189,13 @@ func TestDeleteOrgSucceedsAfterWarehouseRemoved(t *testing.T) { if _, ok := store.orgs["analytics"]; ok { t.Fatal("expected org to be deleted once the warehouse is gone") } + if got := notifier.count("org_deleted"); got != 1 { + t.Fatalf("org_deleted notifications = %d, want 1 (all: %+v)", got, notifier.events) + } } func TestDeleteOrgCascadesDeletedWarehouse(t *testing.T) { + notifier := installAdminFakeNotifier(t) store := newFakeAPIStore() seedOrgWithWarehouse(store, "analytics") // Simulate a completed deprovision: the provisioner leaves the warehouse @@ -1177,6 +1216,9 @@ func TestDeleteOrgCascadesDeletedWarehouse(t *testing.T) { if _, ok := store.warehouses["analytics"]; ok { t.Fatal("expected the deleted warehouse row to be cascaded away") } + if got := notifier.count("org_deleted"); got != 1 { + t.Fatalf("org_deleted notifications = %d, want 1 (all: %+v)", got, notifier.events) + } } func TestGetOrgOmitsMinWorkers(t *testing.T) { diff --git a/controlplane/provisioner/controller.go b/controlplane/provisioner/controller.go index 31951af3..3cdbd24f 100644 --- a/controlplane/provisioner/controller.go +++ b/controlplane/provisioner/controller.go @@ -10,6 +10,7 @@ import ( "github.com/posthog/duckgres/controlplane/configstore" "github.com/posthog/duckgres/internal/analytics" + "github.com/posthog/duckgres/internal/notifications" apierrors "k8s.io/apimachinery/pkg/api/errors" ) @@ -35,6 +36,7 @@ func captureProvisionFailed(w *configstore.ManagedWarehouse, reason string, upda props := provisionEventProps(w) props["reason"] = reason analytics.Default().Capture("warehouse_provision_failed", w.OrgID, props) + notifications.Default().Notify(notifications.Event{Name: "warehouse_provision_failed", OrgID: w.OrgID, Props: props}) } // ducklingCRName returns the Duckling CR name for a warehouse row: the stored @@ -354,7 +356,9 @@ func (c *Controller) reconcileProvisioning(ctx context.Context, w *configstore.M // usable. Guarded on a successful CAS from Provisioning, so this // fires exactly once — the next tick routes a Ready warehouse to // reconcileReady, not here. - analytics.Default().Capture("warehouse_provision_success", w.OrgID, provisionEventProps(w)) + props := provisionEventProps(w) + analytics.Default().Capture("warehouse_provision_success", w.OrgID, props) + notifications.Default().Notify(notifications.Event{Name: "warehouse_provision_success", OrgID: w.OrgID, Props: props}) } } @@ -535,6 +539,11 @@ func (c *Controller) reconcileDeleting(ctx context.Context, w *configstore.Manag analytics.Default().Capture("warehouse_deprovision_failed", w.OrgID, map[string]any{ "reason": "duckling_delete_failed", }) + notifications.Default().Notify(notifications.Event{ + Name: "warehouse_deprovision_failed", + OrgID: w.OrgID, + Props: map[string]any{"reason": "duckling_delete_failed"}, + }) return } } @@ -548,5 +557,6 @@ func (c *Controller) reconcileDeleting(ctx context.Context, w *configstore.Manag // Terminal success: all underlying resources are gone. Guarded on a // successful CAS from Deleting, so this fires exactly once. analytics.Default().Capture("warehouse_deprovision_success", w.OrgID, nil) + notifications.Default().Notify(notifications.Event{Name: "warehouse_deprovision_success", OrgID: w.OrgID}) } } diff --git a/controlplane/provisioner/controller_analytics_test.go b/controlplane/provisioner/controller_analytics_test.go index a88b1855..5f4edab4 100644 --- a/controlplane/provisioner/controller_analytics_test.go +++ b/controlplane/provisioner/controller_analytics_test.go @@ -16,6 +16,7 @@ import ( "github.com/posthog/duckgres/controlplane/configstore" "github.com/posthog/duckgres/internal/analytics" + "github.com/posthog/duckgres/internal/notifications" ) // capturedEvent records one analytics.Capture call. @@ -44,6 +45,53 @@ func installFakeTracker(t *testing.T) *fakeTracker { return fake } +type capturedNotification struct { + event string + orgID string + props map[string]any +} + +type fakeNotifier struct { + events []capturedNotification +} + +func (f *fakeNotifier) Notify(event notifications.Event) { + f.events = append(f.events, capturedNotification{event: event.Name, orgID: event.OrgID, props: event.Props}) +} +func (f *fakeNotifier) Close() {} + +func installFakeNotifier(t *testing.T) *fakeNotifier { + t.Helper() + fake := &fakeNotifier{} + notifications.SetDefault(fake) + t.Cleanup(func() { notifications.SetDefault(nil) }) + return fake +} + +func (f *fakeNotifier) only(t *testing.T, event string) capturedNotification { + t.Helper() + var found []capturedNotification + for _, e := range f.events { + if e.event == event { + found = append(found, e) + } + } + if len(found) != 1 { + t.Fatalf("expected exactly one %q notification, got %d (all: %+v)", event, len(found), f.events) + } + return found[0] +} + +func (f *fakeNotifier) count(event string) int { + n := 0 + for _, e := range f.events { + if e.event == event { + n++ + } + } + return n +} + // only returns the single event with the given name, failing if there isn't // exactly one. func (f *fakeTracker) only(t *testing.T, event string) capturedEvent { @@ -76,6 +124,7 @@ func (f *fakeTracker) count(event string) int { // warehouse metadata, and no failure event. func TestReconcileProvisioningSuccessEmitsEvent(t *testing.T) { fake := installFakeTracker(t) + notifier := installFakeNotifier(t) dc, fakeK8s := newFakeDucklingClient() fs := newFakeStore() fs.warehouses["org-ok"] = &configstore.ManagedWarehouse{ @@ -132,6 +181,13 @@ func TestReconcileProvisioningSuccessEmitsEvent(t *testing.T) { if n := fake.count("warehouse_provision_failed"); n != 0 { t.Errorf("expected no failure event, got %d", n) } + notification := notifier.only(t, "warehouse_provision_success") + if notification.orgID != "org-ok" { + t.Errorf("notification orgID = %q, want org-ok", notification.orgID) + } + if notification.props["metadata_store"] != string(configstore.MetadataStoreKindExternal) { + t.Errorf("notification metadata_store = %v, want external", notification.props["metadata_store"]) + } // A second reconcile finds the warehouse already Ready (routed to // reconcileReady), so the success event must not fire again. @@ -139,12 +195,16 @@ func TestReconcileProvisioningSuccessEmitsEvent(t *testing.T) { if n := fake.count("warehouse_provision_success"); n != 1 { t.Errorf("expected exactly one success event across two reconciles, got %d", n) } + if n := notifier.count("warehouse_provision_success"); n != 1 { + t.Errorf("expected exactly one success notification across two reconciles, got %d", n) + } } // TestReconcileProvisioningTimeoutEmitsFailure asserts the 30-minute timeout // path emits warehouse_provision_failed with reason=provisioning_timeout. func TestReconcileProvisioningTimeoutEmitsFailure(t *testing.T) { fake := installFakeTracker(t) + notifier := installFakeNotifier(t) dc, _ := newFakeDucklingClient() fs := newFakeStore() fs.warehouses["org-slow"] = &configstore.ManagedWarehouse{ @@ -171,6 +231,10 @@ func TestReconcileProvisioningTimeoutEmitsFailure(t *testing.T) { if e.props["metadata_store"] != string(configstore.MetadataStoreKindCnpgShard) { t.Errorf("metadata_store = %v, want cnpg-shard", e.props["metadata_store"]) } + notification := notifier.only(t, "warehouse_provision_failed") + if notification.props["reason"] != "provisioning_timeout" { + t.Errorf("notification reason = %v, want provisioning_timeout", notification.props["reason"]) + } } // TestReconcileProvisioningCrossplaneFailureEmitsFailure asserts that a @@ -178,6 +242,7 @@ func TestReconcileProvisioningTimeoutEmitsFailure(t *testing.T) { // with reason=crossplane_sync_failure. func TestReconcileProvisioningCrossplaneFailureEmitsFailure(t *testing.T) { fake := installFakeTracker(t) + notifier := installFakeNotifier(t) dc, fakeK8s := newFakeDucklingClient() fs := newFakeStore() fs.warehouses["org-xp"] = &configstore.ManagedWarehouse{ @@ -216,12 +281,17 @@ func TestReconcileProvisioningCrossplaneFailureEmitsFailure(t *testing.T) { if n := fake.count("warehouse_provision_success"); n != 0 { t.Errorf("expected no success event, got %d", n) } + notification := notifier.only(t, "warehouse_provision_failed") + if notification.props["reason"] != "crossplane_sync_failure" { + t.Errorf("notification reason = %v, want crossplane_sync_failure", notification.props["reason"]) + } } // TestReconcileDeletingSuccessEmitsEvent asserts that a completed teardown // emits exactly one warehouse_deprovision_success and no failure event. func TestReconcileDeletingSuccessEmitsEvent(t *testing.T) { fake := installFakeTracker(t) + notifier := installFakeNotifier(t) dc, fakeK8s := newFakeDucklingClient() fs := newFakeStore() fs.warehouses["org-del"] = &configstore.ManagedWarehouse{ @@ -251,6 +321,9 @@ func TestReconcileDeletingSuccessEmitsEvent(t *testing.T) { if n := fake.count("warehouse_deprovision_failed"); n != 0 { t.Errorf("expected no failure event, got %d", n) } + if e := notifier.only(t, "warehouse_deprovision_success"); e.orgID != "org-del" { + t.Errorf("notification orgID = %q, want org-del", e.orgID) + } } // TestReconcileDeletingFailureEmitsEvent asserts that a non-NotFound error @@ -258,6 +331,7 @@ func TestReconcileDeletingSuccessEmitsEvent(t *testing.T) { // warehouse in Deleting (so the controller retries). func TestReconcileDeletingFailureEmitsEvent(t *testing.T) { fake := installFakeTracker(t) + notifier := installFakeNotifier(t) dc, fakeK8s := newFakeDucklingClient() fs := newFakeStore() fs.warehouses["org-stuck"] = &configstore.ManagedWarehouse{ @@ -297,4 +371,8 @@ func TestReconcileDeletingFailureEmitsEvent(t *testing.T) { if n := fake.count("warehouse_deprovision_success"); n != 0 { t.Errorf("expected no success event, got %d", n) } + notification := notifier.only(t, "warehouse_deprovision_failed") + if notification.props["reason"] != "duckling_delete_failed" { + t.Errorf("notification reason = %v, want duckling_delete_failed", notification.props["reason"]) + } } diff --git a/controlplane/provisioning/api.go b/controlplane/provisioning/api.go index 79237711..2bc03daf 100644 --- a/controlplane/provisioning/api.go +++ b/controlplane/provisioning/api.go @@ -10,6 +10,7 @@ import ( "github.com/gin-gonic/gin" "github.com/posthog/duckgres/controlplane/configstore" "github.com/posthog/duckgres/internal/analytics" + "github.com/posthog/duckgres/internal/notifications" "gorm.io/gorm" ) @@ -76,7 +77,7 @@ type Store interface { // transaction so partial failure rolls back cleanly. Use this for the // public provision endpoint; the older per-step methods below are kept // for the standalone surfaces (reset-password). - Provision(req ProvisionRequest) error + Provision(req ProvisionRequest) (ProvisionResult, error) CreatePendingWarehouse(orgID, databaseName string, warehouse *configstore.ManagedWarehouse) error CreateOrgUser(orgID, username, passwordHash string) error UpdateOrgUserPassword(orgID, username, passwordHash string) error @@ -311,13 +312,14 @@ func (h *handler) provisionWarehouse(c *gin.Context) { // One transaction wraps warehouse + root user. Failure of any // sub-step rolls the others back so the caller's retry sees the same // starting state (no half-provisioned row blocking re-creation). - if err := h.store.Provision(ProvisionRequest{ + provisionResult, err := h.store.Provision(ProvisionRequest{ OrgID: orgID, DatabaseName: req.DatabaseName, DefaultTeamID: req.DefaultTeamID, Warehouse: warehouse, RootUserHash: hash, - }); err != nil { + }) + if err != nil { // The warehouse-already-exists conflict is the only error // shape that maps to 409. Everything else (DB write failure, // OnConflict surprise) is internal. The sentinel here @@ -355,6 +357,25 @@ func (h *handler) provisionWarehouse(c *gin.Context) { "metadata_store": string(req.MetadataStore.Type), "ducklake_enabled": ducklakeEnabled, }) + if provisionResult.OrgCreated { + notifications.Default().Notify(notifications.Event{ + Name: "org_created", + OrgID: orgID, + Props: map[string]any{ + "source": "provisioning", + "metadata_store": string(req.MetadataStore.Type), + "ducklake_enabled": ducklakeEnabled, + }, + }) + } + notifications.Default().Notify(notifications.Event{ + Name: "warehouse_provision_begin", + OrgID: orgID, + Props: map[string]any{ + "metadata_store": string(req.MetadataStore.Type), + "ducklake_enabled": ducklakeEnabled, + }, + }) resp := gin.H{ "status": "provisioning started", @@ -390,6 +411,7 @@ func (h *handler) deprovisionWarehouse(c *gin.Context) { // are emitted by the async provisioner controller as it deletes the // underlying resources. analytics.Default().Capture("warehouse_deprovision_begin", orgID, nil) + notifications.Default().Notify(notifications.Event{Name: "warehouse_deprovision_begin", OrgID: orgID}) c.JSON(http.StatusAccepted, gin.H{"status": "deprovisioning started", "org": orgID}) return } diff --git a/controlplane/provisioning/api_test.go b/controlplane/provisioning/api_test.go index c2d63168..d99d2e06 100644 --- a/controlplane/provisioning/api_test.go +++ b/controlplane/provisioning/api_test.go @@ -93,13 +93,13 @@ func (s *fakeStore) CreatePendingWarehouse(orgID, databaseName string, warehouse // handler treats partial failure as a complete rollback. func (s *fakeStore) setProvisionUserFailHook(err error) { s.provisionUserFailHook = err } -func (s *fakeStore) Provision(req ProvisionRequest) error { +func (s *fakeStore) Provision(req ProvisionRequest) (ProvisionResult, error) { // Pre-check: warehouse already exists in non-terminal state? Mirrors // createPendingWarehouseTx so the handler's 409 branch is exercised. if existing, ok := s.warehouses[req.OrgID]; ok && existing.State != configstore.ManagedWarehouseStateFailed && existing.State != configstore.ManagedWarehouseStateDeleted { - return ErrWarehouseNonTerminal + return ProvisionResult{}, ErrWarehouseNonTerminal } // Stage the writes into shadow maps so a mid-step failure can roll @@ -111,11 +111,13 @@ func (s *fakeStore) Provision(req ProvisionRequest) error { // 1. Warehouse + Org. Mirrors createPendingWarehouseTx: a NEW org // requires default_team_id (rejected before any write); an existing org // keeps its stored value when the field is omitted (set-only, never wipe). + orgCreated := false if _, ok := s.orgs[req.OrgID]; !ok { if req.DefaultTeamID == 0 { - return ErrDefaultTeamIDRequired + return ProvisionResult{}, ErrDefaultTeamIDRequired } s.orgs[req.OrgID] = &configstore.Org{Name: req.OrgID, DatabaseName: req.DatabaseName} + orgCreated = true } if req.DefaultTeamID != 0 { teamID := req.DefaultTeamID @@ -143,14 +145,14 @@ func (s *fakeStore) Provision(req ProvisionRequest) error { } else { s.warehouses[req.OrgID] = shadowWarehouse } - return s.provisionUserFailHook + return ProvisionResult{}, s.provisionUserFailHook } s.users[configstore.OrgUserKey{OrgID: req.OrgID, Username: "root"}] = req.RootUserHash // Reference the shadow vars so the linter doesn't complain about // declared-and-unused on the success path. _, _ = shadowUserHash, hadUser - return nil + return ProvisionResult{OrgCreated: orgCreated}, nil } func (s *fakeStore) IsDatabaseNameAvailable(name string) (bool, error) { diff --git a/controlplane/provisioning/notifications_test.go b/controlplane/provisioning/notifications_test.go new file mode 100644 index 00000000..d49df0d0 --- /dev/null +++ b/controlplane/provisioning/notifications_test.go @@ -0,0 +1,89 @@ +package provisioning + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + + "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/internal/notifications" +) + +type capturedNotification struct { + name string + orgID string + props map[string]any +} + +type fakeNotifier struct { + events []capturedNotification +} + +func (f *fakeNotifier) Notify(event notifications.Event) { + f.events = append(f.events, capturedNotification{name: event.Name, orgID: event.OrgID, props: event.Props}) +} +func (f *fakeNotifier) Close() {} + +func installFakeNotifier(t *testing.T) *fakeNotifier { + t.Helper() + fake := &fakeNotifier{} + notifications.SetDefault(fake) + t.Cleanup(func() { notifications.SetDefault(nil) }) + return fake +} + +func (f *fakeNotifier) count(name string) int { + n := 0 + for _, event := range f.events { + if event.name == name { + n++ + } + } + return n +} + +func TestProvisionNotifiesOnlyWhenOrgIsCreated(t *testing.T) { + fake := installFakeNotifier(t) + store := newFakeStore() + router := newTestRouter(store) + + body := []byte(`{"database_name": "acme-db", "default_team_id": 1, "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) + req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/acme/provision", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusAccepted { + t.Fatalf("status = %d, want 202: %s", rec.Code, rec.Body.String()) + } + + if got := fake.count("org_created"); got != 1 { + t.Fatalf("org_created notifications = %d, want 1 (all: %+v)", got, fake.events) + } + if got := fake.count("warehouse_provision_begin"); got != 1 { + t.Fatalf("warehouse_provision_begin notifications = %d, want 1 (all: %+v)", got, fake.events) + } +} + +func TestProvisionDoesNotNotifyOrgCreatedForExistingOrg(t *testing.T) { + fake := installFakeNotifier(t) + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme", DatabaseName: "old-db"} + router := newTestRouter(store) + + body := []byte(`{"database_name": "acme-db", "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) + req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/acme/provision", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusAccepted { + t.Fatalf("status = %d, want 202: %s", rec.Code, rec.Body.String()) + } + + if got := fake.count("org_created"); got != 0 { + t.Fatalf("org_created notifications = %d, want 0 (all: %+v)", got, fake.events) + } + if got := fake.count("warehouse_provision_begin"); got != 1 { + t.Fatalf("warehouse_provision_begin notifications = %d, want 1 (all: %+v)", got, fake.events) + } +} diff --git a/controlplane/provisioning/store.go b/controlplane/provisioning/store.go index 29e9a4cb..9b45f61b 100644 --- a/controlplane/provisioning/store.go +++ b/controlplane/provisioning/store.go @@ -51,6 +51,10 @@ type ProvisionRequest struct { RootUserHash string } +type ProvisionResult struct { + OrgCreated bool +} + // gormStore implements Store using a ConfigStore's GORM DB. type gormStore struct { cs *configstore.ConfigStore @@ -90,7 +94,8 @@ func (s *gormStore) CreatePendingWarehouse(orgID, databaseName string, warehouse // No default_team_id on this standalone path — an existing org keeps // its column as-is; creating a NEW org through here now fails with // ErrDefaultTeamIDRequired (same invariant as Provision). - return createPendingWarehouseTx(tx, orgID, databaseName, 0, warehouse) + _, err := createPendingWarehouseTx(tx, orgID, databaseName, 0, warehouse) + return err }) } @@ -103,7 +108,8 @@ func (s *gormStore) CreatePendingWarehouse(orgID, databaseName string, warehouse // Returns a sentinel-comparable error string ("warehouse already // exists in non-terminal state") so HTTP handlers can map to 409 // without an extra error type. -func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTeamID int64, warehouse *configstore.ManagedWarehouse) error { +func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTeamID int64, warehouse *configstore.ManagedWarehouse) (bool, error) { + orgCreated := false // Auto-create org if it doesn't exist (PostHog calls provision, duckgres // creates everything). A NEW org MUST carry default_team_id (pull-based // compute billing keys usage buckets by it; all pre-existing orgs are @@ -114,19 +120,20 @@ func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTe switch { case errors.Is(err, gorm.ErrRecordNotFound): if defaultTeamID == 0 { - return ErrDefaultTeamIDRequired + return false, ErrDefaultTeamIDRequired } org = configstore.Org{Name: orgID, DatabaseName: databaseName, DefaultTeamID: &defaultTeamID} if err := tx.Create(&org).Error; err != nil { - return err + return false, err } + orgCreated = true case err != nil: - return err + return false, err } // Update database name if org already existed with a different one if org.DatabaseName != databaseName { if err := tx.Model(&org).Update("database_name", databaseName).Error; err != nil { - return err + return false, err } } // If a default_team_id was supplied, persist it even when the org already @@ -134,7 +141,7 @@ func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTe // never cleared here, so an omitted value is a no-op rather than a wipe. if defaultTeamID != 0 { if err := tx.Model(&org).Update("default_team_id", defaultTeamID).Error; err != nil { - return err + return false, err } } @@ -144,13 +151,13 @@ func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTe if err == nil { if existing.State != configstore.ManagedWarehouseStateFailed && existing.State != configstore.ManagedWarehouseStateDeleted { - return ErrWarehouseNonTerminal + return false, ErrWarehouseNonTerminal } if err := tx.Delete(&existing).Error; err != nil { - return err + return false, err } } else if !errors.Is(err, gorm.ErrRecordNotFound) { - return err + return false, err } warehouse.OrgID = orgID @@ -159,7 +166,7 @@ func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTe warehouse.S3State = configstore.ManagedWarehouseStatePending warehouse.IdentityState = configstore.ManagedWarehouseStatePending warehouse.SecretsState = configstore.ManagedWarehouseStatePending - return tx.Create(warehouse).Error + return orgCreated, tx.Create(warehouse).Error } // Provision is the all-or-nothing entrypoint for POST /provision: one @@ -179,24 +186,27 @@ func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTe // The "warehouse already exists in non-terminal state" error is // passed through verbatim from createPendingWarehouseTx so the HTTP // handler can map it to 409. -func (s *gormStore) Provision(req ProvisionRequest) error { +func (s *gormStore) Provision(req ProvisionRequest) (ProvisionResult, error) { if req.OrgID == "" { - return errors.New("Provision: OrgID is required") + return ProvisionResult{}, errors.New("Provision: OrgID is required") } if req.DatabaseName == "" { - return errors.New("Provision: DatabaseName is required") + return ProvisionResult{}, errors.New("Provision: DatabaseName is required") } if req.Warehouse == nil { - return errors.New("Provision: Warehouse is required") + return ProvisionResult{}, errors.New("Provision: Warehouse is required") } if req.RootUserHash == "" { - return errors.New("Provision: RootUserHash is required") + return ProvisionResult{}, errors.New("Provision: RootUserHash is required") } - return s.cs.DB().Transaction(func(tx *gorm.DB) error { + var result ProvisionResult + err := s.cs.DB().Transaction(func(tx *gorm.DB) error { // 1. Warehouse + Org (extracted helper). - if err := createPendingWarehouseTx(tx, req.OrgID, req.DatabaseName, req.DefaultTeamID, req.Warehouse); err != nil { + orgCreated, err := createPendingWarehouseTx(tx, req.OrgID, req.DatabaseName, req.DefaultTeamID, req.Warehouse) + if err != nil { return err } + result.OrgCreated = orgCreated // 2. Root user. Same OnConflict semantics as // ConfigStore.CreateOrgUser so a retry overwrites the prior @@ -216,6 +226,7 @@ func (s *gormStore) Provision(req ProvisionRequest) error { return nil }) + return result, err } func (s *gormStore) IsDatabaseNameAvailable(name string) (bool, error) { diff --git a/docs/runbooks/slack-notifications.md b/docs/runbooks/slack-notifications.md new file mode 100644 index 00000000..31dde0a3 --- /dev/null +++ b/docs/runbooks/slack-notifications.md @@ -0,0 +1,70 @@ +# Slack notifications + +Use this runbook to enable and troubleshoot Duckgres operational notifications +to Slack. Notifications are best-effort and must never block signup, +provisioning, deprovisioning, or admin requests. + +## Configuration + +Set these environment variables on the control-plane process: + +```bash +DUCKGRES_SLACK_WEBHOOK_URL= +DUCKGRES_NOTIFICATION_HASH_KEY= +DUCKGRES_NOTIFICATION_QUEUE_SIZE=100 +DUCKGRES_NOTIFICATION_TIMEOUT=2s +``` + +`DUCKGRES_SLACK_WEBHOOK_URL` is required to enable Slack delivery. The queue size +and timeout are optional; the values above are the defaults. The hash key is +optional; set it only if Slack messages need a stable org correlation token. + +Shutdown does not drain the in-memory notification queue. This is intentional: +notifications remain best-effort during process exit just as they are on request +paths. + +Do not put the Slack webhook URL in `duckgres.yaml`, committed examples, PR +descriptions, issue comments, or test fixtures. Treat it as a secret. + +## Payload policy + +Slack messages intentionally omit raw customer and internal identifiers. They +include: + +- Event name, such as `org_created` or `warehouse_provision_failed` +- `org_ref=hmac-sha256:`, a short keyed HMAC of the org ID, only when + `DUCKGRES_NOTIFICATION_HASH_KEY` is set +- Low-risk allowlisted properties: `source`, `metadata_store`, + `ducklake_enabled`, and stable failure `reason` + +They do not include raw org IDs, customer names, database names, endpoints, SQL +text, credentials, secret names, bucket names, or hostnames. + +## Local smoke test + +1. Create a temporary Slack incoming webhook in a non-production Slack channel. +2. Run the control plane with `DUCKGRES_SLACK_WEBHOOK_URL` set. +3. Trigger `POST /api/v1/orgs//provision` with a throwaway org ID and + placeholder metadata. +4. Confirm Slack receives `org_created` and `warehouse_provision_begin`. +5. Confirm the message does not contain the raw org ID or database name. If + `DUCKGRES_NOTIFICATION_HASH_KEY` is set, confirm it contains only the HMAC + org reference. + +## Failure recovery + +- No Slack messages: confirm `DUCKGRES_SLACK_WEBHOOK_URL` is set on the running + control-plane pod or process, then restart/roll the process if the env var was + added after boot. +- Webhook revoked or wrong channel: rotate the Slack webhook URL in the secret + manager or deployment secret, then roll the control plane. +- Slow or failing Slack delivery: Duckgres logs delivery failures at debug level + and increments `duckgres_notification_delivery_failures_total` while keeping + requests flowing. Check the webhook URL, Slack app status, and outbound egress + policy. +- Dropped notifications: increase `DUCKGRES_NOTIFICATION_QUEUE_SIZE` if bursts + exceed the default queue. Watch `duckgres_notifications_dropped_total`. + Dropping is preferred over blocking signup or admin paths. + +Do not paste production org IDs, customer names, internal endpoints, or webhook +URLs into public issues, PRs, docs, or committed artifacts while debugging. diff --git a/duckgres.example.yaml b/duckgres.example.yaml index eda6d9b1..868eb21f 100644 --- a/duckgres.example.yaml +++ b/duckgres.example.yaml @@ -17,6 +17,11 @@ port: 5432 # Default: 60s # worker_queue_timeout: "60s" +# Slack operational notifications are env-only because the incoming webhook URL +# is secret-bearing. Use DUCKGRES_SLACK_WEBHOOK_URL plus optional +# DUCKGRES_NOTIFICATION_HASH_KEY, DUCKGRES_NOTIFICATION_QUEUE_SIZE, and +# DUCKGRES_NOTIFICATION_TIMEOUT. + # Directory for DuckDB database files (one per user) data_dir: "./data" diff --git a/internal/cliboot/notifications.go b/internal/cliboot/notifications.go new file mode 100644 index 00000000..e7d60135 --- /dev/null +++ b/internal/cliboot/notifications.go @@ -0,0 +1,55 @@ +package cliboot + +import ( + "fmt" + "net/http" + "os" + "strconv" + "time" + + "github.com/posthog/duckgres/internal/notifications" +) + +// InitNotifications installs optional fire-and-forget operational notifications. +// Slack delivery is enabled only when DUCKGRES_SLACK_WEBHOOK_URL is set. +func InitNotifications() func() { + webhookURL := os.Getenv("DUCKGRES_SLACK_WEBHOOK_URL") + if webhookURL == "" { + return func() {} + } + + queueSize := notifications.DefaultQueueSize + if v := os.Getenv("DUCKGRES_NOTIFICATION_QUEUE_SIZE"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + fmt.Fprintf(os.Stderr, "Invalid DUCKGRES_NOTIFICATION_QUEUE_SIZE %q; using default %d\n", v, queueSize) + } else { + queueSize = n + } + } + + timeout := notifications.DefaultTimeout + if v := os.Getenv("DUCKGRES_NOTIFICATION_TIMEOUT"); v != "" { + d, err := time.ParseDuration(v) + if err != nil || d <= 0 { + fmt.Fprintf(os.Stderr, "Invalid DUCKGRES_NOTIFICATION_TIMEOUT %q; using default %s\n", v, timeout) + } else { + timeout = d + } + } + + sink := notifications.NewSlackWebhookSinkWithOptions(webhookURL, http.DefaultClient, notifications.SlackWebhookOptions{ + OrgRefKey: os.Getenv("DUCKGRES_NOTIFICATION_HASH_KEY"), + }) + notifier := notifications.NewAsyncNotifier(sink, notifications.AsyncOptions{ + QueueSize: queueSize, + Timeout: timeout, + }) + notifications.SetDefault(notifier) + fmt.Fprintln(os.Stderr, "Slack notifications enabled.") + + return func() { + notifier.Close() + notifications.SetDefault(nil) + } +} diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go new file mode 100644 index 00000000..3b33b129 --- /dev/null +++ b/internal/notifications/notifications.go @@ -0,0 +1,238 @@ +package notifications + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "sort" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +const ( + DefaultQueueSize = 100 + DefaultTimeout = 2 * time.Second +) + +type Event struct { + Name string + OrgID string + Props map[string]any +} + +type Notifier interface { + Notify(Event) + Close() +} + +type Sink interface { + Notify(context.Context, Event) error +} + +type noopNotifier struct{} + +func (noopNotifier) Notify(Event) {} +func (noopNotifier) Close() {} + +type AsyncOptions struct { + QueueSize int + Timeout time.Duration +} + +type AsyncNotifier struct { + sink Sink + timeout time.Duration + ch chan Event + done chan struct{} + once sync.Once + dropped atomic.Uint64 +} + +func NewAsyncNotifier(sink Sink, opts AsyncOptions) *AsyncNotifier { + queueSize := opts.QueueSize + if queueSize <= 0 { + queueSize = DefaultQueueSize + } + timeout := opts.Timeout + if timeout <= 0 { + timeout = DefaultTimeout + } + n := &AsyncNotifier{ + sink: sink, + timeout: timeout, + ch: make(chan Event, queueSize), + done: make(chan struct{}), + } + go n.run() + return n +} + +func (n *AsyncNotifier) Notify(event Event) { + select { + case n.ch <- event: + default: + n.dropped.Add(1) + notificationDropsTotal.Inc() + slog.Warn("Notification queue full; dropping event.", "event", event.Name) + } +} + +func (n *AsyncNotifier) Dropped() uint64 { + return n.dropped.Load() +} + +// Close stops the background sender without draining queued events. Shutdown +// preserves the same best-effort contract as request-time delivery: never block +// process exit on Slack or another external notification sink. +func (n *AsyncNotifier) Close() { + n.once.Do(func() { + close(n.done) + }) +} + +func (n *AsyncNotifier) run() { + for { + select { + case event := <-n.ch: + ctx, cancel := context.WithTimeout(context.Background(), n.timeout) + if err := n.sink.Notify(ctx, event); err != nil { + notificationDeliveryFailuresTotal.Inc() + slog.Debug("Notification delivery failed.", "event", event.Name, "error", err) + } + cancel() + case <-n.done: + return + } + } +} + +var notificationDropsTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "duckgres_notifications_dropped_total", + Help: "Total number of operational notifications dropped because the in-process queue was full.", +}) + +var notificationDeliveryFailuresTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "duckgres_notification_delivery_failures_total", + Help: "Total number of operational notification delivery failures returned by the configured sink.", +}) + +type SlackWebhookSink struct { + webhookURL string + client *http.Client + orgRefKey []byte +} + +type SlackWebhookOptions struct { + OrgRefKey string +} + +func NewSlackWebhookSink(webhookURL string, client *http.Client) *SlackWebhookSink { + return NewSlackWebhookSinkWithOptions(webhookURL, client, SlackWebhookOptions{}) +} + +func NewSlackWebhookSinkWithOptions(webhookURL string, client *http.Client, opts SlackWebhookOptions) *SlackWebhookSink { + if client == nil { + client = http.DefaultClient + } + return &SlackWebhookSink{webhookURL: webhookURL, client: client, orgRefKey: []byte(opts.OrgRefKey)} +} + +func (s *SlackWebhookSink) Notify(ctx context.Context, event Event) error { + if s.webhookURL == "" { + return nil + } + body, err := json.Marshal(map[string]string{"text": s.slackText(event)}) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.webhookURL, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := s.client.Do(req) + if err != nil { + return err + } + defer func() { + if err := resp.Body.Close(); err != nil { + slog.Debug("Failed to close Slack webhook response body.", "error", err) + } + }() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("slack webhook returned %s", resp.Status) + } + return nil +} + +func (s *SlackWebhookSink) slackText(event Event) string { + parts := []string{ + "Duckgres notification", + "event=" + event.Name, + } + if ref := orgRef(event.OrgID, s.orgRefKey); ref != "" { + parts = append(parts, "org_ref=hmac-sha256:"+ref) + } + for _, key := range safePropKeys(event.Props) { + parts = append(parts, key+"="+fmt.Sprint(event.Props[key])) + } + return strings.Join(parts, "\n") +} + +func orgRef(orgID string, key []byte) string { + if orgID == "" || len(key) == 0 { + return "" + } + mac := hmac.New(sha256.New, key) + _, _ = mac.Write([]byte(orgID)) + sum := mac.Sum(nil) + return hex.EncodeToString(sum[:])[:12] +} + +func safePropKeys(props map[string]any) []string { + allow := map[string]bool{ + "ducklake_enabled": true, + "metadata_store": true, + "reason": true, + "source": true, + } + keys := make([]string, 0, len(props)) + for key := range props { + if allow[key] { + keys = append(keys, key) + } + } + sort.Strings(keys) + return keys +} + +var ( + defaultMu sync.RWMutex + current Notifier = noopNotifier{} +) + +func Default() Notifier { + defaultMu.RLock() + defer defaultMu.RUnlock() + return current +} + +func SetDefault(n Notifier) { + defaultMu.Lock() + defer defaultMu.Unlock() + if n == nil { + n = noopNotifier{} + } + current = n +} diff --git a/internal/notifications/notifications_test.go b/internal/notifications/notifications_test.go new file mode 100644 index 00000000..af85766f --- /dev/null +++ b/internal/notifications/notifications_test.go @@ -0,0 +1,169 @@ +package notifications + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + dto "github.com/prometheus/client_model/go" +) + +func TestSlackWebhookSinkRedactsSensitiveIdentifiers(t *testing.T) { + var payload map[string]string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Fatalf("method = %s, want POST", r.Method) + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode payload: %v", err) + } + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(srv.Close) + + sink := NewSlackWebhookSink(srv.URL, srv.Client()) + err := sink.Notify(t.Context(), Event{ + Name: "org_created", + OrgID: "example-org-123", + Props: map[string]any{ + "database_name": "example_database", + "metadata_store": "cnpg-shard", + "ducklake_enabled": true, + "source": "provisioning", + }, + }) + if err != nil { + t.Fatalf("Notify: %v", err) + } + + text := payload["text"] + if text == "" { + t.Fatal("expected Slack text payload") + } + for _, leaked := range []string{"example-org-123", "example_database", "database_name"} { + if strings.Contains(text, leaked) { + t.Fatalf("Slack payload leaked %q: %s", leaked, text) + } + } + if strings.Contains(text, "org_ref=") { + t.Fatalf("Slack payload included org reference without a hash key: %s", text) + } + for _, want := range []string{"org_created", "metadata_store=cnpg-shard", "ducklake_enabled=true", "source=provisioning"} { + if !strings.Contains(text, want) { + t.Fatalf("Slack payload missing %q: %s", want, text) + } + } +} + +func TestSlackWebhookSinkUsesHMACOrgReferenceWhenConfigured(t *testing.T) { + var payload map[string]string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode payload: %v", err) + } + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(srv.Close) + + sink := NewSlackWebhookSinkWithOptions(srv.URL, srv.Client(), SlackWebhookOptions{ + OrgRefKey: "local-test-key", + }) + err := sink.Notify(t.Context(), Event{Name: "org_created", OrgID: "example-org-123"}) + if err != nil { + t.Fatalf("Notify: %v", err) + } + + text := payload["text"] + if strings.Contains(text, "example-org-123") { + t.Fatalf("Slack payload leaked raw org ID: %s", text) + } + if !strings.Contains(text, "org_ref=hmac-sha256:") { + t.Fatalf("Slack payload missing HMAC org reference: %s", text) + } + if strings.Contains(text, "org_ref=sha256:") { + t.Fatalf("Slack payload used unsalted org reference: %s", text) + } +} + +func TestAsyncNotifierDropsWhenQueueFull(t *testing.T) { + before := notificationDropsTotalValue(t) + sink := &blockingSink{release: make(chan struct{})} + notifier := NewAsyncNotifier(sink, AsyncOptions{QueueSize: 1}) + t.Cleanup(func() { + close(sink.release) + notifier.Close() + }) + + notifier.Notify(Event{Name: "first", OrgID: "org-a"}) + notifier.Notify(Event{Name: "second", OrgID: "org-b"}) + notifier.Notify(Event{Name: "third", OrgID: "org-c"}) + + if got := notifier.Dropped(); got == 0 { + t.Fatal("expected at least one dropped notification") + } + if got := notificationDropsTotalValue(t); got <= before { + t.Fatalf("expected notification drop counter to increase, before=%v after=%v", before, got) + } +} + +func TestAsyncNotifierCountsDeliveryFailures(t *testing.T) { + before := notificationDeliveryFailuresTotalValue(t) + notifier := NewAsyncNotifier(failingSink{}, AsyncOptions{QueueSize: 1}) + t.Cleanup(notifier.Close) + + notifier.Notify(Event{Name: "warehouse_provision_failed", OrgID: "example-org"}) + waitForMetricIncrease(t, notificationDeliveryFailuresTotalValue, before) +} + +type blockingSink struct { + release chan struct{} +} + +func (s *blockingSink) Notify(_ context.Context, _ Event) error { + <-s.release + return nil +} + +type failingSink struct{} + +func (failingSink) Notify(context.Context, Event) error { + return errors.New("webhook failed") +} + +func notificationDropsTotalValue(t *testing.T) float64 { + t.Helper() + return counterValue(t, notificationDropsTotal) +} + +func notificationDeliveryFailuresTotalValue(t *testing.T) float64 { + t.Helper() + return counterValue(t, notificationDeliveryFailuresTotal) +} + +func counterValue(t *testing.T, counter interface { + Write(*dto.Metric) error +}) float64 { + t.Helper() + var metric dto.Metric + if err := counter.Write(&metric); err != nil { + t.Fatalf("read counter: %v", err) + } + return metric.GetCounter().GetValue() +} + +func waitForMetricIncrease(t *testing.T, value func(*testing.T) float64, before float64) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if value(t) > before { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("metric did not increase before timeout: before=%v after=%v", before, value(t)) +} diff --git a/main.go b/main.go index 53d071c2..b3e68cd1 100644 --- a/main.go +++ b/main.go @@ -154,6 +154,10 @@ func main() { fmt.Fprintf(os.Stderr, " DUCKGRES_CONFIG_POLL_INTERVAL Config store poll interval (default: 30s)\n") fmt.Fprintf(os.Stderr, " DUCKGRES_INTERNAL_SECRET Shared secret for API authentication\n") fmt.Fprintf(os.Stderr, " DUCKGRES_INTERNAL_SECRET_FALLBACKS Comma-separated previous internal secrets still accepted during rotation\n") + fmt.Fprintf(os.Stderr, " DUCKGRES_SLACK_WEBHOOK_URL Slack incoming webhook URL for operational notifications (disabled when unset)\n") + fmt.Fprintf(os.Stderr, " DUCKGRES_NOTIFICATION_HASH_KEY Optional secret key for HMAC org references in Slack notifications\n") + fmt.Fprintf(os.Stderr, " DUCKGRES_NOTIFICATION_QUEUE_SIZE Notification queue size (default: 100)\n") + fmt.Fprintf(os.Stderr, " DUCKGRES_NOTIFICATION_TIMEOUT Per-notification delivery timeout (default: 2s)\n") fmt.Fprintf(os.Stderr, " DUCKGRES_AWS_REGION AWS region for STS client\n") fmt.Fprintf(os.Stderr, " DUCKGRES_LOG_LEVEL Log level: debug, info, warn, error (default: info)\n") fmt.Fprintf(os.Stderr, "\nPrecedence: CLI flags > environment variables > config file > defaults\n") @@ -206,6 +210,9 @@ func main() { analyticsShutdown := cliboot.InitAnalytics() defer analyticsShutdown() + notificationsShutdown := cliboot.InitNotifications() + defer notificationsShutdown() + tracingShutdown := cliboot.InitTracing() defer tracingShutdown()