From b56086d8df5bfe41336597fca0d805ad15159341 Mon Sep 17 00:00:00 2001 From: Alec Thomas Date: Wed, 28 Oct 2026 19:42:02 +1100 Subject: [PATCH 1/2] feat(server): store divergences in Redis The SPECTRE server kept divergences only in memory, so a restart lost every reported quarantine and replicas could not share them. A storage interface keeps the in-memory behaviour as the default and lets the server use Redis when divergences must persist. Co-Authored-By: Claude Opus 5.5 --- CONTRIBUTING.md | 2 + cli_internal_test.go | 9 +- go.mod | 7 +- go.sum | 30 +++-- internal/cli/divergences_internal_test.go | 3 +- internal/server/config.go | 10 +- internal/server/divergences.go | 60 +++++++++ internal/server/server_test.go | 9 +- internal/server/service.go | 42 ++++-- internal/server/storage/memory.go | 66 ++++++++++ internal/server/storage/redis.go | 110 ++++++++++++++++ internal/server/storage/storage.go | 76 +++++++++++ internal/server/storage/storage_test.go | 151 ++++++++++++++++++++++ internal/server/store.go | 103 --------------- 14 files changed, 542 insertions(+), 136 deletions(-) create mode 100644 internal/server/divergences.go create mode 100644 internal/server/storage/memory.go create mode 100644 internal/server/storage/redis.go create mode 100644 internal/server/storage/storage.go create mode 100644 internal/server/storage/storage_test.go delete mode 100644 internal/server/store.go diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1df7eeb..39a64ff 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -66,6 +66,8 @@ and `docker.io/blockossreleases/spectre`. `spectre serve` collects the quarantines that mirrors report. It listens on `127.0.0.1:50070` by default and keeps them in memory, so a restart loses them. +To keep them in Redis instead, pass `--storage=redis` and `--redis-url`, such as +`redis://localhost:6379/0`. Every mirror reports to a server, so `spectre supervise` requires its URL, `--server-url`, and the mirror's identity, `--service`, `--environment`, and diff --git a/cli_internal_test.go b/cli_internal_test.go index 21059e0..7dcd98e 100644 --- a/cli_internal_test.go +++ b/cli_internal_test.go @@ -14,6 +14,7 @@ import ( "github.com/block/spectre/internal/cli" "github.com/block/spectre/internal/server" "github.com/block/spectre/internal/server/client" + "github.com/block/spectre/internal/server/storage" "github.com/block/spectre/internal/supervisor" ) @@ -122,9 +123,13 @@ func TestServeCommandEmbedsConfig(t *testing.T) { command := &cli.CommandLine{} parser, err := newParser(command) assert.NoError(t, err) - _, err = parser.Parse([]string{"serve", "--listen=127.0.0.1:50071"}) + _, err = parser.Parse([]string{"serve", "--listen=127.0.0.1:50071", "--storage=redis", "--redis-url=redis://localhost:6379/0"}) assert.NoError(t, err) - assert.Equal(t, server.ServeConfig{Listen: "127.0.0.1:50071"}, command.Serve.Serve) + expected := server.ServeConfig{ + Listen: "127.0.0.1:50071", + Storage: storage.Config{Storage: storage.BackendRedis, RedisURL: "redis://localhost:6379/0"}, + } + assert.Equal(t, expected, command.Serve.Serve) } func TestSuperviseCommandRequiresFlags(t *testing.T) { diff --git a/go.mod b/go.mod index 067e87c..3bb1f6d 100644 --- a/go.mod +++ b/go.mod @@ -13,12 +13,14 @@ require ( github.com/alecthomas/kong v1.16.1 github.com/alecthomas/kong-toml v0.4.1 github.com/alecthomas/types v0.20.1 + github.com/alicebob/miniredis/v2 v2.40.0 github.com/block/scaffolder v1.12.0 github.com/gobwas/glob v1.0.0 github.com/grafana/sobek v0.0.0-20260915160442-8a431c44cd7b github.com/lmittmann/tint v1.2.0 github.com/microsoft/TypeScript/tsc/shim/typescript v0.0.0 github.com/pelletier/go-toml v1.9.5 + github.com/redis/go-redis/v9 v9.23.0 github.com/santhosh-tekuri/jsonschema/v6 v6.0.3 go.jetify.com/typeid v1.3.0 golang.org/x/mod v0.41.0 @@ -32,6 +34,7 @@ require ( github.com/Microsoft/go-winio v0.6.2 // indirect github.com/alecthomas/repr v0.5.2 // indirect github.com/block/getit v0.1.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/dlclark/regexp2 v1.11.4 // indirect github.com/dlclark/regexp2/v2 v2.8.0 // indirect github.com/dop251/goja v0.0.0-20260311135729-065cd970411c // indirect @@ -40,9 +43,11 @@ require ( github.com/google/pprof v0.0.0-20230207041349-798e818bf904 // indirect github.com/hexops/gotextdiff v1.0.3 // indirect github.com/klauspost/compress v1.20.0 // indirect - github.com/klauspost/cpuid/v2 v2.2.10 // indirect + github.com/klauspost/cpuid/v2 v2.4.0 // indirect github.com/microsoft/TypeScript/tsc v0.0.0-20261001235638-09b1db061731 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect github.com/zeebo/xxh3 v1.1.0 // indirect + go.uber.org/atomic v1.12.0 // indirect golang.org/x/text v0.42.0 // indirect ) diff --git a/go.sum b/go.sum index 6c41843..fc3a3f7 100644 --- a/go.sum +++ b/go.sum @@ -18,12 +18,18 @@ github.com/alecthomas/repr v0.5.2 h1:SU73FTI9D1P5UNtvseffFSGmdNci/O6RsqzeXJtP0Qs github.com/alecthomas/repr v0.5.2/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4= github.com/alecthomas/types v0.20.1 h1:NwFTHt5VQSOa+7Q8ZgdZIZY9SgnbxZxSIut2RuK7xOE= github.com/alecthomas/types v0.20.1/go.mod h1:FPpx/TFcz1NP/fsbg3iy4mnEFfw0YOEuhXYwZQSxL5I= +github.com/alicebob/miniredis/v2 v2.40.0 h1:6IgTQbvm2JiMYZ8yur/acNJzVjtSOWCelxXbk/LUBXc= +github.com/alicebob/miniredis/v2 v2.40.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/block/getit v0.1.0 h1:14J3b9xL9uKn6GIctLPr3MnSKAUXzFHTr7RDjwizLBU= github.com/block/getit v0.1.0/go.mod h1:wldKOgqqKEfPvEQMM5DgRJqWU1ob6n940/GDFFOft+4= github.com/block/scaffolder v1.12.0 h1:VTKV3/yxd1O7Blpx9EsZwHi2VtzoLhpD2Fkv3xu57nI= github.com/block/scaffolder v1.12.0/go.mod h1:TmpCQ6cBwO29/TWCdPYeZ0nkefccHZeo19ZgMX5FzUk= -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/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +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/dlclark/regexp2 v1.11.4 h1:rPYF9/LECdNymJufQKmri9gV604RvvABwgOA8un7yAo= github.com/dlclark/regexp2 v1.11.4/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/dlclark/regexp2/v2 v2.8.0 h1:CekDhPLGfm+GAJmPVFIG+5dqMIQPkyHJll7BbdneDfw= @@ -52,8 +58,8 @@ github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUq github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg= github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= -github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= -github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/klauspost/cpuid/v2 v2.4.0 h1:S6Hrbc7+ywsr0r+RLapfGBHfyefhCTwEh3A0tV913Dw= +github.com/klauspost/cpuid/v2 v2.4.0/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU= github.com/lmittmann/tint v1.2.0 h1:AogHRHy8HUJUnNJBHJlYa+fR4YY8mko2cnCp67xn9JY= github.com/lmittmann/tint v1.2.0/go.mod h1:HIS3gSy7qNwGCj+5oRjAutErFBl4BzdQP6cJZ0NfMwE= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= @@ -66,20 +72,26 @@ github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3v github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c= github.com/peter-evans/patience v0.3.0 h1:rX0JdJeepqdQl1Sk9c9uvorjYYzL2TfgLX1adqYm9cA= github.com/peter-evans/patience v0.3.0/go.mod h1:Kmxu5sY1NmBLFSStvXjX1wS9mIv7wMcP/ubucyMOAu0= -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/redis/go-redis/v9 v9.23.0 h1:/TTVdJa8BRNsybwwGGnmiYktHNlUXtbOQxlWpHBPL3s= +github.com/redis/go-redis/v9 v9.23.0/go.mod h1:EWP2UUk+XDaI1s5nQB5mablZN6vccXlmBQTyfnF87D0= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/santhosh-tekuri/jsonschema/v6 v6.0.3 h1:1EYB5IzjZawrrnELUi78f9fPu57HuXjmddZPjrls/28= github.com/santhosh-tekuri/jsonschema/v6 v6.0.3/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU= -github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= -github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ= github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.jetify.com/typeid v1.3.0 h1:fuWV7oxO4mSsgpxwhaVpFXgt0IfjogR29p+XAjDCVKY= go.jetify.com/typeid v1.3.0/go.mod h1:CtVGyt2+TSp4Rq5+ARLvGsJqdNypKBAC6INQ9TLPlmk= +go.uber.org/atomic v1.12.0 h1:BvcXdFKuviU4fTL/f+SxdQ5qJX/Jix8pAkgdUcb3XOE= +go.uber.org/atomic v1.12.0/go.mod h1:I6c4cg+6HCxRjfjSsYtApoFILnpc0CGUdGkXVqbYVNk= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b h1:M2rDM6z3Fhozi9O7NWsxAkg/yqS/lQJ6PmkyIV3YP+o= golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8= golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= @@ -96,8 +108,6 @@ google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fw google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= modernc.org/libc v1.66.3 h1:cfCbjTUcdsKyyZZfEUKfoHcP3S0Wkvz3jgSzByEWVCQ= diff --git a/internal/cli/divergences_internal_test.go b/internal/cli/divergences_internal_test.go index 559917a..0113440 100644 --- a/internal/cli/divergences_internal_test.go +++ b/internal/cli/divergences_internal_test.go @@ -14,12 +14,13 @@ import ( "github.com/block/spectre/internal/server" "github.com/block/spectre/internal/server/client" + "github.com/block/spectre/internal/server/storage" pb "github.com/block/spectre/spectrepb" "github.com/block/spectre/spectrepb/spectrepbconnect" ) func TestDivergencesListCommand(t *testing.T) { - httpServer := httptest.NewServer(server.NewService().Handler()) + httpServer := httptest.NewServer(server.NewService(storage.NewMemory()).Handler()) t.Cleanup(httpServer.Close) rpc := spectrepbconnect.NewSpectreServiceClient(httpServer.Client(), httpServer.URL) first := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) diff --git a/internal/server/config.go b/internal/server/config.go index 7cb74c4..f99b261 100644 --- a/internal/server/config.go +++ b/internal/server/config.go @@ -1,8 +1,11 @@ // Package server is the SPECTRE server, which collects the quarantines that -// mirrors report. It keeps them in memory, so a restart loses them. +// mirrors report. It keeps them in a storage backend, in memory or in Redis. package server -import "github.com/block/spectre/internal/server/client" +import ( + "github.com/block/spectre/internal/server/client" + "github.com/block/spectre/internal/server/storage" +) // ReporterConfig configures how a mirror reports quarantines to the SPECTRE server. type ReporterConfig struct { @@ -14,7 +17,8 @@ type ReporterConfig struct { // ServeConfig configures the SPECTRE server. type ServeConfig struct { - Listen string `default:"127.0.0.1:50070" help:"Address for the SPECTRE server: host:port, or unix:."` + Listen string `default:"127.0.0.1:50070" help:"Address for the SPECTRE server: host:port, or unix:."` + Storage storage.Config `embed:""` } // ClearConfig names the service environment whose divergences to clear. diff --git a/internal/server/divergences.go b/internal/server/divergences.go new file mode 100644 index 0000000..5d47c77 --- /dev/null +++ b/internal/server/divergences.go @@ -0,0 +1,60 @@ +package server + +import ( + "cmp" + "slices" + + "github.com/alecthomas/errors" + . "github.com/alecthomas/types/optional" + + pb "github.com/block/spectre/spectrepb" +) + +// validateDivergence checks that a reported divergence identifies its source +// and quarantine, so every store can rely on them. +func validateDivergence(divergence *pb.Divergence) error { + if divergence.GetService() == "" || divergence.GetEnvironment() == "" || divergence.GetRevision() == "" { + return errors.New("divergence needs a service, environment, and revision") + } + if _, err := DirectionFromMessage(divergence.GetDirection()); err != nil { + return err + } + quarantine := divergence.GetQuarantine() + if quarantine.GetSince() == nil { + return errors.New("divergence needs a quarantine with its start time") + } + if _, err := ReasonFromMessage(quarantine); err != nil { + return errors.Wrap(err, "quarantine") + } + return nil +} + +// filterDivergences returns the divergences that match the request, ordered by source. +func filterDivergences(divergences []*pb.Divergence, request *pb.ListDivergencesRequest) []*pb.Divergence { + matching := []*pb.Divergence{} + for _, divergence := range divergences { + quarantine := divergence.GetQuarantine() + // Getters hide whether an optional field is set, so presence is a nil check. + if matches(From(request.GetService(), request.Service != nil), Some(divergence.GetService())) && + matches(From(request.GetEnvironment(), request.Environment != nil), Some(divergence.GetEnvironment())) && + matches(From(request.GetRevision(), request.Revision != nil), Some(divergence.GetRevision())) && + matches(From(request.GetNode(), request.Node != nil), From(quarantine.GetNode(), quarantine.Node != nil)) { + matching = append(matching, divergence) + } + } + slices.SortFunc(matching, func(a, b *pb.Divergence) int { + return cmp.Or( + cmp.Compare(a.GetService(), b.GetService()), + cmp.Compare(a.GetEnvironment(), b.GetEnvironment()), + cmp.Compare(a.GetDirection(), b.GetDirection()), + ) + }) + return matching +} + +// matches reports whether value passes a filter. An unset filter matches +// every value, including an absent one. +func matches(filter, value Option[string]) bool { + expected, ok := filter.Get() + return !ok || value == Some(expected) +} diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 944232d..da6cf2c 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -17,12 +17,13 @@ import ( "github.com/block/spectre/internal/model" "github.com/block/spectre/internal/server" "github.com/block/spectre/internal/server/client" + "github.com/block/spectre/internal/server/storage" pb "github.com/block/spectre/spectrepb" "github.com/block/spectre/spectrepb/spectrepbconnect" ) func TestDivergencesKeepLatestQuarantine(t *testing.T) { - httpServer := httptest.NewServer(server.NewService().Handler()) + httpServer := httptest.NewServer(server.NewService(storage.NewMemory()).Handler()) t.Cleanup(httpServer.Close) rpc := spectrepbconnect.NewSpectreServiceClient(httpServer.Client(), httpServer.URL) first := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) @@ -76,7 +77,7 @@ func TestDivergencesKeepLatestQuarantine(t *testing.T) { } func TestClearDivergencesOfServiceEnvironment(t *testing.T) { - httpServer := httptest.NewServer(server.NewService().Handler()) + httpServer := httptest.NewServer(server.NewService(storage.NewMemory()).Handler()) t.Cleanup(httpServer.Close) rpc := spectrepbconnect.NewSpectreServiceClient(httpServer.Client(), httpServer.URL) divergence := func(environment string, direction pb.Direction) *pb.Divergence { @@ -109,7 +110,7 @@ func TestClearDivergencesOfServiceEnvironment(t *testing.T) { } func TestQuarantineRejectsInvalidDivergence(t *testing.T) { - httpServer := httptest.NewServer(server.NewService().Handler()) + httpServer := httptest.NewServer(server.NewService(storage.NewMemory()).Handler()) t.Cleanup(httpServer.Close) rpc := spectrepbconnect.NewSpectreServiceClient(httpServer.Client(), httpServer.URL) valid := func() *pb.Divergence { @@ -152,7 +153,7 @@ func TestQuarantineRejectsInvalidDivergence(t *testing.T) { func TestReporterRetriesQuarantine(t *testing.T) { reports := make(chan struct{}, 8) var attempts atomic.Int32 - service := server.NewService().Handler() + service := server.NewService(storage.NewMemory()).Handler() httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { if request.URL.Path == spectrepbconnect.SpectreServiceQuarantineProcedure && attempts.Add(1) == 1 { http.Error(w, "unavailable", http.StatusServiceUnavailable) diff --git a/internal/server/service.go b/internal/server/service.go index 249458a..e6a59b3 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -13,18 +13,19 @@ import ( "github.com/block/spectre/internal/httpserver/middleware/logging" "github.com/block/spectre/internal/logger" "github.com/block/spectre/internal/netaddr" + "github.com/block/spectre/internal/server/storage" pb "github.com/block/spectre/spectrepb" "github.com/block/spectre/spectrepb/spectrepbconnect" ) // Service serves the SPECTRE service API. type Service struct { - store *store + store storage.Store } -// NewService returns a SPECTRE server with no divergences. -func NewService() *Service { - return &Service{store: newStore()} +// NewService returns a SPECTRE server that keeps divergences in store. +func NewService(store storage.Store) *Service { + return &Service{store: store} } // Handler serves the API. @@ -35,22 +36,34 @@ func (s *Service) Handler() http.Handler { } // Quarantine records that a mirror quarantined its candidate. -func (s *Service) Quarantine(_ context.Context, request *connect.Request[pb.QuarantineRequest]) (*connect.Response[pb.QuarantineResponse], error) { - if err := s.store.record(request.Msg.GetDivergence()); err != nil { +func (s *Service) Quarantine(ctx context.Context, request *connect.Request[pb.QuarantineRequest]) (*connect.Response[pb.QuarantineResponse], error) { + divergence := request.Msg.GetDivergence() + if err := validateDivergence(divergence); err != nil { return nil, connect.NewError(connect.CodeInvalidArgument, err) } + if err := s.store.Record(ctx, divergence); err != nil { + return nil, connect.NewError(connect.CodeUnavailable, err) + } return connect.NewResponse(&pb.QuarantineResponse{}), nil } // ListDivergences returns the latest quarantine of each mirror direction. -func (s *Service) ListDivergences(_ context.Context, request *connect.Request[pb.ListDivergencesRequest]) (*connect.Response[pb.ListDivergencesResponse], error) { - return connect.NewResponse(&pb.ListDivergencesResponse{Divergences: s.store.listDivergences(request.Msg)}), nil +func (s *Service) ListDivergences(ctx context.Context, request *connect.Request[pb.ListDivergencesRequest]) (*connect.Response[pb.ListDivergencesResponse], error) { + divergences, err := s.store.List(ctx) + if err != nil { + return nil, connect.NewError(connect.CodeUnavailable, err) + } + return connect.NewResponse(&pb.ListDivergencesResponse{Divergences: filterDivergences(divergences, request.Msg)}), nil } // ClearDivergences removes the divergences of one service environment. -func (s *Service) ClearDivergences(_ context.Context, request *connect.Request[pb.ClearDivergencesRequest]) (*connect.Response[pb.ClearDivergencesResponse], error) { - if err := s.store.clear(request.Msg.GetService(), request.Msg.GetEnvironment()); err != nil { - return nil, connect.NewError(connect.CodeInvalidArgument, err) +func (s *Service) ClearDivergences(ctx context.Context, request *connect.Request[pb.ClearDivergencesRequest]) (*connect.Response[pb.ClearDivergencesResponse], error) { + service, environment := request.Msg.GetService(), request.Msg.GetEnvironment() + if service == "" || environment == "" { + return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("clearing divergences needs a service and environment")) + } + if err := s.store.Clear(ctx, service, environment); err != nil { + return nil, connect.NewError(connect.CodeUnavailable, err) } return connect.NewResponse(&pb.ClearDivergencesResponse{}), nil } @@ -62,7 +75,12 @@ func Serve(ctx context.Context, config ServeConfig) error { if err != nil { return errors.Wrap(err, "listen for SPECTRE server requests") } - handler := health.New(logging.New(NewService().Handler(), logger.EventReceived)) + store, err := storage.Open(ctx, config.Storage) + if err != nil { + return errors.Wrap(err, "open divergence storage") + } + defer store.Close() + handler := health.New(logging.New(NewService(store).Handler(), logger.EventReceived)) protocols := new(http.Protocols) protocols.SetHTTP1(true) protocols.SetHTTP2(true) diff --git a/internal/server/storage/memory.go b/internal/server/storage/memory.go new file mode 100644 index 0000000..3d02722 --- /dev/null +++ b/internal/server/storage/memory.go @@ -0,0 +1,66 @@ +package storage + +import ( + "context" + "sync" + + pb "github.com/block/spectre/spectrepb" +) + +// Memory keeps divergences in memory, so a restart loses them. Stored messages +// are never mutated, so they may be shared. +type Memory struct { + mu sync.Mutex + divergences map[sourceKey]*pb.Divergence +} + +var _ Store = (*Memory)(nil) + +// sourceKey identifies the proxy of one direction of a service environment's mirror. +type sourceKey struct { + service, environment string + direction pb.Direction +} + +// NewMemory returns an empty in-memory store. +func NewMemory() *Memory { + return &Memory{divergences: map[sourceKey]*pb.Divergence{}} +} + +// Record keeps a divergence unless its source has already reported a later quarantine. +func (m *Memory) Record(_ context.Context, divergence *pb.Divergence) error { + key := sourceKey{service: divergence.GetService(), environment: divergence.GetEnvironment(), direction: divergence.GetDirection()} + m.mu.Lock() + defer m.mu.Unlock() + if existing, ok := m.divergences[key]; ok && quarantinedAfter(existing, divergence) { + return nil + } + m.divergences[key] = divergence + return nil +} + +// Clear removes the divergences of both directions of a service environment's mirror. +func (m *Memory) Clear(_ context.Context, service, environment string) error { + m.mu.Lock() + defer m.mu.Unlock() + for key := range m.divergences { + if key.service == service && key.environment == environment { + delete(m.divergences, key) + } + } + return nil +} + +// List returns every stored divergence, in no particular order. +func (m *Memory) List(_ context.Context) ([]*pb.Divergence, error) { + m.mu.Lock() + defer m.mu.Unlock() + divergences := make([]*pb.Divergence, 0, len(m.divergences)) + for _, divergence := range m.divergences { + divergences = append(divergences, divergence) + } + return divergences, nil +} + +// Close does nothing, because memory holds no resources. +func (m *Memory) Close() error { return nil } diff --git a/internal/server/storage/redis.go b/internal/server/storage/redis.go new file mode 100644 index 0000000..8e0f718 --- /dev/null +++ b/internal/server/storage/redis.go @@ -0,0 +1,110 @@ +package storage + +import ( + "context" + "strconv" + + "github.com/alecthomas/errors" + "github.com/redis/go-redis/v9" + "google.golang.org/protobuf/proto" + + pb "github.com/block/spectre/spectrepb" +) + +// divergencesKey is the Redis hash that maps each sourceField to its divergence. +const divergencesKey = "spectre:divergences" + +// Redis keeps divergences in a Redis hash, so they outlive the SPECTRE server +// and are shared by every server using the same Redis database. +type Redis struct { + client *redis.Client +} + +var _ Store = (*Redis)(nil) + +// NewRedis connects to the Redis server the configuration names. +func NewRedis(ctx context.Context, config Config) (*Redis, error) { + options, err := redis.ParseURL(config.RedisURL) + if err != nil { + return nil, errors.Wrap(err, "redis URL") + } + client := redis.NewClient(options) + if err := client.Ping(ctx).Err(); err != nil { + _ = client.Close() //nolint:errcheck // The connection failed, so closing has nothing to report. + return nil, errors.Wrap(err, "connect to Redis") + } + return &Redis{client: client}, nil +} + +// Record keeps a divergence unless its source has already reported a later quarantine. +func (r *Redis) Record(ctx context.Context, divergence *pb.Divergence) error { + field := sourceField(divergence.GetService(), divergence.GetEnvironment(), divergence.GetDirection()) + value, err := proto.Marshal(divergence) + if err != nil { + return errors.Wrap(err, "encode divergence") + } + // The comparison happens in Go, so watch the hash and retry whenever another + // write lands between reading the stored divergence and replacing it. + for { + err := r.client.Watch(ctx, func(tx *redis.Tx) error { + existing, err := tx.HGet(ctx, divergencesKey, field).Bytes() + switch { + case errors.Is(err, redis.Nil): + case err != nil: + return errors.Wrap(err, "read divergence") + default: + stored := &pb.Divergence{} + if err := proto.Unmarshal(existing, stored); err != nil { + return errors.Wrap(err, "decode divergence") + } + if quarantinedAfter(stored, divergence) { + return nil + } + } + _, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error { + return errors.WithStack(pipe.HSet(ctx, divergencesKey, field, value).Err()) + }) + return errors.WithStack(err) + }, divergencesKey) + if !errors.Is(err, redis.TxFailedErr) { + return errors.Wrap(err, "record divergence") + } + } +} + +// Clear removes the divergences of both directions of a service environment's mirror. +func (r *Redis) Clear(ctx context.Context, service, environment string) error { + err := r.client.HDel(ctx, divergencesKey, + sourceField(service, environment, pb.Direction_DIRECTION_INGRESS), + sourceField(service, environment, pb.Direction_DIRECTION_EGRESS), + ).Err() + return errors.Wrap(err, "clear divergences") +} + +// List returns every stored divergence, in no particular order. +func (r *Redis) List(ctx context.Context) ([]*pb.Divergence, error) { + values, err := r.client.HVals(ctx, divergencesKey).Result() + if err != nil { + return nil, errors.Wrap(err, "list divergences") + } + divergences := make([]*pb.Divergence, 0, len(values)) + for _, value := range values { + divergence := &pb.Divergence{} + if err := proto.Unmarshal([]byte(value), divergence); err != nil { + return nil, errors.Wrap(err, "decode divergence") + } + divergences = append(divergences, divergence) + } + return divergences, nil +} + +// Close disconnects from Redis. +func (r *Redis) Close() error { + return errors.Wrap(r.client.Close(), "close Redis client") +} + +// sourceField identifies the proxy of one direction of a service environment's +// mirror. Quoting keeps the parts unambiguous whatever characters they hold. +func sourceField(service, environment string, direction pb.Direction) string { + return strconv.Quote(service) + " " + strconv.Quote(environment) + " " + direction.String() +} diff --git a/internal/server/storage/storage.go b/internal/server/storage/storage.go new file mode 100644 index 0000000..af4df08 --- /dev/null +++ b/internal/server/storage/storage.go @@ -0,0 +1,76 @@ +// Package storage keeps the divergences that mirrors report to the SPECTRE +// server, either in memory or in Redis. +package storage + +import ( + "context" + + "github.com/alecthomas/errors" + "github.com/redis/go-redis/v9" + + pb "github.com/block/spectre/spectrepb" +) + +// Backend names where the SPECTRE server keeps divergences. +type Backend string + +// Backends the SPECTRE server can keep divergences in. +const ( + BackendMemory Backend = "memory" + BackendRedis Backend = "redis" +) + +// Config selects where the SPECTRE server keeps divergences. +type Config struct { + Storage Backend `default:"memory" enum:"memory,redis" help:"Where to keep divergences: memory, which a restart loses, or redis."` + RedisURL string `name:"redis-url" placeholder:"URL" help:"URL of the Redis server that keeps divergences when storage is redis, such as redis://localhost:6379/0."` +} + +// Validate checks that the selected backend is configured. +func (c Config) Validate() error { + switch c.Storage { + case BackendMemory: + return nil + case BackendRedis: + if c.RedisURL == "" { + return errors.New("redis storage needs a Redis URL") + } + if _, err := redis.ParseURL(c.RedisURL); err != nil { + return errors.Wrap(err, "redis URL") + } + return nil + } + return errors.Errorf("unknown storage %q", c.Storage) +} + +// Store keeps the latest divergence of each mirror direction. Callers validate +// divergences before recording them and never mutate stored or listed ones. +type Store interface { + // Record keeps a divergence unless its source has already reported a later quarantine. + Record(ctx context.Context, divergence *pb.Divergence) error + // Clear removes the divergences of both directions of a service environment's mirror. + Clear(ctx context.Context, service, environment string) error + // List returns every stored divergence, in no particular order. + List(ctx context.Context) ([]*pb.Divergence, error) + Close() error +} + +// Open returns the store the configuration selects. +func Open(ctx context.Context, config Config) (Store, error) { + if err := config.Validate(); err != nil { + return nil, err + } + switch config.Storage { + case BackendMemory: + return NewMemory(), nil + case BackendRedis: + return NewRedis(ctx, config) + } + return nil, errors.Errorf("unknown storage %q", config.Storage) +} + +// quarantinedAfter reports whether existing began after incoming. A retry from a +// mirror that has since restarted and quarantined again must not replace it. +func quarantinedAfter(existing, incoming *pb.Divergence) bool { + return existing.GetQuarantine().GetSince().AsTime().After(incoming.GetQuarantine().GetSince().AsTime()) +} diff --git a/internal/server/storage/storage_test.go b/internal/server/storage/storage_test.go new file mode 100644 index 0000000..1af0e28 --- /dev/null +++ b/internal/server/storage/storage_test.go @@ -0,0 +1,151 @@ +package storage_test + +import ( + "cmp" + "slices" + "strconv" + "sync" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + "github.com/alicebob/miniredis/v2" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/timestamppb" + + "github.com/block/spectre/internal/server/storage" + pb "github.com/block/spectre/spectrepb" +) + +func TestStores(t *testing.T) { + backends := []struct { + name string + open func(t *testing.T) storage.Store + }{ + {name: "Memory", open: func(*testing.T) storage.Store { return storage.NewMemory() }}, + {name: "Redis", open: openRedis}, + } + tests := []struct { + name string + test func(t *testing.T, store storage.Store) + }{ + {name: "KeepsLatestQuarantine", test: testKeepsLatestQuarantine}, + {name: "ClearsServiceEnvironment", test: testClearsServiceEnvironment}, + {name: "RecordsConcurrently", test: testRecordsConcurrently}, + } + for _, backend := range backends { + t.Run(backend.name, func(t *testing.T) { + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + store := backend.open(t) + t.Cleanup(func() { assert.NoError(t, store.Close()) }) + test.test(t, store) + }) + } + }) + } +} + +func TestOpen(t *testing.T) { + server := miniredis.RunT(t) + tests := []struct { + name string + config storage.Config + err string + }{ + {name: "Memory", config: storage.Config{Storage: storage.BackendMemory}}, + {name: "Redis", config: storage.Config{Storage: storage.BackendRedis, RedisURL: "redis://" + server.Addr()}}, + {name: "RedisWithoutURL", config: storage.Config{Storage: storage.BackendRedis}, err: "redis storage needs a Redis URL"}, + {name: "InvalidRedisURL", config: storage.Config{Storage: storage.BackendRedis, RedisURL: "http://localhost"}, err: "redis URL"}, + {name: "UnreachableRedis", config: storage.Config{Storage: storage.BackendRedis, RedisURL: "redis://127.0.0.1:1"}, err: "connect to Redis"}, + {name: "UnknownBackend", config: storage.Config{Storage: "disk"}, err: `unknown storage "disk"`}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + store, err := storage.Open(t.Context(), test.config) + if test.err != "" { + assert.Contains(t, err.Error(), test.err) + return + } + assert.NoError(t, err) + assert.NoError(t, store.Close()) + }) + } +} + +func openRedis(t *testing.T) storage.Store { + t.Helper() + server := miniredis.RunT(t) + store, err := storage.NewRedis(t.Context(), storage.Config{Storage: storage.BackendRedis, RedisURL: "redis://" + server.Addr()}) + assert.NoError(t, err) + return store +} + +func testKeepsLatestQuarantine(t *testing.T, store storage.Store) { + first := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) + latest := divergence("production", pb.Direction_DIRECTION_INGRESS, first.Add(time.Minute)) + egress := divergence("production", pb.Direction_DIRECTION_EGRESS, first) + for _, reported := range []*pb.Divergence{ + divergence("production", pb.Direction_DIRECTION_INGRESS, first), + latest, + // A stale retry from before the mirror restarted. + divergence("production", pb.Direction_DIRECTION_INGRESS, first.Add(-time.Minute)), + egress, + } { + assert.NoError(t, store.Record(t.Context(), reported)) + } + assertStored(t, store, latest, egress) +} + +func testClearsServiceEnvironment(t *testing.T, store storage.Store) { + since := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) + staging := divergence("staging", pb.Direction_DIRECTION_INGRESS, since) + for _, reported := range []*pb.Divergence{ + divergence("production", pb.Direction_DIRECTION_INGRESS, since), + divergence("production", pb.Direction_DIRECTION_EGRESS, since), + staging, + } { + assert.NoError(t, store.Record(t.Context(), reported)) + } + assert.NoError(t, store.Clear(t.Context(), "users", "production")) + assertStored(t, store, staging) +} + +// testRecordsConcurrently checks that concurrent writes to different sources +// are all kept, which makes the Redis store retry its transactions. +func testRecordsConcurrently(t *testing.T, store storage.Store) { + since := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) + expected := []*pb.Divergence{} + for i := range 20 { + expected = append(expected, divergence("environment-"+strconv.Itoa(i), pb.Direction_DIRECTION_INGRESS, since)) + } + var wg sync.WaitGroup + for _, reported := range expected { + wg.Go(func() { assert.NoError(t, store.Record(t.Context(), reported)) }) + } + wg.Wait() + assertStored(t, store, expected...) +} + +func divergence(environment string, direction pb.Direction, since time.Time) *pb.Divergence { + return &pb.Divergence{ + Service: "users", + Environment: environment, + Revision: "abc123", + Direction: direction, + Quarantine: &pb.Quarantine{Since: timestamppb.New(since), Kind: pb.Quarantine_KIND_CAPACITY}, + } +} + +// assertStored checks the store holds exactly the expected divergences, in any order. +func assertStored(t *testing.T, store storage.Store, expected ...*pb.Divergence) { + t.Helper() + stored, err := store.List(t.Context()) + assert.NoError(t, err) + bySource := func(a, b *pb.Divergence) int { + return cmp.Or(cmp.Compare(a.GetEnvironment(), b.GetEnvironment()), cmp.Compare(a.GetDirection(), b.GetDirection())) + } + slices.SortFunc(stored, bySource) + expected = slices.SortedFunc(slices.Values(expected), bySource) + assert.True(t, proto.Equal(&pb.ListDivergencesResponse{Divergences: expected}, &pb.ListDivergencesResponse{Divergences: stored}), "%v", stored) +} diff --git a/internal/server/store.go b/internal/server/store.go deleted file mode 100644 index 74e5a6a..0000000 --- a/internal/server/store.go +++ /dev/null @@ -1,103 +0,0 @@ -package server - -import ( - "cmp" - "slices" - "sync" - - "github.com/alecthomas/errors" - . "github.com/alecthomas/types/optional" - - pb "github.com/block/spectre/spectrepb" -) - -// store holds the latest quarantine of each mirror direction. Stored messages -// are never mutated, so they may be shared. -type store struct { - mu sync.Mutex - divergences map[sourceKey]*pb.Divergence -} - -func newStore() *store { - return &store{divergences: map[sourceKey]*pb.Divergence{}} -} - -// sourceKey identifies the proxy of one direction of a service environment's mirror. -type sourceKey struct { - service, environment string - direction pb.Direction -} - -// record keeps a divergence unless its source has already reported a later one. -func (s *store) record(divergence *pb.Divergence) error { - if divergence.GetService() == "" || divergence.GetEnvironment() == "" || divergence.GetRevision() == "" { - return errors.New("divergence needs a service, environment, and revision") - } - if _, err := DirectionFromMessage(divergence.GetDirection()); err != nil { - return err - } - quarantine := divergence.GetQuarantine() - if quarantine.GetSince() == nil { - return errors.New("divergence needs a quarantine with its start time") - } - if _, err := ReasonFromMessage(quarantine); err != nil { - return errors.Wrap(err, "quarantine") - } - key := sourceKey{service: divergence.GetService(), environment: divergence.GetEnvironment(), direction: divergence.GetDirection()} - s.mu.Lock() - defer s.mu.Unlock() - // A retry from a mirror that has since restarted and quarantined again must not - // replace the newer quarantine. - if existing, ok := s.divergences[key]; ok && existing.GetQuarantine().GetSince().AsTime().After(quarantine.GetSince().AsTime()) { - return nil - } - s.divergences[key] = divergence - return nil -} - -// clear removes the divergences of both directions of a service environment's mirror. -func (s *store) clear(service, environment string) error { - if service == "" || environment == "" { - return errors.New("clearing divergences needs a service and environment") - } - s.mu.Lock() - defer s.mu.Unlock() - for key := range s.divergences { - if key.service == service && key.environment == environment { - delete(s.divergences, key) - } - } - return nil -} - -// listDivergences returns the matching divergences, ordered by source. -func (s *store) listDivergences(request *pb.ListDivergencesRequest) []*pb.Divergence { - s.mu.Lock() - defer s.mu.Unlock() - divergences := []*pb.Divergence{} - for _, divergence := range s.divergences { - quarantine := divergence.GetQuarantine() - // Getters hide whether an optional field is set, so presence is a nil check. - if matches(From(request.GetService(), request.Service != nil), Some(divergence.GetService())) && - matches(From(request.GetEnvironment(), request.Environment != nil), Some(divergence.GetEnvironment())) && - matches(From(request.GetRevision(), request.Revision != nil), Some(divergence.GetRevision())) && - matches(From(request.GetNode(), request.Node != nil), From(quarantine.GetNode(), quarantine.Node != nil)) { - divergences = append(divergences, divergence) - } - } - slices.SortFunc(divergences, func(a, b *pb.Divergence) int { - return cmp.Or( - cmp.Compare(a.GetService(), b.GetService()), - cmp.Compare(a.GetEnvironment(), b.GetEnvironment()), - cmp.Compare(a.GetDirection(), b.GetDirection()), - ) - }) - return divergences -} - -// matches reports whether value passes a filter. An unset filter matches -// every value, including an absent one. -func matches(filter, value Option[string]) bool { - expected, ok := filter.Get() - return !ok || value == Some(expected) -} From 905331272dc70d4a1b23db11081abc8dcfbfbdf4 Mon Sep 17 00:00:00 2001 From: Alec Thomas Date: Wed, 28 Oct 2026 21:22:37 +1100 Subject: [PATCH 2/2] test: give the integration test more time on slow CI runners On a loaded runner the response comparison integration test ran past its one-minute deadline with no failed check, and the supervisor used most of its readiness timeout. The same code passed in under a minute elsewhere. Co-Authored-By: Claude Opus 5.5 --- Procfile.integration | 6 +++--- scripts/response-comparison.integration.test.sh | 3 ++- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/Procfile.integration b/Procfile.integration index 1065a0f..880a4ff 100644 --- a/Procfile.integration +++ b/Procfile.integration @@ -1,5 +1,5 @@ setup! : mkdir -p dist/sockets -forecasts after=setup ready=http+unix:dist/sockets/forecasts.sock:/readyz=204 timeout=30s: spectre-sample-forecasts --listen="unix:$(pwd)/dist/sockets/forecasts.sock" --log-level=debug --log-json=DEFAULT -server ready=http:50070/readyz=204 timeout=30s: spectre serve --log-level=debug --log-json=DEFAULT -supervisor after=forecasts,server ready=http+unix:dist/sockets/ingress.sock:/readyz=204 timeout=30s: spectre supervise --server-url=http://127.0.0.1:50070 --service=sample --environment=integration --revision=integration --no-limits --listen="unix:$(pwd)/dist/sockets/ingress.sock" --socket-dir="$(pwd)/dist/sockets" --reference='spectre-sample --listen=$SPECTRE_LISTEN --forecasts=$SPECTRE_EGRESS --log-level=debug --log-json=DEFAULT' --candidate='spectre-sample --listen=$SPECTRE_LISTEN --forecasts=$SPECTRE_EGRESS --data=${SPECTRE_CANDIDATE_DATA:-internal/sample/testdata/users.json} --weather=${SPECTRE_CANDIDATE_WEATHER:-internal/sample/testdata/weather.json} --revision=candidate --log-level=debug --log-json=DEFAULT' --destination="forecasts.example=http+unix:$(pwd)/dist/sockets/forecasts.sock" --scripts-dir="${SPECTRE_SCRIPTS_DIR:-internal/sample/scripts}" --schema-dir=internal/sample/schema --descriptors-dir=dist/descriptors --match-window=1s --log-level=debug --log-json=DEFAULT +forecasts after=setup ready=http+unix:dist/sockets/forecasts.sock:/readyz=204 timeout=90s: spectre-sample-forecasts --listen="unix:$(pwd)/dist/sockets/forecasts.sock" --log-level=debug --log-json=DEFAULT +server ready=http:50070/readyz=204 timeout=90s: spectre serve --log-level=debug --log-json=DEFAULT +supervisor after=forecasts,server ready=http+unix:dist/sockets/ingress.sock:/readyz=204 timeout=90s: spectre supervise --server-url=http://127.0.0.1:50070 --service=sample --environment=integration --revision=integration --no-limits --listen="unix:$(pwd)/dist/sockets/ingress.sock" --socket-dir="$(pwd)/dist/sockets" --reference='spectre-sample --listen=$SPECTRE_LISTEN --forecasts=$SPECTRE_EGRESS --log-level=debug --log-json=DEFAULT' --candidate='spectre-sample --listen=$SPECTRE_LISTEN --forecasts=$SPECTRE_EGRESS --data=${SPECTRE_CANDIDATE_DATA:-internal/sample/testdata/users.json} --weather=${SPECTRE_CANDIDATE_WEATHER:-internal/sample/testdata/weather.json} --revision=candidate --log-level=debug --log-json=DEFAULT' --destination="forecasts.example=http+unix:$(pwd)/dist/sockets/forecasts.sock" --scripts-dir="${SPECTRE_SCRIPTS_DIR:-internal/sample/scripts}" --schema-dir=internal/sample/schema --descriptors-dir=dist/descriptors --match-window=1s --log-level=debug --log-json=DEFAULT verify! after=supervisor: if [ -n "${SPECTRE_PROCTOR_LOG:-}" ]; then response-comparison.integration.test.sh verify; fi diff --git a/scripts/response-comparison.integration.test.sh b/scripts/response-comparison.integration.test.sh index 66072d8..8811a72 100755 --- a/scripts/response-comparison.integration.test.sh +++ b/scripts/response-comparison.integration.test.sh @@ -183,8 +183,9 @@ cd "$source_root" proctor "$source_root/Procfile.integration" > "$SPECTRE_PROCTOR_LOG" 2>&1 & proctor_pid=$! +# A loaded CI runner can take a minute to pass, so allow five. attempt=0 -while [ "$attempt" -lt 600 ]; do +while [ "$attempt" -lt 3000 ]; do if [ -f "$SPECTRE_PROCTOR_RESULT" ]; then kill -TERM "$proctor_pid" 2>/dev/null || true wait "$proctor_pid" 2>/dev/null || true