From aa2852e541337635d8593e1f43a3ab9a33f14279 Mon Sep 17 00:00:00 2001 From: Thibault Le Ouay Date: Thu, 22 Oct 2026 10:30:36 +0200 Subject: [PATCH] checker: consume cron checks from Pub/Sub (#2963) Behind PUBSUB_ENABLED, each machine pulls checker- and checker--30s next to the HTTP routes. A message sleeps until its cronTimestamp, takes a probe slot (the 30s loop caps running checks separately from held ones), and is acked once the result is recorded, nacked on an infra error, and dropped when it can never succeed. On SIGTERM in-flight checks get 50s before being nacked, under the new 60s kill_timeout. scripts/pubsub-gate is the go/no-go check before the workflows cutover. Bumps otel to 1.44 (pulled by pubsub) and semconv to v1.41.0 to match its resource schema. Co-authored-by: Claude Opus 5.5 --- apps/checker/checker/update.go | 24 +- apps/checker/cmd/server/main.go | 101 +++++++- apps/checker/fly.toml | 3 + apps/checker/go.mod | 55 +++-- apps/checker/go.sum | 207 +++++++++++----- apps/checker/handlers/handler.go | 6 +- apps/checker/justfile | 3 + apps/checker/pkg/gcp/credentials.go | 33 +++ apps/checker/pkg/otel/otel.go | 2 +- apps/checker/pkg/queue/consumer.go | 257 +++++++++++++++++++ apps/checker/pkg/queue/consumer_test.go | 278 +++++++++++++++++++++ apps/checker/scripts/pubsub-gate/main.go | 301 +++++++++++++++++++++++ 12 files changed, 1162 insertions(+), 108 deletions(-) create mode 100644 apps/checker/pkg/gcp/credentials.go create mode 100644 apps/checker/pkg/queue/consumer.go create mode 100644 apps/checker/pkg/queue/consumer_test.go create mode 100644 apps/checker/scripts/pubsub-gate/main.go diff --git a/apps/checker/checker/update.go b/apps/checker/checker/update.go index b1299cfea..6f299b8a4 100644 --- a/apps/checker/checker/update.go +++ b/apps/checker/checker/update.go @@ -6,13 +6,13 @@ import ( "encoding/json" "fmt" "os" - "strings" "sync" "github.com/rs/zerolog/log" "google.golang.org/api/option" - "cloud.google.com/go/auth" + "github.com/openstatushq/openstatus/apps/checker/pkg/gcp" + cloudtasks "cloud.google.com/go/cloudtasks/apiv2" taskspb "cloud.google.com/go/cloudtasks/apiv2/cloudtaskspb" ) @@ -43,27 +43,11 @@ func cloudTasksClient() (*cloudtasks.Client, error) { return tasksClient, nil } - c := os.Getenv("GCP_PRIVATE_KEY") - c = strings.ReplaceAll(c, "\\n", "\n") - opts := &auth.Options2LO{ - Email: os.Getenv("GCP_CLIENT_EMAIL"), - PrivateKey: []byte(c), - PrivateKeyID: os.Getenv("GCP_PRIVATE_KEY_ID"), - Scopes: []string{ - "https://www.googleapis.com/auth/cloud-platform", - }, - TokenURL: "https://oauth2.googleapis.com/token", - } - - tp, err := auth.New2LOTokenProvider(opts) + creds, err := gcp.Credentials() if err != nil { - return nil, fmt.Errorf("creating token provider: %w", err) + return nil, err } - creds := auth.NewCredentials(&auth.CredentialsOptions{ - TokenProvider: tp, - }) - // The client lives as long as the process, so it takes neither the // cancellation nor the values (logger, monitor) of the request that built it. client, err := cloudtasks.NewClient(context.Background(), option.WithAuthCredentials(creds)) diff --git a/apps/checker/cmd/server/main.go b/apps/checker/cmd/server/main.go index 60ef72dc0..a0374e72f 100644 --- a/apps/checker/cmd/server/main.go +++ b/apps/checker/cmd/server/main.go @@ -9,14 +9,20 @@ import ( "net/http" "os" "os/signal" + "strconv" "syscall" "time" + "cloud.google.com/go/pubsub/v2" + "google.golang.org/api/option" + "github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/openstatushq/openstatus/apps/checker/handlers" + "github.com/openstatushq/openstatus/apps/checker/pkg/gcp" "github.com/openstatushq/openstatus/apps/checker/pkg/logger" + "github.com/openstatushq/openstatus/apps/checker/pkg/queue" "github.com/openstatushq/openstatus/apps/checker/pkg/tinybird" "github.com/rs/zerolog/log" "go.opentelemetry.io/contrib/bridges/otelslog" @@ -26,7 +32,7 @@ import ( "go.opentelemetry.io/otel/log/global" sdklog "go.opentelemetry.io/otel/sdk/log" "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.39.0" + semconv "go.opentelemetry.io/otel/semconv/v1.41.0" ) func shouldSample(event map[string]any) bool { @@ -272,12 +278,101 @@ func main() { } }() + consumerDone := make(chan struct{}) + if env("PUBSUB_ENABLED", "false") == "true" { + go func() { + defer close(consumerDone) + if err := runConsumer(ctx, h, region); err != nil { + log.Error().Err(err).Msg("pubsub consumer stopped") + } + // A consumer that stopped while the process keeps serving /health + // would never be restarted, so take the whole process down. + cancel() + }() + } else { + close(consumerDone) + } + <-ctx.Done() - if err := httpServer.Shutdown(ctx); err != nil { + // The consumer drains on its own from ctx; stop the listener alongside it. + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), consumerDrain) + defer shutdownCancel() + if err := httpServer.Shutdown(shutdownCtx); err != nil { log.Ctx(ctx).Error().Err(err).Msg("failed to shutdown http server") + } + <-consumerDone +} + +// consumerDrain stays under the 60s fly kill_timeout, so unfinished checks are +// nacked for redelivery before the machine is killed. +const consumerDrain = 50 * time.Second + +func runConsumer(ctx context.Context, h *handlers.Handler, region string) error { + creds, err := gcp.Credentials() + if err != nil { + return err + } + + client, err := pubsub.NewClient(context.Background(), env("GCP_PROJECT_ID", ""), option.WithAuthCredentials(creds)) + if err != nil { + return fmt.Errorf("creating pubsub client: %w", err) + } + defer client.Close() + + consumer := &queue.Consumer{ + Client: client, + Runner: h, + Loops: []queue.Loop{ + { + Subscription: queue.Subscription(region, false), + MaxOutstanding: envInt("PUBSUB_MAX_OUTSTANDING", 500), + }, + { + Subscription: queue.Subscription(region, true), + MaxOutstanding: envInt("PUBSUB_30S_MAX_OUTSTANDING", 2000), + MaxConcurrent: envInt("PUBSUB_30S_MAX_CONCURRENT", 200), + }, + }, + Drain: consumerDrain, + OnEvent: logConsumerEvent, + } - return + log.Info().Str("region", region).Msg("pubsub consumer started") + + return consumer.Run(ctx) +} + +func logConsumerEvent(ctx context.Context, e queue.Event) { + event := map[string]any{ + "timestamp": time.Now().Format(time.RFC3339), + "subscription": e.Subscription, + "monitor_id": e.MonitorID, + "periodicity": e.Periodicity, + "job_type": e.JobType, + "delivery_attempt": e.DeliveryAttempt, + "wait_ms": int(e.Wait.Milliseconds()), + "duration_ms": int(e.Duration.Milliseconds()), + "outcome": string(e.Outcome), + } + if e.Err != nil { + event["error"] = map[string]any{ + "type": "ConsumerError", + "message": e.Err.Error(), + } } + + if shouldSample(event) { + slog.LogAttrs(ctx, slog.LevelInfo, "message done", MapToAttrs(event)...) + } +} + +func envInt(key string, fallback int) int { + value, err := strconv.Atoi(env(key, "")) + if err != nil || value <= 0 { + return fallback + } + + return value } func env(key, fallback string) string { diff --git a/apps/checker/fly.toml b/apps/checker/fly.toml index 2cbf46a0f..c3fcb5ef9 100644 --- a/apps/checker/fly.toml +++ b/apps/checker/fly.toml @@ -5,6 +5,9 @@ app = "openstatus-checker" primary_region = "ams" +# The Pub/Sub consumer drains in-flight checks for 50s on SIGTERM (cmd/server/main.go). +kill_timeout = "60s" +kill_signal = "SIGTERM" [build] dockerfile = "./Dockerfile" diff --git a/apps/checker/go.mod b/apps/checker/go.mod index deee0d1f1..f82e9bcf5 100644 --- a/apps/checker/go.mod +++ b/apps/checker/go.mod @@ -3,8 +3,10 @@ module github.com/openstatushq/openstatus/apps/checker go 1.25.2 require ( - cloud.google.com/go/auth v0.18.2 + cloud.google.com/go/auth v0.20.0 cloud.google.com/go/cloudtasks v1.13.7 + cloud.google.com/go/monitoring v1.30.0 + cloud.google.com/go/pubsub/v2 v2.7.0 connectrpc.com/connect v1.19.1 github.com/cenkalti/backoff/v5 v5.0.3 github.com/gin-gonic/gin v1.12.0 @@ -13,30 +15,32 @@ require ( github.com/rs/zerolog v1.34.0 github.com/stretchr/testify v1.11.1 go.opentelemetry.io/contrib/bridges/otelslog v0.16.0 - go.opentelemetry.io/otel v1.41.0 + go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.17.0 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.41.0 go.opentelemetry.io/otel/log v0.17.0 - go.opentelemetry.io/otel/metric v1.41.0 - go.opentelemetry.io/otel/sdk v1.41.0 + go.opentelemetry.io/otel/metric v1.44.0 + go.opentelemetry.io/otel/sdk v1.44.0 go.opentelemetry.io/otel/sdk/log v0.17.0 - go.opentelemetry.io/otel/sdk/metric v1.41.0 - golang.org/x/net v0.51.0 - google.golang.org/api v0.269.0 - google.golang.org/grpc v1.79.1 + go.opentelemetry.io/otel/sdk/metric v1.44.0 + golang.org/x/net v0.58.0 + golang.org/x/sync v0.22.0 + google.golang.org/api v0.287.1 + google.golang.org/grpc v1.83.2 google.golang.org/protobuf v1.36.11 ) require ( + cloud.google.com/go v0.123.0 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect cloud.google.com/go/compute/metadata v0.9.0 // indirect - cloud.google.com/go/iam v1.5.3 // indirect + cloud.google.com/go/iam v1.11.0 // indirect github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic v1.15.0 // indirect github.com/bytedance/sonic/loader v0.5.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect - github.com/davecgh/go-spew v1.1.1 // indirect + github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/gabriel-vasile/mimetype v1.4.13 // indirect github.com/gin-contrib/sse v1.1.0 // indirect @@ -47,9 +51,10 @@ require ( github.com/go-playground/validator/v10 v10.30.1 // indirect github.com/goccy/go-json v0.10.5 // indirect github.com/goccy/go-yaml v1.19.2 // indirect + github.com/google/go-cmp v0.7.0 // indirect github.com/google/s2a-go v0.1.9 // indirect - github.com/googleapis/enterprise-certificate-proxy v0.3.12 // indirect - github.com/googleapis/gax-go/v2 v2.17.0 // indirect + github.com/googleapis/enterprise-certificate-proxy v0.3.17 // indirect + github.com/googleapis/gax-go/v2 v2.23.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect @@ -59,27 +64,27 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect - github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/quic-go/qpack v0.6.0 // indirect github.com/quic-go/quic-go v0.59.0 // indirect github.com/rs/xid v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.3.1 // indirect go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect + go.opencensus.io v0.24.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.66.0 // indirect - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0 // indirect - go.opentelemetry.io/otel/trace v1.41.0 // indirect + go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect golang.org/x/arch v0.24.0 // indirect - golang.org/x/crypto v0.48.0 // indirect - golang.org/x/oauth2 v0.35.0 // indirect - golang.org/x/sync v0.19.0 // indirect - golang.org/x/sys v0.41.0 // indirect - golang.org/x/text v0.34.0 // indirect - golang.org/x/time v0.14.0 // indirect - google.golang.org/genproto v0.0.0-20260226221140-a57be14db171 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260226221140-a57be14db171 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 // indirect + golang.org/x/crypto v0.55.0 // indirect + golang.org/x/oauth2 v0.36.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.41.0 // indirect + golang.org/x/time v0.15.0 // indirect + google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/apps/checker/go.sum b/apps/checker/go.sum index bb6dff56a..e14524a05 100644 --- a/apps/checker/go.sum +++ b/apps/checker/go.sum @@ -1,15 +1,23 @@ -cloud.google.com/go/auth v0.18.2 h1:+Nbt5Ev0xEqxlNjd6c+yYUeosQ5TtEUaNcN/3FozlaM= -cloud.google.com/go/auth v0.18.2/go.mod h1:xD+oY7gcahcu7G2SG2DsBerfFxgPAJz17zz2joOFF3M= +cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= +cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= +cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU= +cloud.google.com/go/auth v0.20.0 h1:kXTssoVb4azsVDoUiF8KvxAqrsQcQtB53DcSgta74CA= +cloud.google.com/go/auth v0.20.0/go.mod h1:942/yi/itH1SsmpyrbnTMDgGfdy2BUqIKyd0cyYLc5Q= cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= cloud.google.com/go/cloudtasks v1.13.7 h1:H2v8GEolNtMFfYzUpZBaZbydqU7drpyo99GtAgA+m4I= cloud.google.com/go/cloudtasks v1.13.7/go.mod h1:H0TThOUG+Ml34e2+ZtW6k6nt4i9KuH3nYAJ5mxh7OM4= cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs= cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10= -cloud.google.com/go/iam v1.5.3 h1:+vMINPiDF2ognBJ97ABAYYwRgsaqxPbQDlMnbHMjolc= -cloud.google.com/go/iam v1.5.3/go.mod h1:MR3v9oLkZCTlaqljW6Eb2d3HGDGK5/bDv93jhfISFvU= +cloud.google.com/go/iam v1.11.0 h1:KieQ9Pb+LLPak1O3Rv3GgCxhnmkYf7Xyh0P5HfF1jFM= +cloud.google.com/go/iam v1.11.0/go.mod h1:KP+nKGugNJW4LcLx1uEZcq1ok5sQHFaQehQNl4QDgV4= +cloud.google.com/go/monitoring v1.30.0 h1:r/d+JUbyKmJ8b07iznuKfzVzrIXTWxHQ3lBRm3x2LlY= +cloud.google.com/go/monitoring v1.30.0/go.mod h1:htlUR0QWVMrjFzZmN4LGnMAve9xB/eduwjmINxVZ8RM= +cloud.google.com/go/pubsub/v2 v2.7.0 h1:MFrBTZZa6PDWZzCi4NJRsHKMm2w0a4oAaYNqwjgbQTE= +cloud.google.com/go/pubsub/v2 v2.7.0/go.mod h1:JaFvWNVRk3Knoil/4M1ECeLOaI9D8drbmJWypQlK5aM= connectrpc.com/connect v1.19.1 h1:R5M57z05+90EfEvCY1b7hBxDVOUl45PrtXtAV2fOC14= connectrpc.com/connect v1.19.1/go.mod h1:tN20fjdGlewnSFeZxLKb0xwIZ6ozc3OQs2hTXy4du9w= +github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE= @@ -18,21 +26,29 @@ github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiD github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M= github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU= -github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5 h1:6xNmx7iTtyBRev0+D/Tv1FZd4SCg8axKApyNyRsAt/w= -github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5/go.mod h1:KdCmV+x/BuvyMxRnYBlmVaq4OLiKW6iRQfvC62cvdkI= +github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc= +github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 h1:aBangftG7EVZoUb69Os8IaYg++6uMOdKK83QtkkvJik= +github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4= github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= +github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= +github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= github.com/envoyproxy/go-control-plane v0.14.0 h1:hbG2kr4RuFj222B6+7T83thSPqLjwBIfQawTkC++2HA= -github.com/envoyproxy/go-control-plane/envoy v1.36.0 h1:yg/JjO5E7ubRyKX3m07GF3reDNEnfOboJ0QySbH736g= -github.com/envoyproxy/go-control-plane/envoy v1.36.0/go.mod h1:ty89S1YCCVruQAm9OtKeEkQLTb+Lkz0k8v9W0Oxsv98= -github.com/envoyproxy/protoc-gen-validate v1.3.0 h1:TvGH1wof4H33rezVKWSpqKz5NXWg5VPuZ0uONDT6eb4= -github.com/envoyproxy/protoc-gen-validate v1.3.0/go.mod h1:HvYl7zwPa5mffgyeTUHA9zHIH36nmrm7oCbo4YKoSWA= +github.com/envoyproxy/go-control-plane/envoy v1.37.0 h1:u3riX6BoYRfF4Dr7dwSOroNfdSbEPe9Yyl09/B6wBrQ= +github.com/envoyproxy/go-control-plane/envoy v1.37.0/go.mod h1:DReE9MMrmecPy+YvQOAOHNYMALuowAnbjjEMkkWOi6A= +github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= +github.com/envoyproxy/protoc-gen-validate v1.3.3 h1:MVQghNeW+LZcmXe7SY1V36Z+WFMDjpqGAGacLe2T0ds= +github.com/envoyproxy/protoc-gen-validate v1.3.3/go.mod h1:TsndJ/ngyIdQRhMcVVGDDHINPLWB7C82oDArY51KfB0= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/gabriel-vasile/mimetype v1.4.13 h1:46nXokslUBsAJE/wMsp5gtO500a4F3Nkz9Ufpk2AcUM= @@ -59,19 +75,40 @@ github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PU github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM= github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= +github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= +github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= +github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE= +github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= +github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A= +github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8= +github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA= +github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs= +github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w= +github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0= +github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8= +github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= +github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/s2a-go v0.1.9 h1:LGD7gtMgezd8a/Xak7mEWL0PjoTQFvpRudN895yqKW0= github.com/google/s2a-go v0.1.9/go.mod h1:YA0Ei2ZQL3acow2O62kdp9UlnvMmU7kA6Eutn0dXayM= +github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/googleapis/enterprise-certificate-proxy v0.3.12 h1:Fg+zsqzYEs1ZnvmcztTYxhgCBsx3eEhEwQ1W/lHq/sQ= -github.com/googleapis/enterprise-certificate-proxy v0.3.12/go.mod h1:vqVt9yG9480NtzREnTlmGSBmFrA+bzb0yl0TxoBQXOg= -github.com/googleapis/gax-go/v2 v2.17.0 h1:RksgfBpxqff0EZkDWYuz9q/uWsTVz+kf43LsZ1J6SMc= -github.com/googleapis/gax-go/v2 v2.17.0/go.mod h1:mzaqghpQp4JDh3HvADwrat+6M3MOIDp5YKHhb9PAgDY= +github.com/googleapis/enterprise-certificate-proxy v0.3.17 h1:73NfMHdiqo9JFU9+7a5ExpVa10/R29pXfZIaW559nrg= +github.com/googleapis/enterprise-certificate-proxy v0.3.17/go.mod h1:rSEsBUemEBZEexP2y6jPp16LUmUbjmSbcPMQizR0o4k= +github.com/googleapis/gax-go/v2 v2.23.0 h1:Tchl7qkvE7Ip3y+ztvNufYFvkfqTe7NfLTYGIdJRLuE= +github.com/googleapis/gax-go/v2 v2.23.0/go.mod h1:rBQKOVJCdb8IFEzg+FCwlt1LP/xMDGuqUXhUG+XMXEg= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= @@ -103,8 +140,10 @@ github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo= github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= @@ -122,6 +161,7 @@ github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= @@ -132,69 +172,122 @@ github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE= go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= +go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0= +go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/bridges/otelslog v0.16.0 h1:ZtVk8SzgioZhBJmJoezi6Jl5uuXoNVLnZxcJCDTqSbM= go.opentelemetry.io/contrib/bridges/otelslog v0.16.0/go.mod h1:p0C45DA3hvvo+5hwDilrMIp43ddVBGmwWEHZft4pY6c= -go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.66.0 h1:w/o339tDd6Qtu3+ytwt+/jon2yjAs3Ot8Xq8pelfhSo= -go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.66.0/go.mod h1:pdhNtM9C4H5fRdrnwO7NjxzQWhKSSxCHk/KluVqDVC0= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0 h1:PnV4kVnw0zOmwwFkAzCN5O07fw1YOIQor120zrh0AVo= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0/go.mod h1:ofAwF4uinaf8SXdVzzbL4OsxJ3VfeEg3f/F6CeF49/Y= -go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= -go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 h1:yI1/OhfEPy7J9eoa6Sj051C7n5dvpj0QX8g4sRchg04= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0/go.mod h1:NoUCKYWK+3ecatC4HjkRktREheMeEtrXoQxrqYFeHSc= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 h1:OyrsyzuttWTSur2qN/Lm0m2a8yqyIjUVBZcxFPuXq2o= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0/go.mod h1:C2NGBr+kAB4bk3xtMXfZ94gqFDtg/GkI7e9zqGh5Beg= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.17.0 h1:GcSx2UgcMuQEu0vHq823xR5LCN3WqEx5yKhqDkv1pwY= go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.17.0/go.mod h1:ctNT8t8Vzx9sb1oWAozighT3guWorr8xdCboBvkT5yg= go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.41.0 h1:MMrOAN8H1FrvDyq9UJ4lu5/+ss49Qgfgb7Zpm0m8ABo= go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.41.0/go.mod h1:Na+2NNASJtF+uT4NxDe0G+NQb+bUgdPDfwxY/6JmS/c= go.opentelemetry.io/otel/log v0.17.0 h1:blZWM4y7n+KSa9OywwGWyBMPpeVoCl/NCw+jMps8afM= go.opentelemetry.io/otel/log v0.17.0/go.mod h1:VXhjKYep6/laSgf/tjdh2SMAt18Z9XotBFBO0jxSE24= -go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= -go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= -go.opentelemetry.io/otel/sdk v1.41.0 h1:YPIEXKmiAwkGl3Gu1huk1aYWwtpRLeskpV+wPisxBp8= -go.opentelemetry.io/otel/sdk v1.41.0/go.mod h1:ahFdU0G5y8IxglBf0QBJXgSe7agzjE4GiTJ6HT9ud90= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/metric/x v0.66.0 h1:YkCrx1zLOChi9ZcZ6euupOcsgzbVlec7D/xoEU1+cTA= +go.opentelemetry.io/otel/metric/x v0.66.0/go.mod h1:d1+BDj9t96do0/1LoU1ayfCv79ZgNE41qbhBvnMOBZk= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= go.opentelemetry.io/otel/sdk/log v0.17.0 h1:stWOgJB8bWieSlX4VO+gD7BrRZ/Dh1H/u7115amleGE= go.opentelemetry.io/otel/sdk/log v0.17.0/go.mod h1:LQKPUyHraLka2sRvNQ5+W456+sElomqR7VWpOnOefZg= go.opentelemetry.io/otel/sdk/log/logtest v0.17.0 h1:Z4S9W5piCH88itCkWDtX5ppRgO0UTkLXVK/6tPOMM2w= go.opentelemetry.io/otel/sdk/log/logtest v0.17.0/go.mod h1:d9iIX/BwLfu1BTPxO0wi4ucyCenCckfuf9LC0aJDjqM= -go.opentelemetry.io/otel/sdk/metric v1.41.0 h1:siZQIYBAUd1rlIWQT2uCxWJxcCO7q3TriaMlf08rXw8= -go.opentelemetry.io/otel/sdk/metric v1.41.0/go.mod h1:HNBuSvT7ROaGtGI50ArdRLUnvRTRGniSUZbxiWxSO8Y= -go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= -go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= golang.org/x/arch v0.24.0 h1:qlJ3M9upxvFfwRM51tTg3Yl+8CP9vCC1E7vlFpgv99Y= golang.org/x/arch v0.24.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A= -golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= -golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= -golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= -golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= -golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= -golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= +golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= +golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= +golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= +golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU= +golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= +golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= +golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= +golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= +golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= -golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= -golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= -golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= -golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= -gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= -gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= -google.golang.org/api v0.269.0 h1:qDrTOxKUQ/P0MveH6a7vZ+DNHxJQjtGm/uvdbdGXCQg= -google.golang.org/api v0.269.0/go.mod h1:N8Wpcu23Tlccl0zSHEkcAZQKDLdquxK+l9r2LkwAauE= -google.golang.org/genproto v0.0.0-20260226221140-a57be14db171 h1:RxhCsti413yL0IjU9dVvuTbCISo8gs3RW1jPMStck+4= -google.golang.org/genproto v0.0.0-20260226221140-a57be14db171/go.mod h1:uhvzakVEqAuXU3TC2JCsxIRe5f77l+JySE3EqPoMyqM= -google.golang.org/genproto/googleapis/api v0.0.0-20260226221140-a57be14db171 h1:tu/dtnW1o3wfaxCOjSLn5IRX4YDcJrtlpzYkhHhGaC4= -google.golang.org/genproto/googleapis/api v0.0.0-20260226221140-a57be14db171/go.mod h1:M5krXqk4GhBKvB596udGL3UyjL4I1+cTbK0orROM9ng= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 h1:ggcbiqK8WWh6l1dnltU4BgWGIGo+EVYxCaAPih/zQXQ= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.79.1 h1:zGhSi45ODB9/p3VAawt9a+O/MULLl9dpizzNNpq7flY= -google.golang.org/grpc v1.79.1/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY= +golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= +golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/api v0.287.1 h1:LiyJx32VU3cwQfLchn/513qKhc25hq0pEANYJoWNnnI= +google.golang.org/api v0.287.1/go.mod h1:lM2kYRzYUCBY91P9h6VF1PYmvhxii3O5hji37qRvIcY= +google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM= +google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= +google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= +google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc= +google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo= +google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 h1:XzmzkmB14QhVhgnawEVsOn6OFsnpyxNPRY9QV01dNB0= +google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7/go.mod h1:L43LFes82YgSonw6iTXTxXUX1OlULt4AQtkik4ULL/I= +google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7 h1:jQ9p21COKWjP3VwuFrNRiiOTMh3mPpN45R7SLrH/HUU= +google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7/go.mod h1:KqHwBx2upmfa1XSi1WuRvC+2VGCLtooKkfmyvRbUmqA= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7 h1:eM/YSd5bBFagF51o1E745Ta7RwzpW0h+z+QDNZOgmQ8= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= +google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg= +google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY= +google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk= +google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= +google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= +google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= +google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= +google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE= +google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo= +google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= +google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= +google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= +google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= @@ -203,3 +296,5 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EV gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= +honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= diff --git a/apps/checker/handlers/handler.go b/apps/checker/handlers/handler.go index 954633f9a..07049831e 100644 --- a/apps/checker/handlers/handler.go +++ b/apps/checker/handlers/handler.go @@ -15,9 +15,9 @@ import ( // retryBudget stops scheduling new attempts once a cron check's retry loop has // run this long. The attempt in flight still finishes, so the worst case is -// retryBudget plus one monitor timeout (max 5m): 20m, inside the 30m Cloud -// Tasks dispatch deadline. Past that deadline Cloud Tasks resends the task and -// the check is recorded twice. +// retryBudget plus one monitor timeout (max 5m): 20m, inside both the 30m Cloud +// Tasks dispatch deadline and the 25m Pub/Sub lease extension (pkg/queue). +// Past either, the check is redelivered and recorded twice. const retryBudget = 15 * time.Minute type Handler struct { diff --git a/apps/checker/justfile b/apps/checker/justfile index a23de2394..97ff71248 100644 --- a/apps/checker/justfile +++ b/apps/checker/justfile @@ -18,3 +18,6 @@ clean: test: echo "Running tests..." go test ./... -v + +pubsub-gate monitor workspace: + go run ./scripts/pubsub-gate -monitor-id {{monitor}} -workspace-id {{workspace}} diff --git a/apps/checker/pkg/gcp/credentials.go b/apps/checker/pkg/gcp/credentials.go new file mode 100644 index 000000000..bd25e1d00 --- /dev/null +++ b/apps/checker/pkg/gcp/credentials.go @@ -0,0 +1,33 @@ +package gcp + +import ( + "fmt" + "os" + "strings" + + "cloud.google.com/go/auth" +) + +// Credentials builds the service account credentials shared by every GCP +// client of the checker from the GCP_* environment variables. +func Credentials() (*auth.Credentials, error) { + privateKey := strings.ReplaceAll(os.Getenv("GCP_PRIVATE_KEY"), "\\n", "\n") + opts := &auth.Options2LO{ + Email: os.Getenv("GCP_CLIENT_EMAIL"), + PrivateKey: []byte(privateKey), + PrivateKeyID: os.Getenv("GCP_PRIVATE_KEY_ID"), + Scopes: []string{ + "https://www.googleapis.com/auth/cloud-platform", + }, + TokenURL: "https://oauth2.googleapis.com/token", + } + + tp, err := auth.New2LOTokenProvider(opts) + if err != nil { + return nil, fmt.Errorf("creating token provider: %w", err) + } + + return auth.NewCredentials(&auth.CredentialsOptions{ + TokenProvider: tp, + }), nil +} diff --git a/apps/checker/pkg/otel/otel.go b/apps/checker/pkg/otel/otel.go index dfce446e0..f4f4bd0b4 100644 --- a/apps/checker/pkg/otel/otel.go +++ b/apps/checker/pkg/otel/otel.go @@ -15,7 +15,7 @@ import ( sdkMetrics "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.39.0" + semconv "go.opentelemetry.io/otel/semconv/v1.41.0" ) func setupOTelSDK(ctx context.Context, url string, headers map[string]string) (shutdown func(context.Context) error, err error) { diff --git a/apps/checker/pkg/queue/consumer.go b/apps/checker/pkg/queue/consumer.go new file mode 100644 index 000000000..34b0db995 --- /dev/null +++ b/apps/checker/pkg/queue/consumer.go @@ -0,0 +1,257 @@ +package queue + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "cloud.google.com/go/pubsub/v2" + "golang.org/x/sync/errgroup" + + "github.com/openstatushq/openstatus/apps/checker/checker" + "github.com/openstatushq/openstatus/apps/checker/handlers" + "github.com/openstatushq/openstatus/apps/checker/request" +) + +type Runner interface { + RunHTTPCheck(ctx context.Context, req request.HttpCheckerRequest) (checker.Response, error) + RunTCPCheck(ctx context.Context, req request.TCPCheckerRequest) (checker.TCPResponse, error) + RunDNSCheck(ctx context.Context, req request.DNSCheckerRequest) (handlers.DNSResponse, error) + RunICMPCheck(ctx context.Context, req request.ICMPCheckerRequest) (checker.ICMPResponse, error) + RunGRPCCheck(ctx context.Context, req request.GRPCCheckerRequest) (checker.GRPCResponse, error) +} + +type Outcome string + +const ( + OutcomeAck Outcome = "ack" + OutcomeNack Outcome = "nack" + OutcomePoison Outcome = "poison" + OutcomeDrain Outcome = "drain" +) + +type Event struct { + Subscription string + MonitorID string + Periodicity string + JobType string + DeliveryAttempt int + Wait time.Duration + Duration time.Duration + Outcome Outcome + Err error +} + +type Loop struct { + Subscription string + // MaxOutstanding caps the messages held at once, sleeping ones included. + MaxOutstanding int + // MaxConcurrent caps the checks running at once; zero means MaxOutstanding. + MaxConcurrent int +} + +type Consumer struct { + Client *pubsub.Client + Runner Runner + Loops []Loop + // Drain is how long in-flight checks get to finish after shutdown starts + // before their message is nacked for redelivery. + Drain time.Duration + OnEvent func(ctx context.Context, e Event) +} + +const ( + maxOutstandingBytes = 64 << 20 + maxExtension = 25 * time.Minute +) + +// Subscription names the region's subscription for a periodicity class. +func Subscription(region string, thirtySeconds bool) string { + if thirtySeconds { + return fmt.Sprintf("checker-%s-30s", region) + } + + return fmt.Sprintf("checker-%s", region) +} + +// Run blocks until ctx is done and every loop has drained. +func (c *Consumer) Run(ctx context.Context) error { + g, gctx := errgroup.WithContext(ctx) + for _, loop := range c.Loops { + g.Go(func() error { + return c.receive(gctx, loop) + }) + } + + return g.Wait() +} + +func (c *Consumer) receive(ctx context.Context, loop Loop) error { + if loop.MaxOutstanding <= 0 { + return fmt.Errorf("max outstanding for %s must be positive", loop.Subscription) + } + + sub := c.Client.Subscriber(loop.Subscription) + sub.ReceiveSettings.MaxOutstandingMessages = loop.MaxOutstanding + sub.ReceiveSettings.MaxOutstandingBytes = maxOutstandingBytes + sub.ReceiveSettings.MaxExtension = maxExtension + sub.ReceiveSettings.ShutdownOptions = &pubsub.ShutdownOptions{ + Timeout: c.Drain + 5*time.Second, + Behavior: pubsub.ShutdownBehaviorWaitForProcessing, + } + + concurrent := loop.MaxConcurrent + if concurrent <= 0 { + concurrent = loop.MaxOutstanding + } + slots := make(chan struct{}, concurrent) + + err := sub.Receive(ctx, func(msgCtx context.Context, msg *pubsub.Message) { + e := c.handle(msgCtx, loop.Subscription, slots, msg) + if c.OnEvent != nil { + c.OnEvent(msgCtx, e) + } + }) + if err != nil { + return fmt.Errorf("receiving from %s: %w", loop.Subscription, err) + } + + return nil +} + +type envelope struct { + CronTimestamp int64 `json:"cronTimestamp"` +} + +func (c *Consumer) handle(ctx context.Context, subscription string, slots chan struct{}, msg *pubsub.Message) Event { + start := time.Now() + e := Event{ + Subscription: subscription, + MonitorID: msg.Attributes["monitorId"], + Periodicity: msg.Attributes["periodicity"], + JobType: msg.Attributes["jobType"], + } + if msg.DeliveryAttempt != nil { + e.DeliveryAttempt = *msg.DeliveryAttempt + } + finish := func(outcome Outcome, err error) Event { + switch outcome { + case OutcomeAck, OutcomePoison: + msg.Ack() + default: + msg.Nack() + } + e.Outcome = outcome + e.Err = err + e.Duration = max(0, time.Since(start)-e.Wait) + + return e + } + + var env envelope + if err := json.Unmarshal(msg.Data, &env); err != nil { + return finish(OutcomePoison, fmt.Errorf("decoding message: %w", err)) + } + if env.CronTimestamp <= 0 { + return finish(OutcomePoison, errors.New("missing cronTimestamp")) + } + + if wait := time.Until(time.UnixMilli(env.CronTimestamp)); wait > 0 { + e.Wait = wait + timer := time.NewTimer(wait) + select { + case <-timer.C: + case <-ctx.Done(): + timer.Stop() + return finish(OutcomeDrain, ctx.Err()) + } + } + + select { + case slots <- struct{}{}: + case <-ctx.Done(): + return finish(OutcomeDrain, ctx.Err()) + } + + // The check outlives a shutdown signal: Drain bounds how long it may run on. + checkCtx, cancelCheck := context.WithCancel(context.WithoutCancel(ctx)) + defer cancelCheck() + done := make(chan error, 1) + go func() { + defer func() { <-slots }() + done <- c.dispatch(checkCtx, e.JobType, msg.Data) + }() + + var err error + select { + case err = <-done: + case <-ctx.Done(): + drain := time.NewTimer(c.Drain) + defer drain.Stop() + select { + case err = <-done: + case <-drain.C: + // Stop the check before the nack so a redelivery cannot run it + // alongside this one and record the result twice. + cancelCheck() + <-done + return finish(OutcomeDrain, errors.New("check still running at shutdown")) + } + } + + var invalid handlers.InvalidRequestError + switch { + case err == nil: + return finish(OutcomeAck, nil) + case errors.Is(err, errUnknownJobType), errors.As(err, &invalid): + return finish(OutcomePoison, err) + default: + return finish(OutcomeNack, err) + } +} + +var errUnknownJobType = errors.New("unknown job type") + +func (c *Consumer) dispatch(ctx context.Context, jobType string, data []byte) error { + switch jobType { + case "http": + var req request.HttpCheckerRequest + if err := json.Unmarshal(data, &req); err != nil { + return handlers.InvalidRequestError{Reason: err.Error()} + } + _, err := c.Runner.RunHTTPCheck(ctx, req) + return err + case "tcp": + var req request.TCPCheckerRequest + if err := json.Unmarshal(data, &req); err != nil { + return handlers.InvalidRequestError{Reason: err.Error()} + } + _, err := c.Runner.RunTCPCheck(ctx, req) + return err + case "dns": + var req request.DNSCheckerRequest + if err := json.Unmarshal(data, &req); err != nil { + return handlers.InvalidRequestError{Reason: err.Error()} + } + _, err := c.Runner.RunDNSCheck(ctx, req) + return err + case "icmp": + var req request.ICMPCheckerRequest + if err := json.Unmarshal(data, &req); err != nil { + return handlers.InvalidRequestError{Reason: err.Error()} + } + _, err := c.Runner.RunICMPCheck(ctx, req) + return err + case "grpc": + var req request.GRPCCheckerRequest + if err := json.Unmarshal(data, &req); err != nil { + return handlers.InvalidRequestError{Reason: err.Error()} + } + _, err := c.Runner.RunGRPCCheck(ctx, req) + return err + default: + return fmt.Errorf("%w: %q", errUnknownJobType, jobType) + } +} diff --git a/apps/checker/pkg/queue/consumer_test.go b/apps/checker/pkg/queue/consumer_test.go new file mode 100644 index 000000000..2ffca2e9d --- /dev/null +++ b/apps/checker/pkg/queue/consumer_test.go @@ -0,0 +1,278 @@ +package queue_test + +import ( + "context" + "encoding/json" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "cloud.google.com/go/pubsub/v2" + pb "cloud.google.com/go/pubsub/v2/apiv1/pubsubpb" + "cloud.google.com/go/pubsub/v2/pstest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/api/option" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/openstatushq/openstatus/apps/checker/checker" + "github.com/openstatushq/openstatus/apps/checker/handlers" + "github.com/openstatushq/openstatus/apps/checker/pkg/queue" + "github.com/openstatushq/openstatus/apps/checker/request" +) + +type fakeRunner struct { + run func(ctx context.Context, monitorID string) error +} + +func (f fakeRunner) RunHTTPCheck(ctx context.Context, req request.HttpCheckerRequest) (checker.Response, error) { + return checker.Response{}, f.run(ctx, req.MonitorID) +} + +func (f fakeRunner) RunTCPCheck(ctx context.Context, req request.TCPCheckerRequest) (checker.TCPResponse, error) { + return checker.TCPResponse{}, f.run(ctx, req.MonitorID) +} + +func (f fakeRunner) RunDNSCheck(ctx context.Context, req request.DNSCheckerRequest) (handlers.DNSResponse, error) { + return handlers.DNSResponse{}, f.run(ctx, req.MonitorID) +} + +func (f fakeRunner) RunICMPCheck(ctx context.Context, req request.ICMPCheckerRequest) (checker.ICMPResponse, error) { + return checker.ICMPResponse{}, f.run(ctx, req.MonitorID) +} + +func (f fakeRunner) RunGRPCCheck(ctx context.Context, req request.GRPCCheckerRequest) (checker.GRPCResponse, error) { + return checker.GRPCResponse{}, f.run(ctx, req.MonitorID) +} + +const ( + project = "projects/test" + subscription = "checker-local" +) + +type harness struct { + srv *pstest.Server + client *pubsub.Client + events chan queue.Event +} + +func newHarness(t *testing.T) *harness { + t.Helper() + ctx := context.Background() + srv := pstest.NewServer() + t.Cleanup(func() { srv.Close() }) + + _, err := srv.GServer.CreateTopic(ctx, &pb.Topic{Name: project + "/topics/" + subscription}) + require.NoError(t, err) + _, err = srv.GServer.CreateSubscription(ctx, &pb.Subscription{ + Name: project + "/subscriptions/" + subscription, + Topic: project + "/topics/" + subscription, + AckDeadlineSeconds: 10, + }) + require.NoError(t, err) + + conn, err := grpc.NewClient(srv.Addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { conn.Close() }) + + client, err := pubsub.NewClient(ctx, "test", option.WithGRPCConn(conn)) + require.NoError(t, err) + t.Cleanup(func() { client.Close() }) + + return &harness{srv: srv, client: client, events: make(chan queue.Event, 16)} +} + +func (h *harness) publish(t *testing.T, jobType string, monitorID string, cronTimestamp time.Time) string { + t.Helper() + data, err := json.Marshal(map[string]any{ + "monitorId": monitorID, + "workspaceId": "1", + "cronTimestamp": cronTimestamp.UnixMilli(), + }) + require.NoError(t, err) + + return h.srv.Publish(project+"/topics/"+subscription, data, map[string]string{ + "jobType": jobType, + "monitorId": monitorID, + }) +} + +func (h *harness) start(t *testing.T, runner queue.Runner, loop queue.Loop, drain time.Duration) (context.CancelFunc, <-chan error) { + t.Helper() + loop.Subscription = subscription + c := &queue.Consumer{ + Client: h.client, + Runner: runner, + Loops: []queue.Loop{loop}, + Drain: drain, + OnEvent: func(_ context.Context, e queue.Event) { + h.events <- e + }, + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- c.Run(ctx) }() + t.Cleanup(cancel) + + return cancel, done +} + +func (h *harness) next(t *testing.T) queue.Event { + t.Helper() + select { + case e := <-h.events: + return e + case <-time.After(10 * time.Second): + t.Fatal("no event") + return queue.Event{} + } +} + +func TestConsumer_AcksRecordedCheck(t *testing.T) { + h := newHarness(t) + id := h.publish(t, "http", "1", time.Now()) + h.start(t, fakeRunner{run: func(context.Context, string) error { return nil }}, queue.Loop{MaxOutstanding: 10}, time.Second) + + e := h.next(t) + + assert.Equal(t, queue.OutcomeAck, e.Outcome) + assert.Equal(t, "http", e.JobType) + assert.Equal(t, "1", e.MonitorID) + assert.Eventually(t, func() bool { return h.srv.Message(id).Acks == 1 }, 5*time.Second, 20*time.Millisecond) +} + +func TestConsumer_NacksRecordFailure(t *testing.T) { + h := newHarness(t) + id := h.publish(t, "tcp", "1", time.Now()) + var calls atomic.Int32 + h.start(t, fakeRunner{run: func(context.Context, string) error { + if calls.Add(1) == 1 { + return errors.New("tinybird down") + } + return nil + }}, queue.Loop{MaxOutstanding: 10}, time.Second) + + assert.Equal(t, queue.OutcomeNack, h.next(t).Outcome) + assert.Equal(t, queue.OutcomeAck, h.next(t).Outcome) + assert.GreaterOrEqual(t, h.srv.Message(id).Deliveries, 2) +} + +func TestConsumer_AcksPoisonWithoutRetrying(t *testing.T) { + h := newHarness(t) + unknown := h.publish(t, "smtp", "1", time.Now()) + invalid := h.publish(t, "dns", "2", time.Now()) + var calls atomic.Int32 + h.start(t, fakeRunner{run: func(context.Context, string) error { + calls.Add(1) + return handlers.InvalidRequestError{Reason: "invalid monitor id"} + }}, queue.Loop{MaxOutstanding: 10}, time.Second) + + assert.Equal(t, queue.OutcomePoison, h.next(t).Outcome) + assert.Equal(t, queue.OutcomePoison, h.next(t).Outcome) + assert.Eventually(t, func() bool { + return h.srv.Message(unknown).Acks == 1 && h.srv.Message(invalid).Acks == 1 + }, 5*time.Second, 20*time.Millisecond) + assert.Equal(t, int32(1), calls.Load()) +} + +func TestConsumer_WaitsForCronTimestamp(t *testing.T) { + h := newHarness(t) + at := time.Now().Add(500 * time.Millisecond) + h.publish(t, "icmp", "1", at) + var ranAt atomic.Int64 + h.start(t, fakeRunner{run: func(context.Context, string) error { + ranAt.Store(time.Now().UnixMilli()) + return nil + }}, queue.Loop{MaxOutstanding: 10}, time.Second) + + e := h.next(t) + + assert.Equal(t, queue.OutcomeAck, e.Outcome) + assert.Positive(t, e.Wait) + assert.GreaterOrEqual(t, ranAt.Load(), at.UnixMilli()) +} + +func TestConsumer_SleepingMessageHoldsNoProbeSlot(t *testing.T) { + h := newHarness(t) + h.publish(t, "http", "later", time.Now().Add(2*time.Second)) + h.publish(t, "http", "now", time.Now()) + var mu sync.Mutex + var order []string + h.start(t, fakeRunner{run: func(_ context.Context, monitorID string) error { + mu.Lock() + order = append(order, monitorID) + mu.Unlock() + return nil + }}, queue.Loop{MaxOutstanding: 10, MaxConcurrent: 1}, time.Second) + + first := h.next(t) + + assert.Equal(t, "now", first.MonitorID) + assert.Equal(t, "later", h.next(t).MonitorID) + mu.Lock() + defer mu.Unlock() + assert.Equal(t, []string{"now", "later"}, order) +} + +func TestConsumer_NacksCheckStillRunningAfterDrain(t *testing.T) { + h := newHarness(t) + h.publish(t, "grpc", "1", time.Now()) + started := make(chan struct{}) + var stopped atomic.Bool + cancel, done := h.start(t, fakeRunner{run: func(ctx context.Context, _ string) error { + close(started) + <-ctx.Done() + stopped.Store(true) + return ctx.Err() + }}, queue.Loop{MaxOutstanding: 10}, 100*time.Millisecond) + + <-started + cancel() + + e := h.next(t) + assert.Equal(t, queue.OutcomeDrain, e.Outcome) + assert.True(t, stopped.Load(), "check still running after its message was nacked") + select { + case err := <-done: + assert.NoError(t, err) + case <-time.After(10 * time.Second): + t.Fatal("consumer did not stop") + } +} + +func TestConsumer_AcksMissingCronTimestampAsPoison(t *testing.T) { + h := newHarness(t) + id := h.srv.Publish(project+"/topics/"+subscription, []byte(`{"monitorId":"1"}`), map[string]string{ + "jobType": "http", + "monitorId": "1", + }) + var calls atomic.Int32 + h.start(t, fakeRunner{run: func(context.Context, string) error { + calls.Add(1) + return nil + }}, queue.Loop{MaxOutstanding: 10}, time.Second) + + assert.Equal(t, queue.OutcomePoison, h.next(t).Outcome) + assert.Eventually(t, func() bool { return h.srv.Message(id).Acks == 1 }, 5*time.Second, 20*time.Millisecond) + assert.Equal(t, int32(0), calls.Load()) +} + +func TestConsumer_RejectsNonPositiveMaxOutstanding(t *testing.T) { + h := newHarness(t) + _, done := h.start(t, fakeRunner{run: func(context.Context, string) error { return nil }}, queue.Loop{}, time.Second) + + select { + case err := <-done: + assert.ErrorContains(t, err, "must be positive") + case <-time.After(5 * time.Second): + t.Fatal("consumer did not fail") + } +} + +func TestSubscription(t *testing.T) { + assert.Equal(t, "checker-ams", queue.Subscription("ams", false)) + assert.Equal(t, "checker-koyeb_fra-30s", queue.Subscription("koyeb_fra", true)) +} diff --git a/apps/checker/scripts/pubsub-gate/main.go b/apps/checker/scripts/pubsub-gate/main.go new file mode 100644 index 000000000..a243d7b6e --- /dev/null +++ b/apps/checker/scripts/pubsub-gate/main.go @@ -0,0 +1,301 @@ +// Command pubsub-gate is the go/no-go check before routing cron checks through +// Pub/Sub: every checker subscription must have a live consumer, and one +// synthetic check published to each topic must land in Tinybird. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "net/http" + "net/url" + "os" + "sort" + "strings" + "text/tabwriter" + "time" + + monitoring "cloud.google.com/go/monitoring/apiv3/v2" + "cloud.google.com/go/monitoring/apiv3/v2/monitoringpb" + "cloud.google.com/go/pubsub/v2" + "google.golang.org/api/iterator" + "google.golang.org/api/option" + "google.golang.org/protobuf/types/known/timestamppb" + + "github.com/openstatushq/openstatus/apps/checker/pkg/gcp" + "github.com/openstatushq/openstatus/apps/checker/pkg/queue" +) + +type target struct { + region string + subscription string + cronTimestamp int64 + pulls int64 + latency time.Duration + landed bool +} + +func main() { + project := flag.String("project", "openstatus-402607", "GCP project id") + regionsFile := flag.String("regions", "../workflows/src/cron/checker-regions.json", "checker regions JSON") + monitorID := flag.String("monitor-id", "", "id of the inactive pubsub-gate monitor") + workspaceID := flag.String("workspace-id", "", "workspace of the pubsub-gate monitor") + probeURL := flag.String("url", "https://www.openstatus.dev/api/ping", "URL the synthetic check probes") + timeout := flag.Duration("timeout", 3*time.Minute, "how long to wait for every Tinybird row") + flag.Parse() + + if *monitorID == "" || *workspaceID == "" { + fmt.Fprintln(os.Stderr, "-monitor-id and -workspace-id are required") + os.Exit(2) + } + tinybirdToken := os.Getenv("TINYBIRD_TOKEN") + if tinybirdToken == "" { + fmt.Fprintln(os.Stderr, "TINYBIRD_TOKEN (with read access to ping_response__v8) is required") + os.Exit(2) + } + + regions, err := readRegions(*regionsFile) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(2) + } + + ctx := context.Background() + credentials, err := gcp.Credentials() + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(2) + } + creds := option.WithAuthCredentials(credentials) + + base := time.Now().UnixMilli() + targets := make([]*target, 0, len(regions)*2) + for _, region := range regions { + for i, thirty := range []bool{false, true} { + targets = append(targets, &target{ + region: region, + subscription: queue.Subscription(region, thirty), + cronTimestamp: base + int64(i), + }) + } + } + + if err := countStreamingPulls(ctx, *project, creds, targets); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + + if err := publishSynthetic(ctx, *project, creds, targets, *monitorID, *workspaceID, *probeURL); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + + waitForRows(ctx, tinybirdToken, *monitorID, base, targets, *timeout) + + if !report(targets) { + os.Exit(1) + } +} + +func readRegions(path string) ([]string, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("reading regions: %w", err) + } + var regions []string + if err := json.Unmarshal(raw, ®ions); err != nil { + return nil, fmt.Errorf("decoding regions: %w", err) + } + if len(regions) == 0 { + return nil, fmt.Errorf("no regions in %s", path) + } + + return regions, nil +} + +func countStreamingPulls(ctx context.Context, project string, creds option.ClientOption, targets []*target) error { + client, err := monitoring.NewMetricClient(ctx, creds) + if err != nil { + return fmt.Errorf("creating monitoring client: %w", err) + } + defer client.Close() + + now := time.Now() + it := client.ListTimeSeries(ctx, &monitoringpb.ListTimeSeriesRequest{ + Name: "projects/" + project, + Filter: `metric.type = "pubsub.googleapis.com/subscription/open_streaming_pulls" AND resource.labels.subscription_id = starts_with("checker-")`, + Interval: &monitoringpb.TimeInterval{ + StartTime: timestamppb.New(now.Add(-5 * time.Minute)), + EndTime: timestamppb.New(now), + }, + View: monitoringpb.ListTimeSeriesRequest_FULL, + }) + + pulls := map[string]int64{} + for { + ts, err := it.Next() + if errors.Is(err, iterator.Done) { + break + } + if err != nil { + return fmt.Errorf("listing open_streaming_pulls: %w", err) + } + if ts.Metric.Labels["status"] != "ok" || len(ts.Points) == 0 { + continue + } + pulls[ts.Resource.Labels["subscription_id"]] += ts.Points[0].Value.GetInt64Value() + } + + for _, t := range targets { + t.pulls = pulls[t.subscription] + } + + return nil +} + +func publishSynthetic(ctx context.Context, project string, creds option.ClientOption, targets []*target, monitorID, workspaceID, probeURL string) error { + client, err := pubsub.NewClient(ctx, project, creds) + if err != nil { + return fmt.Errorf("creating pubsub client: %w", err) + } + defer client.Close() + + results := make([]*pubsub.PublishResult, len(targets)) + for i, t := range targets { + data, err := json.Marshal(map[string]any{ + "workspaceId": workspaceID, + "monitorId": monitorID, + "url": probeURL, + "method": "GET", + "cronTimestamp": t.cronTimestamp, + "body": "", + "headers": []string{}, + "status": "active", + "assertions": nil, + "degradedAfter": nil, + "timeout": 45000, + "trigger": "cron", + "retry": 1, + "followRedirects": true, + }) + if err != nil { + return fmt.Errorf("encoding payload: %w", err) + } + publisher := client.Publisher(t.subscription) + defer publisher.Stop() + results[i] = publisher.Publish(ctx, &pubsub.Message{ + Data: data, + Attributes: map[string]string{ + "jobType": "http", + "periodicity": "gate", + "monitorId": monitorID, + "region": t.region, + }, + }) + } + + for i, r := range results { + if _, err := r.Get(ctx); err != nil { + return fmt.Errorf("publishing to %s: %w", targets[i].subscription, err) + } + } + + return nil +} + +func waitForRows(ctx context.Context, token, monitorID string, since int64, targets []*target, timeout time.Duration) { + byKey := map[string]*target{} + for _, t := range targets { + byKey[fmt.Sprintf("%s/%d", t.region, t.cronTimestamp)] = t + } + + query := fmt.Sprintf( + "SELECT region, cronTimestamp, min(timestamp) AS at FROM ping_response__v8 WHERE monitorId = '%s' AND cronTimestamp >= %d GROUP BY region, cronTimestamp FORMAT JSON", + strings.ReplaceAll(monitorID, "'", ""), since, + ) + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + rows, err := tinybirdRows(ctx, token, query) + if err != nil { + fmt.Fprintln(os.Stderr, err) + } + pending := 0 + for _, row := range rows { + if t, ok := byKey[fmt.Sprintf("%s/%d", row.Region, row.CronTimestamp)]; ok && !t.landed { + t.landed = true + t.latency = time.Duration(row.At-row.CronTimestamp) * time.Millisecond + } + } + for _, t := range targets { + if !t.landed { + pending++ + } + } + if pending == 0 { + return + } + time.Sleep(5 * time.Second) + } +} + +type tinybirdRow struct { + Region string `json:"region"` + CronTimestamp int64 `json:"cronTimestamp"` + At int64 `json:"at"` +} + +func tinybirdRows(ctx context.Context, token, query string) ([]tinybirdRow, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://api.tinybird.co/v0/sql?q="+url.QueryEscape(query), nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+token) + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, fmt.Errorf("querying tinybird: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("querying tinybird: status %d", resp.StatusCode) + } + + var body struct { + Data []tinybirdRow `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + return nil, fmt.Errorf("decoding tinybird response: %w", err) + } + + return body.Data, nil +} + +func report(targets []*target) bool { + sort.Slice(targets, func(i, j int) bool { return targets[i].subscription < targets[j].subscription }) + w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintln(w, "subscription\tpulls\tsynthetic\t") + ok := true + for _, t := range targets { + synthetic := "MISSING" + if t.landed { + synthetic = fmt.Sprintf("ok %s", t.latency.Round(time.Millisecond)) + } + verdict := "" + if t.pulls < 1 || !t.landed { + verdict = "<- no-go" + ok = false + } + fmt.Fprintf(w, "%s\t%d\t%s\t%s\n", t.subscription, t.pulls, synthetic, verdict) + } + w.Flush() + + if ok { + fmt.Println("GO: every subscription has a consumer and delivered its synthetic check") + } else { + fmt.Println("NO-GO") + } + + return ok +} -- 2.54.0