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 +}