Skip to content

Commit 39ff054

Browse files
feat(worker): flow_synthetic continuous-monitoring runner (P0 matrix → metrics + NR event + reap) (#91)
* feat(worker): flow_synthetic continuous-monitoring runner (P0 matrix → metrics + NR event + reap) The capstone of the observability pillar: a River-scheduled job that runs the P0 user-flow matrix against prod every 5 min and pushes a green/red matrix to NR. Mirrors the auth_probe.go / deploy_probe.go prober pattern (per-flow result enum pass|fail|degraded, audit_log row on fail + slog ERROR NR fallback) and adds the two things the single-flow probers don't need: - the common/analyticsevent backend→NR bridge (RecordFlowTest → InstantFlowTest custom event, cohort="synthetic" so real dashboards exclude it), and - a create→assert→REAP lifecycle (rule-24 cleanup ledger): every resource a flow provisions is deleted via the real DELETE /api/v1/resources/:id path + a synthetic.reaped audit row, with a 30m orphan-sweep backstop. Flows (this PR, all prod-safe / API layer): - healthz (anon) GET /healthz → 200 + commit_id (also the per-tick commitId correlation for every event, rule 14/15) - auth_me (human) GET /auth/me with a minted session JWT → 200 + email - provision_reap(agent) POST /db/new → DELETE resource round-trip Brevo-free auth: the runner mints an HS256 session JWT locally (JWT_SECRET, the same value the api verifies against — like api/e2e makeSessionJWT), and idempotently seeds a deterministic cohort-tagged synthetic team + primary user, elevating its tier via the same UpgradeTeamAllTiers logic the Razorpay webhook uses. The cohort flag means every team-iterating background job (the already-shipped test_cohort.go guards) no-ops for it. Flag-gated OFF (FLOW_SYNTHETIC_ENABLED, default false — the DoD habit): the whole layer is inert in prod until an operator lights it. Per-flow kill list (FLOW_SYNTHETIC_DISABLED) + a Team-tier sub-flag (FLOW_SYNTHETIC_TEAM_ENABLED, off until Team GA). Registered in buildPeriodicJobs with UniqueOpts (replicas:2 safe), RunOnStart=true. Metrics (rule 25 — alert + tile + catalog land in the paired infra PR): instant_flow_test_total{flow,actor,tier,layer,result} instant_flow_test_latency_seconds{flow,actor,tier,layer} instant_flow_synthetic_reaped_total{flow,outcome} (leaked MUST stay 0) Tests: flag-off no-op, full happy-path matrix + reap + event assertions, per-flow fail/degraded/leak paths, kill switch, no-JWT/seed-failure degrade, orphan-sweep backstop, the Prom impl, and a DB-gated *Integration test proving the cohort-team seed + reaper soft-delete against real Postgres. make gate green (build + vet + go test ./... -short -count=1). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(worker): flow_synthetic patch coverage → 100% (test seams + reachable branches) The patch-coverage gate (diff-cover --fail-under=100) flagged defensive/error branches in flow_synthetic.go. Make every changed line covered: - injectable per-flow budget override (SetBudgetOverrideForTest) → drives the degraded-latency branches deterministically (0 budget = always over). - extract flowSyntheticNoRedirect + RunPanickingFlowForTest + ReapNilDBForTest test seams for the default-client CheckRedirect, the recover() boundary (also covers flowActorForFlow's default arm via an unknown flow id), and the db==nil guards in recordReap/emitFlowTestFailed. - tests for bad-base-URL build_request, closed-port http_error, seed sub-errors (begin/user/elevate/commit), reap http_error (conn hijack), audit-insert error (non-fatal), orphan scan/rows/update errors. - remove two genuinely-unreachable defensive branches (reapResource build_request — url.PathEscape sanitises; mintSessionJWT marshal error — a string/int map can't fail to marshal), matching the deploy_probe `_ =` posture. workers.go: Factory's Override path can't error (analyticsnr.New is never nil) → discard the structurally-dead advisory branch. diff-cover: internal/{config,jobs,workers}.go all 100% on the patch (0 missing). make gate green; golangci-lint clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 55c7b9f commit 39ff054

6 files changed

Lines changed: 2475 additions & 27 deletions

File tree

internal/config/config.go

Lines changed: 50 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -100,9 +100,9 @@ type Config struct {
100100
// post-cutover. The actual SDK path is the same minio-go client either
101101
// way (DO Spaces speaks the S3 API natively) — the env var lets ops flip
102102
// behavior (e.g. retention windows, bucket names) without a code change.
103-
ObjectStoreBackend string // OBJECT_STORE_BACKEND — "minio" (default) | "do-spaces"
104-
BackupS3Bucket string // BACKUP_S3_BUCKET — defaults to ObjectStoreBucket if empty
105-
BackupS3PathPrefix string // BACKUP_S3_PATH_PREFIX — defaults to "backups/"
103+
ObjectStoreBackend string // OBJECT_STORE_BACKEND — "minio" (default) | "do-spaces"
104+
BackupS3Bucket string // BACKUP_S3_BUCKET — defaults to ObjectStoreBucket if empty
105+
BackupS3PathPrefix string // BACKUP_S3_PATH_PREFIX — defaults to "backups/"
106106

107107
// PlatformBackupS3Prefix is the inner prefix under
108108
// BACKUP_S3_PATH_PREFIX for platform DB dumps. Distinct from the
@@ -151,6 +151,22 @@ type Config struct {
151151
DeployProbeBaseURL string // DEPLOY_PROBE_BASE_URL — default https://api.instanode.dev
152152
DeployProbeDeployHost string // DEPLOY_PROBE_DEPLOY_HOST — default deployment.instanode.dev
153153
DeployProbeBearerToken string // DEPLOY_PROBE_BEARER_TOKEN — probe-team session JWT (required)
154+
155+
// Continuous-monitoring synthetic flow runner (flow_synthetic.go) — runs a
156+
// matrix of P0 flows against prod, emits the instant_flow_test_* metrics +
157+
// the InstantFlowTest NR event, and pushes the green/red matrix dashboard.
158+
// INERT unless FlowSyntheticEnabled is true (the master flag) — a single env
159+
// flip turns the whole layer off. JWTSecret mints the Brevo-free session JWT
160+
// (plan §1.1); it reuses JWT_SECRET (the same value the api verifies against).
161+
// FlowSyntheticTeamEnabled stays OFF until Team is GA
162+
// (project_team_plan_not_rolled_out_no_payment.md).
163+
FlowSyntheticEnabled bool // FLOW_SYNTHETIC_ENABLED — master flag (default false)
164+
FlowSyntheticTeamEnabled bool // FLOW_SYNTHETIC_TEAM_ENABLED — Team-tier flows (default false)
165+
FlowSyntheticBaseURL string // FLOW_SYNTHETIC_BASE_URL — default https://api.instanode.dev
166+
FlowSyntheticEmail string // FLOW_SYNTHETIC_EMAIL — synthetic team primary-user email
167+
FlowSyntheticTier string // FLOW_SYNTHETIC_TIER — seeded tier (default free)
168+
FlowSyntheticDisabled string // FLOW_SYNTHETIC_DISABLED — comma list of per-flow kill switches
169+
FlowSyntheticJWTSecret string // JWT_SECRET — shared with api; mints the synthetic session JWT
154170
}
155171

156172
// ErrMissingConfig is returned when a required env var is absent.
@@ -180,24 +196,24 @@ func require(key string) string {
180196
// Load reads configuration from environment variables. Panics on missing required fields.
181197
func Load() *Config {
182198
cfg := &Config{
183-
DatabaseURL: require("DATABASE_URL"),
184-
RedisURL: getenv("REDIS_URL", "redis://localhost:6379"),
185-
ProvisionerAddr: os.Getenv("PROVISIONER_ADDR"),
186-
ProvisionerSecret: os.Getenv("PROVISIONER_SECRET"),
187-
EmailProvider: os.Getenv("EMAIL_PROVIDER"),
188-
BrevoAPIKey: os.Getenv("BREVO_API_KEY"),
189-
BrevoTemplateIDs: parseBrevoTemplateIDs(os.Getenv("BREVO_TEMPLATE_IDS")),
190-
BrevoSenderEmail: os.Getenv("BREVO_SENDER_EMAIL"),
191-
BrevoSenderName: os.Getenv("BREVO_SENDER_NAME"),
192-
SESAWSRegion: os.Getenv("SES_AWS_REGION"),
193-
SESAWSAccessKey: os.Getenv("SES_AWS_ACCESS_KEY_ID"),
194-
SESAWSSecretKey: os.Getenv("SES_AWS_SECRET_ACCESS_KEY"),
195-
SESFromEmail: os.Getenv("SES_FROM_EMAIL"),
196-
SESTemplateNames: parseSESTemplateNames(os.Getenv("SES_TEMPLATE_NAMES")),
197-
Environment: getenv("ENVIRONMENT", "development"),
198-
MaxMindLicenseKey: os.Getenv("MAXMIND_LICENSE_KEY"),
199-
GeoLite2DBPath: getenv("GEOLITE2_DB_PATH", "./GeoLite2-City.mmdb"),
200-
PlansPath: os.Getenv("PLANS_PATH"),
199+
DatabaseURL: require("DATABASE_URL"),
200+
RedisURL: getenv("REDIS_URL", "redis://localhost:6379"),
201+
ProvisionerAddr: os.Getenv("PROVISIONER_ADDR"),
202+
ProvisionerSecret: os.Getenv("PROVISIONER_SECRET"),
203+
EmailProvider: os.Getenv("EMAIL_PROVIDER"),
204+
BrevoAPIKey: os.Getenv("BREVO_API_KEY"),
205+
BrevoTemplateIDs: parseBrevoTemplateIDs(os.Getenv("BREVO_TEMPLATE_IDS")),
206+
BrevoSenderEmail: os.Getenv("BREVO_SENDER_EMAIL"),
207+
BrevoSenderName: os.Getenv("BREVO_SENDER_NAME"),
208+
SESAWSRegion: os.Getenv("SES_AWS_REGION"),
209+
SESAWSAccessKey: os.Getenv("SES_AWS_ACCESS_KEY_ID"),
210+
SESAWSSecretKey: os.Getenv("SES_AWS_SECRET_ACCESS_KEY"),
211+
SESFromEmail: os.Getenv("SES_FROM_EMAIL"),
212+
SESTemplateNames: parseSESTemplateNames(os.Getenv("SES_TEMPLATE_NAMES")),
213+
Environment: getenv("ENVIRONMENT", "development"),
214+
MaxMindLicenseKey: os.Getenv("MAXMIND_LICENSE_KEY"),
215+
GeoLite2DBPath: getenv("GEOLITE2_DB_PATH", "./GeoLite2-City.mmdb"),
216+
PlansPath: os.Getenv("PLANS_PATH"),
201217
ObjectStoreEndpoint: os.Getenv("OBJECT_STORE_ENDPOINT"),
202218
ObjectStoreAccessKey: os.Getenv("OBJECT_STORE_ACCESS_KEY"),
203219
ObjectStoreSecretKey: os.Getenv("OBJECT_STORE_SECRET_KEY"),
@@ -253,6 +269,19 @@ func Load() *Config {
253269
DeployProbeBaseURL: os.Getenv("DEPLOY_PROBE_BASE_URL"),
254270
DeployProbeDeployHost: os.Getenv("DEPLOY_PROBE_DEPLOY_HOST"),
255271
DeployProbeBearerToken: os.Getenv("DEPLOY_PROBE_BEARER_TOKEN"),
272+
273+
// Continuous-monitoring synthetic flow runner. INERT unless
274+
// FLOW_SYNTHETIC_ENABLED=true (default off, the DoD habit). JWTSecret
275+
// reuses JWT_SECRET so the worker mints a Brevo-free session JWT the api
276+
// verifies against. All other fields optional; defaults applied inside
277+
// jobs.FlowSyntheticConfig.Defaults().
278+
FlowSyntheticEnabled: os.Getenv("FLOW_SYNTHETIC_ENABLED") == "true",
279+
FlowSyntheticTeamEnabled: os.Getenv("FLOW_SYNTHETIC_TEAM_ENABLED") == "true",
280+
FlowSyntheticBaseURL: os.Getenv("FLOW_SYNTHETIC_BASE_URL"),
281+
FlowSyntheticEmail: os.Getenv("FLOW_SYNTHETIC_EMAIL"),
282+
FlowSyntheticTier: os.Getenv("FLOW_SYNTHETIC_TIER"),
283+
FlowSyntheticDisabled: os.Getenv("FLOW_SYNTHETIC_DISABLED"),
284+
FlowSyntheticJWTSecret: os.Getenv("JWT_SECRET"),
256285
}
257286

258287
// Fall back to the shared object-store bucket when the operator hasn't

0 commit comments

Comments
 (0)