From c8b7da934c9890afde126fb98127b0aa815f2ad8 Mon Sep 17 00:00:00 2001 From: Alec Thomas Date: Wed, 28 Oct 2026 21:04:38 +1100 Subject: [PATCH] refactor: keep protobuf messages at the server API boundary Storage, the reporter, and the command line passed generated protocol messages around, which tied the internals to the public protocol and spread validation across callers. Converting at the API boundary lets every other package work with validated domain values, and the architecture linter now rejects protocol imports outside it. The Redis backend stores JSON of the domain type instead of protobuf. It has not been released, so existing data needs no migration. Co-Authored-By: Claude Opus 5.5 --- .tsk/architecture.ts | 36 +++- .../internal/server/storage/storage.go | 5 + .../internal/server/storage/storage_test.go | 5 + .../architecture/internal/server/wire/wire.go | 5 + .../internal/server/wire/wire_test.go | 5 + .../architecture/spectrepb/spectrepb.go | 3 + .../spectrepbconnect/spectrepbconnect.go | 5 + AGENTS.md | 4 + cli_internal_test.go | 7 +- internal/cli/divergences.go | 49 ++--- internal/cli/divergences_internal_test.go | 64 +++--- internal/model/divergence.go | 39 ++++ internal/model/divergence_test.go | 40 ++++ internal/model/quarantine.go | 16 +- internal/model/traffic.go | 23 ++- internal/server/client/client.go | 22 ++- internal/server/config.go | 8 - internal/server/divergences.go | 53 +---- internal/server/{ => reporter}/reporter.go | 24 ++- internal/server/reporter/reporter_test.go | 72 +++++++ internal/server/server_test.go | 60 ------ internal/server/service.go | 11 +- internal/server/storage/memory.go | 22 +-- internal/server/storage/redis.go | 52 +++-- internal/server/storage/storage.go | 10 +- internal/server/storage/storage_test.go | 59 +++--- internal/server/wire.go | 117 ----------- internal/server/wire/wire.go | 187 ++++++++++++++++++ internal/server/wire/wire_test.go | 67 +++++++ internal/supervisor/config.go | 8 +- internal/supervisor/supervisor.go | 8 +- 31 files changed, 692 insertions(+), 394 deletions(-) create mode 100644 .tsk/testdata/architecture/internal/server/storage/storage.go create mode 100644 .tsk/testdata/architecture/internal/server/storage/storage_test.go create mode 100644 .tsk/testdata/architecture/internal/server/wire/wire.go create mode 100644 .tsk/testdata/architecture/internal/server/wire/wire_test.go create mode 100644 .tsk/testdata/architecture/spectrepb/spectrepb.go create mode 100644 .tsk/testdata/architecture/spectrepb/spectrepbconnect/spectrepbconnect.go create mode 100644 internal/model/divergence.go create mode 100644 internal/model/divergence_test.go rename internal/server/{ => reporter}/reporter.go (64%) create mode 100644 internal/server/reporter/reporter_test.go delete mode 100644 internal/server/wire.go create mode 100644 internal/server/wire/wire.go create mode 100644 internal/server/wire/wire_test.go diff --git a/.tsk/architecture.ts b/.tsk/architecture.ts index 3c63561..d911741 100644 --- a/.tsk/architecture.ts +++ b/.tsk/architecture.ts @@ -3,6 +3,10 @@ import { defineAnalyzer } from "tsk"; // Module-relative path of the command line, which owns all terminal output. const cliPackage = "internal/cli"; const terminalPackage = cliPackage + "/terminal"; +// Module-relative paths of the SPECTRE service protocol's generated packages. +const protocolPackages = ["spectrepb", "spectrepb/spectrepbconnect"]; +// The server's API boundary, which converts protocol messages to domain types. +const protocolBoundary = ["internal/server", "internal/server/wire", "internal/server/client"]; export default defineAnalyzer({ name: "architecture", @@ -13,7 +17,11 @@ belongs in internal/. Keep a startup helper in a main package with a //nolint:architecture comment explaining why it is not application logic. Only ${cliPackage} and its sub-packages may import ${terminalPackage}, because -terminal output is the domain of the command line. Domain packages return data.`, +terminal output is the domain of the command line. Domain packages return data. + +Only the SPECTRE server's API boundary, ${protocolBoundary.join(", ")}, may +import ${protocolPackages.join(" or ")}, because protocol messages stop at the +API. Everything else uses internal/model types, which the boundary converts.`, run(pass) { // Generated test main packages live in the build cache, not the source tree. if (pass.pkg.path().endsWith(".test")) { @@ -25,16 +33,24 @@ terminal output is the domain of the command line. Domain packages return data.` // An external test package belongs to the package it tests. const owner = relative(pass.pkg.path()).replace(/_test$/, ""); const inCLI = owner === cliPackage || owner.startsWith(cliPackage + "/"); + const atProtocolBoundary = protocolBoundary.includes(owner) || protocolPackages.includes(owner); for (const file of pass.files) { - if (!inCLI) { - for (const spec of file.imports) { - const path = spec?.path?.value.slice(1, -1); - if (spec !== null && path !== undefined && relative(path) === terminalPackage) { - pass.report({ - pos: spec.pos(), - message: `only ${cliPackage} may import ${path}: return data and let the command line write it`, - }); - } + for (const spec of file.imports) { + const path = spec?.path?.value.slice(1, -1); + if (spec === null || path === undefined) { + continue; + } + if (!inCLI && relative(path) === terminalPackage) { + pass.report({ + pos: spec.pos(), + message: `only ${cliPackage} may import ${path}: return data and let the command line write it`, + }); + } + if (!atProtocolBoundary && protocolPackages.includes(relative(path))) { + pass.report({ + pos: spec.pos(), + message: `only the SPECTRE server API boundary may import ${path}: use internal/model types, converted by internal/server/wire`, + }); } } if (pass.pkg.name() !== "main" || pass.fset.position(file.package).filename.endsWith("_test.go")) { diff --git a/.tsk/testdata/architecture/internal/server/storage/storage.go b/.tsk/testdata/architecture/internal/server/storage/storage.go new file mode 100644 index 0000000..e719b54 --- /dev/null +++ b/.tsk/testdata/architecture/internal/server/storage/storage.go @@ -0,0 +1,5 @@ +package storage + +import "example.com/architecture/spectrepb" // want `only the SPECTRE server API boundary may import example.com/architecture/spectrepb` + +var _ = spectrepb.Divergence{} diff --git a/.tsk/testdata/architecture/internal/server/storage/storage_test.go b/.tsk/testdata/architecture/internal/server/storage/storage_test.go new file mode 100644 index 0000000..817956e --- /dev/null +++ b/.tsk/testdata/architecture/internal/server/storage/storage_test.go @@ -0,0 +1,5 @@ +package storage_test + +import "example.com/architecture/spectrepb/spectrepbconnect" // want `only the SPECTRE server API boundary may import` + +var _ = spectrepbconnect.Client{} diff --git a/.tsk/testdata/architecture/internal/server/wire/wire.go b/.tsk/testdata/architecture/internal/server/wire/wire.go new file mode 100644 index 0000000..af90283 --- /dev/null +++ b/.tsk/testdata/architecture/internal/server/wire/wire.go @@ -0,0 +1,5 @@ +package wire + +import "example.com/architecture/spectrepb" + +var _ = spectrepb.Divergence{} diff --git a/.tsk/testdata/architecture/internal/server/wire/wire_test.go b/.tsk/testdata/architecture/internal/server/wire/wire_test.go new file mode 100644 index 0000000..7c03368 --- /dev/null +++ b/.tsk/testdata/architecture/internal/server/wire/wire_test.go @@ -0,0 +1,5 @@ +package wire_test + +import "example.com/architecture/spectrepb" + +var _ = spectrepb.Divergence{} diff --git a/.tsk/testdata/architecture/spectrepb/spectrepb.go b/.tsk/testdata/architecture/spectrepb/spectrepb.go new file mode 100644 index 0000000..77672f9 --- /dev/null +++ b/.tsk/testdata/architecture/spectrepb/spectrepb.go @@ -0,0 +1,3 @@ +package spectrepb + +type Divergence struct{} diff --git a/.tsk/testdata/architecture/spectrepb/spectrepbconnect/spectrepbconnect.go b/.tsk/testdata/architecture/spectrepb/spectrepbconnect/spectrepbconnect.go new file mode 100644 index 0000000..e0d4984 --- /dev/null +++ b/.tsk/testdata/architecture/spectrepb/spectrepbconnect/spectrepbconnect.go @@ -0,0 +1,5 @@ +package spectrepbconnect + +import "example.com/architecture/spectrepb" + +type Client struct{ Divergence spectrepb.Divergence } diff --git a/AGENTS.md b/AGENTS.md index 8570252..88409d7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -44,6 +44,10 @@ - The SPECTRE service protocol in `proto/spectre/v1` and its generated `spectrepb` packages are also public. Keep its definitions independent of internal types. +- Use protocol messages only at the SPECTRE server's API boundary: its handler + in `internal/server`, the conversions in `internal/server/wire`, and + `internal/server/client`. Everything else uses `internal/model` types. The + `architecture` linter enforces this. - Expose only extension points that have been explicitly agreed. Add each as one public contract and one `With...` option. - Never expose an internal type in the public API, even through an alias. Mirror diff --git a/cli_internal_test.go b/cli_internal_test.go index 7dcd98e..d1924dd 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/reporter" "github.com/block/spectre/internal/server/storage" "github.com/block/spectre/internal/supervisor" ) @@ -24,8 +25,8 @@ func reportingFlags() []string { } // expectedReporting is the reporter config that reportingFlags parse to. -func expectedReporting() server.ReporterConfig { - return server.ReporterConfig{ +func expectedReporting() reporter.Config { + return reporter.Config{ Client: client.Config{URL: "http://server.example"}, Service: "users", Environment: "production", @@ -81,7 +82,7 @@ func TestSuperviseCommandReadsFlagsFromEnvironment(t *testing.T) { "--schema-dir=" + writeSchemaDir(t), }) assert.NoError(t, err) - expected := server.ReporterConfig{ + expected := reporter.Config{ Client: client.Config{URL: "http://server.example"}, Service: "invoices", Environment: "production-us-west-2", diff --git a/internal/cli/divergences.go b/internal/cli/divergences.go index bb98f5c..2a235da 100644 --- a/internal/cli/divergences.go +++ b/internal/cli/divergences.go @@ -7,12 +7,11 @@ import ( "github.com/alecthomas/errors" . "github.com/alecthomas/types/optional" - "google.golang.org/protobuf/encoding/protojson" - "google.golang.org/protobuf/proto" + "github.com/block/spectre/internal/model" "github.com/block/spectre/internal/server" "github.com/block/spectre/internal/server/client" - pb "github.com/block/spectre/spectrepb" + "github.com/block/spectre/internal/server/wire" ) // divergencesCommand groups the commands for the divergences mirrors reported. @@ -33,33 +32,28 @@ type divergencesListCommand struct { func (c *divergencesListCommand) Run(ctx context.Context, out Output) error { // No reported value is empty, so an empty flag leaves its filter unset. - request := &pb.ListDivergencesRequest{ - Service: Zero(c.Service).Ptr(), - Environment: Zero(c.Environment).Ptr(), - Revision: Zero(c.Revision).Ptr(), - Node: Zero(c.Node).Ptr(), + filter := model.DivergenceFilter{ + Service: Zero(c.Service), + Environment: Zero(c.Environment), + Revision: Zero(c.Revision), + Node: Zero(c.Node), } - response, err := client.New(c.Client).ListDivergences(ctx, request) + divergences, err := client.New(c.Client).ListDivergences(ctx, filter) if err != nil { return errors.WithStack(err) } if c.JSON { - return writeJSON(out, response) + return writeJSON(out, divergences) } - divergences := response.GetDivergences() palette := out.Palette if len(divergences) == 0 { _, err := fmt.Fprintln(out.Writer, palette.Dim("No divergences.")) return errors.Wrap(err, "write divergences") } for _, divergence := range divergences { - description, err := describe(divergence) - if err != nil { - return err - } - since := "quarantined " + divergence.GetQuarantine().GetSince().AsTime().UTC().Format(time.RFC3339) - _, err = fmt.Fprintf(out.Writer, "%s %s %s\n %s\n %s\n", palette.Bold(divergence.GetService()), divergence.GetRevision(), - palette.Dim(divergence.GetEnvironment()), palette.Red(description), palette.Dim(since)) + since := "quarantined " + divergence.Quarantine.Since.UTC().Format(time.RFC3339) + _, err = fmt.Fprintf(out.Writer, "%s %s %s\n %s\n %s\n", palette.Bold(divergence.Service), divergence.Revision, + palette.Dim(divergence.Environment), palette.Red(describe(divergence.Direction, divergence.Quarantine.Reason)), palette.Dim(since)) if err != nil { return errors.Wrap(err, "write divergences") } @@ -78,15 +72,7 @@ func (c *divergencesClearCommand) Run(ctx context.Context) error { // describe renders a quarantine as the mirror logged it, with the direction, // route, and node it names. -func describe(divergence *pb.Divergence) (string, error) { - direction, err := server.DirectionFromMessage(divergence.GetDirection()) - if err != nil { - return "", errors.Wrap(err, "describe divergence") - } - reason, err := server.ReasonFromMessage(divergence.GetQuarantine()) - if err != nil { - return "", errors.Wrap(err, "describe divergence") - } +func describe(direction model.Direction, reason model.QuarantineReason) string { text := fmt.Sprintf("%s: %s", direction, reason) if traffic, ok := reason.Traffic.Get(); ok { text = fmt.Sprintf("%s %s %s: %s", direction, traffic.Protocol, traffic.Route, reason) @@ -94,13 +80,14 @@ func describe(divergence *pb.Divergence) (string, error) { if node, ok := reason.Node.Get(); ok { text += fmt.Sprintf(" (node %s)", node) } - return text, nil + return text } -func writeJSON(out Output, message proto.Message) error { - data, err := protojson.MarshalOptions{Multiline: true}.Marshal(message) +// writeJSON prints divergences as the SPECTRE service protocol's list response. +func writeJSON(out Output, divergences []model.Divergence) error { + data, err := wire.DivergencesJSON(divergences) if err != nil { - return errors.Wrap(err, "encode JSON") + return errors.WithStack(err) } _, err = fmt.Fprintln(out.Writer, string(data)) return errors.Wrap(err, "write JSON") diff --git a/internal/cli/divergences_internal_test.go b/internal/cli/divergences_internal_test.go index 0113440..b6731bf 100644 --- a/internal/cli/divergences_internal_test.go +++ b/internal/cli/divergences_internal_test.go @@ -1,60 +1,54 @@ package cli import ( + "encoding/json" "net/http/httptest" "strings" "testing" "time" - "connectrpc.com/connect" "github.com/alecthomas/assert/v2" - "google.golang.org/protobuf/encoding/protojson" - "google.golang.org/protobuf/proto" - "google.golang.org/protobuf/types/known/timestamppb" + . "github.com/alecthomas/types/optional" + "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 TestDivergencesListCommand(t *testing.T) { httpServer := httptest.NewServer(server.NewService(storage.NewMemory()).Handler()) t.Cleanup(httpServer.Close) - rpc := spectrepbconnect.NewSpectreServiceClient(httpServer.Client(), httpServer.URL) + config := client.Config{URL: httpServer.URL} first := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) - divergent := &pb.Divergence{ + divergent := model.Divergence{ Service: "users", Environment: "production", Revision: "abc123", - Direction: pb.Direction_DIRECTION_INGRESS, - Quarantine: &pb.Quarantine{ - Since: timestamppb.New(first.Add(time.Minute)), - Kind: pb.Quarantine_KIND_DIVERGENT, - Traffic: &pb.Traffic{Direction: pb.Direction_DIRECTION_INGRESS, Protocol: "http", Route: "GET /v1/users"}, - Node: proto.String("users"), + Direction: model.Ingress, + Quarantine: model.Quarantine{Since: first.Add(time.Minute), Reason: model.QuarantineReason{ + Kind: model.QuarantineDivergent, + Traffic: Some(model.Traffic{Direction: model.Ingress, Protocol: model.ProtocolHTTP, Route: "GET /v1/users"}), + Node: Some("users"), Differences: []string{"body.name"}, - }, + }}, } - unmatched := &pb.Divergence{ + unmatched := model.Divergence{ Service: "users", Environment: "staging", Revision: "abc123", - Direction: pb.Direction_DIRECTION_EGRESS, - Quarantine: &pb.Quarantine{ - Since: timestamppb.New(first), - Kind: pb.Quarantine_KIND_UNMATCHED, + Direction: model.Egress, + Quarantine: model.Quarantine{Since: first, Reason: model.QuarantineReason{ + Kind: model.QuarantineUnmatched, Detail: "no matching reference call", - }, + }}, } - for _, divergence := range []*pb.Divergence{divergent, unmatched} { - _, err := rpc.Quarantine(t.Context(), connect.NewRequest(&pb.QuarantineRequest{Divergence: divergence})) - assert.NoError(t, err) + for _, divergence := range []model.Divergence{divergent, unmatched} { + assert.NoError(t, client.New(config).Quarantine(t.Context(), divergence)) } var text strings.Builder - command := divergencesListCommand{Client: client.Config{URL: httpServer.URL}} + command := divergencesListCommand{Client: config} assert.NoError(t, command.Run(t.Context(), Output{Writer: &text})) assert.Equal(t, `users abc123 production ingress http GET /v1/users: response differences: body.name (node users) @@ -68,7 +62,21 @@ users abc123 staging command.JSON = true command.Node = "users" assert.NoError(t, command.Run(t.Context(), Output{Writer: &data})) - decoded := &pb.ListDivergencesResponse{} - assert.NoError(t, protojson.Unmarshal([]byte(data.String()), decoded)) - assert.True(t, proto.Equal(&pb.ListDivergencesResponse{Divergences: []*pb.Divergence{divergent}}, decoded), "%v", decoded) + // Protobuf JSON output varies its whitespace, so compare decoded values. + var expected, printed any + assert.NoError(t, json.Unmarshal([]byte(`{"divergences": [{ + "service": "users", + "environment": "production", + "revision": "abc123", + "direction": "DIRECTION_INGRESS", + "quarantine": { + "since": "2026-10-01T12:01:00Z", + "kind": "KIND_DIVERGENT", + "traffic": {"direction": "DIRECTION_INGRESS", "protocol": "http", "route": "GET /v1/users"}, + "node": "users", + "differences": ["body.name"] + } + }]}`), &expected)) + assert.NoError(t, json.Unmarshal([]byte(data.String()), &printed)) + assert.Equal(t, expected, printed) } diff --git a/internal/model/divergence.go b/internal/model/divergence.go new file mode 100644 index 0000000..2d555d9 --- /dev/null +++ b/internal/model/divergence.go @@ -0,0 +1,39 @@ +package model + +import . "github.com/alecthomas/types/optional" + +// Divergence is the quarantine that the mirror of a service environment +// reported for one of its directions. +type Divergence struct { + Service string `json:"service"` + Environment string `json:"environment"` + // Revision is the commit of the candidate the mirror runs. + Revision string `json:"revision"` + // Direction is the proxy that quarantined the candidate. + Direction Direction `json:"direction"` + Quarantine Quarantine `json:"quarantine"` +} + +// DivergenceFilter selects divergences. An absent field matches every value. +type DivergenceFilter struct { + Service Option[string] + Environment Option[string] + Revision Option[string] + // Node matches only divergences on routes of that migration state node. + Node Option[string] +} + +// Matches reports whether divergence passes every field of the filter. +func (f DivergenceFilter) Matches(divergence Divergence) bool { + return matches(f.Service, Some(divergence.Service)) && + matches(f.Environment, Some(divergence.Environment)) && + matches(f.Revision, Some(divergence.Revision)) && + matches(f.Node, divergence.Quarantine.Reason.Node) +} + +// matches reports whether value passes a filter. An absent 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/model/divergence_test.go b/internal/model/divergence_test.go new file mode 100644 index 0000000..f903d19 --- /dev/null +++ b/internal/model/divergence_test.go @@ -0,0 +1,40 @@ +package model_test + +import ( + "testing" + + "github.com/alecthomas/assert/v2" + . "github.com/alecthomas/types/optional" + + "github.com/block/spectre/internal/model" +) + +func TestDivergenceFilterMatches(t *testing.T) { + divergence := model.Divergence{ + Service: "users", + Environment: "production", + Revision: "abc123", + Direction: model.Ingress, + Quarantine: model.Quarantine{Reason: model.QuarantineReason{Kind: model.QuarantineDivergent, Node: Some("users")}}, + } + withoutNode := divergence + withoutNode.Quarantine.Reason.Node = None[string]() + tests := []struct { + name string + filter model.DivergenceFilter + divergence model.Divergence + matches bool + }{ + {name: "Unset", divergence: divergence, matches: true}, + {name: "UnsetMatchesAbsentNode", divergence: withoutNode, matches: true}, + {name: "AllFields", filter: model.DivergenceFilter{Service: Some("users"), Environment: Some("production"), Revision: Some("abc123"), Node: Some("users")}, divergence: divergence, matches: true}, + {name: "OtherEnvironment", filter: model.DivergenceFilter{Environment: Some("staging")}, divergence: divergence}, + {name: "OtherRevision", filter: model.DivergenceFilter{Revision: Some("def456")}, divergence: divergence}, + {name: "NodeRejectsAbsentNode", filter: model.DivergenceFilter{Node: Some("users")}, divergence: withoutNode}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.matches, test.filter.Matches(test.divergence)) + }) + } +} diff --git a/internal/model/quarantine.go b/internal/model/quarantine.go index 0b6b67b..797896f 100644 --- a/internal/model/quarantine.go +++ b/internal/model/quarantine.go @@ -28,14 +28,14 @@ const ( // QuarantineReason is why the candidate was quarantined. Detail and Differences // must never quote payload values, because reasons are reported outside the mirror. type QuarantineReason struct { - Kind QuarantineKind + Kind QuarantineKind `json:"kind"` // Traffic is the request that triggered quarantine. - Traffic Option[Traffic] + Traffic Option[Traffic] `json:"traffic,omitzero"` // Node is the migration state node that owns an ingress route. - Node Option[string] + Node Option[string] `json:"node,omitzero"` // Differences are the field paths of a divergence, without values. - Differences []string - Detail string + Differences []string `json:"differences,omitempty"` + Detail string `json:"detail,omitempty"` } // String describes the reason for logs. @@ -47,7 +47,7 @@ func (r QuarantineReason) String() string { } // Quarantine is a quarantine in effect: why it began and when. -type Quarantine struct { - Reason QuarantineReason - Since time.Time +type Quarantine struct { //nolint:singleusestruct // Proxies report quarantines on their own, outside any divergence. + Reason QuarantineReason `json:"reason"` //nolint:narrowdeps // Other packages read the whole reason. + Since time.Time `json:"since"` } diff --git a/internal/model/traffic.go b/internal/model/traffic.go index 4d494b2..d72dfa8 100644 --- a/internal/model/traffic.go +++ b/internal/model/traffic.go @@ -2,7 +2,13 @@ // boundary through its proxies, and why a candidate was quarantined. package model -import "fmt" +import ( + "bytes" + "encoding/json" + "fmt" + + "github.com/alecthomas/errors" +) // Direction names the proxy that carries traffic: ingress into the service, or // egress from it to a dependency. @@ -38,7 +44,16 @@ func (k IOKind) String() string { // Traffic identifies one request route of a kind of IO. type Traffic struct { - Direction Direction - Protocol Protocol - Route Route + Direction Direction `json:"direction"` + Protocol Protocol `json:"protocol"` + Route Route `json:"route"` +} + +// UnmarshalJSON rejects unknown fields. Option decodes with json.Unmarshal, so +// a reason's traffic must enforce strict decoding itself. +func (t *Traffic) UnmarshalJSON(data []byte) error { + type plain Traffic + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + return errors.WithStack(decoder.Decode((*plain)(t))) } diff --git a/internal/server/client/client.go b/internal/server/client/client.go index 493d9a7..fbba98e 100644 --- a/internal/server/client/client.go +++ b/internal/server/client/client.go @@ -10,6 +10,8 @@ import ( "github.com/alecthomas/errors" "github.com/block/spectre/internal/logger" + "github.com/block/spectre/internal/model" + "github.com/block/spectre/internal/server/wire" pb "github.com/block/spectre/spectrepb" "github.com/block/spectre/spectrepb/spectrepbconnect" ) @@ -37,8 +39,8 @@ func New(config Config) *Client { // Quarantine records a mirror's quarantine. A mirror reports each one only once, // so failures retry with backoff until the server accepts or rejects it. -func (c *Client) Quarantine(ctx context.Context, divergence *pb.Divergence) error { - request := &pb.QuarantineRequest{Divergence: divergence} +func (c *Client) Quarantine(ctx context.Context, divergence model.Divergence) error { + request := &pb.QuarantineRequest{Divergence: wire.DivergenceMessage(divergence)} for delay := minRetryDelay; ; delay = min(delay*2, maxRetryDelay) { err := c.quarantineOnce(ctx, request) if err == nil || ctx.Err() != nil || connect.CodeOf(err) == connect.CodeInvalidArgument { @@ -60,13 +62,21 @@ func (c *Client) quarantineOnce(ctx context.Context, request *pb.QuarantineReque return errors.Wrap(err, "report quarantine") } -// ListDivergences returns the divergences that match request's filters. -func (c *Client) ListDivergences(ctx context.Context, request *pb.ListDivergencesRequest) (*pb.ListDivergencesResponse, error) { - response, err := c.rpc.ListDivergences(ctx, connect.NewRequest(request)) +// ListDivergences returns the divergences that filter selects, ordered by source. +func (c *Client) ListDivergences(ctx context.Context, filter model.DivergenceFilter) ([]model.Divergence, error) { + response, err := c.rpc.ListDivergences(ctx, connect.NewRequest(wire.FilterMessage(filter))) if err != nil { return nil, errors.Wrap(err, "list divergences") } - return response.Msg, nil + divergences := make([]model.Divergence, 0, len(response.Msg.GetDivergences())) + for _, message := range response.Msg.GetDivergences() { + divergence, err := wire.DivergenceFromMessage(message) + if err != nil { + return nil, errors.Wrap(err, "list divergences") + } + divergences = append(divergences, divergence) + } + return divergences, nil } // ClearDivergences removes the divergences of one service environment. diff --git a/internal/server/config.go b/internal/server/config.go index f99b261..a7e29d4 100644 --- a/internal/server/config.go +++ b/internal/server/config.go @@ -7,14 +7,6 @@ import ( "github.com/block/spectre/internal/server/storage" ) -// ReporterConfig configures how a mirror reports quarantines to the SPECTRE server. -type ReporterConfig struct { - Client client.Config `embed:""` - Service string `required:"" help:"Service the mirror runs, reported to the SPECTRE server."` - Environment string `required:"" help:"Environment the mirror runs in, such as production-us-west-2, reported to the SPECTRE server."` - Revision string `required:"" help:"Commit of the candidate the mirror runs, reported to the SPECTRE server."` -} - // 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:."` diff --git a/internal/server/divergences.go b/internal/server/divergences.go index 5d47c77..9c5a8fe 100644 --- a/internal/server/divergences.go +++ b/internal/server/divergences.go @@ -4,57 +4,24 @@ import ( "cmp" "slices" - "github.com/alecthomas/errors" - . "github.com/alecthomas/types/optional" - - pb "github.com/block/spectre/spectrepb" + "github.com/block/spectre/internal/model" ) -// 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{} +// filterDivergences returns the divergences that filter selects, ordered by source. +func filterDivergences(divergences []model.Divergence, filter model.DivergenceFilter) []model.Divergence { + matching := []model.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)) { + if filter.Matches(divergence) { matching = append(matching, divergence) } } - slices.SortFunc(matching, func(a, b *pb.Divergence) int { + slices.SortFunc(matching, func(a, b model.Divergence) int { return cmp.Or( - cmp.Compare(a.GetService(), b.GetService()), - cmp.Compare(a.GetEnvironment(), b.GetEnvironment()), - cmp.Compare(a.GetDirection(), b.GetDirection()), + cmp.Compare(a.Service, b.Service), + cmp.Compare(a.Environment, b.Environment), + // Reversed, so ingress lists before egress. + cmp.Compare(b.Direction, a.Direction), ) }) 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/reporter.go b/internal/server/reporter/reporter.go similarity index 64% rename from internal/server/reporter.go rename to internal/server/reporter/reporter.go index b76e298..f325f83 100644 --- a/internal/server/reporter.go +++ b/internal/server/reporter/reporter.go @@ -1,4 +1,5 @@ -package server +// Package reporter reports a mirror's quarantines to the SPECTRE server. +package reporter import ( "context" @@ -7,9 +8,16 @@ import ( "github.com/block/spectre/internal/logger" "github.com/block/spectre/internal/model" "github.com/block/spectre/internal/server/client" - pb "github.com/block/spectre/spectrepb" ) +// Config configures how a mirror reports quarantines to the SPECTRE server. +type Config struct { + Client client.Config `embed:""` + Service string `required:"" help:"Service the mirror runs, reported to the SPECTRE server."` + Environment string `required:"" help:"Environment the mirror runs in, such as production-us-west-2, reported to the SPECTRE server."` + Revision string `required:"" help:"Commit of the candidate the mirror runs, reported to the SPECTRE server."` +} + // Source is a proxy whose quarantine a mirror reports. type Source struct { Direction model.Direction @@ -19,13 +27,13 @@ type Source struct { // Reporter sends each quarantine of a mirror's sources to the SPECTRE server. type Reporter struct { - config ReporterConfig + config Config client *client.Client sources []Source } -// NewReporter reports the quarantines of a mirror's sources through server. -func NewReporter(config ReporterConfig, server *client.Client, sources ...Source) *Reporter { +// New reports the quarantines of a mirror's sources through server. +func New(config Config, server *client.Client, sources ...Source) *Reporter { return &Reporter{config: config, client: server, sources: sources} } @@ -48,12 +56,12 @@ func (r *Reporter) reportSource(ctx context.Context, source Source) { case <-ctx.Done(): return } - divergence := &pb.Divergence{ + divergence := model.Divergence{ Service: r.config.Service, Environment: r.config.Environment, Revision: r.config.Revision, - Direction: directionMessage(source.Direction), - Quarantine: quarantineMessage(status), + Direction: source.Direction, + Quarantine: status, } if err := r.client.Quarantine(ctx, divergence); err != nil && ctx.Err() == nil { logger.FromContext(ctx).ErrorContext(ctx, "Failed to report quarantine to the SPECTRE server", "error", err) diff --git a/internal/server/reporter/reporter_test.go b/internal/server/reporter/reporter_test.go new file mode 100644 index 0000000..3cad8ed --- /dev/null +++ b/internal/server/reporter/reporter_test.go @@ -0,0 +1,72 @@ +package reporter_test + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + . "github.com/alecthomas/types/optional" + + "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/reporter" + "github.com/block/spectre/internal/server/storage" +) + +// quarantineProcedure is the protocol's path for reporting a quarantine. +const quarantineProcedure = "/spectre.v1.SpectreService/Quarantine" + +func TestReporterRetriesQuarantine(t *testing.T) { + reports := make(chan struct{}, 8) + var attempts atomic.Int32 + service := server.NewService(storage.NewMemory()).Handler() + httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + if request.URL.Path == quarantineProcedure && attempts.Add(1) == 1 { + http.Error(w, "unavailable", http.StatusServiceUnavailable) + return + } + service.ServeHTTP(w, request) + if request.URL.Path == quarantineProcedure { + reports <- struct{}{} + } + })) + t.Cleanup(httpServer.Close) + ingress := make(chan model.Quarantine, 1) + egress := make(chan model.Quarantine, 1) + config := reporter.Config{ + Client: client.Config{URL: httpServer.URL}, + Service: "users", + Environment: "production", + Revision: "abc123", + } + spectre := client.New(config.Client) + mirror := reporter.New(config, spectre, reporter.Source{Direction: model.Ingress, Quarantined: ingress}, reporter.Source{Direction: model.Egress, Quarantined: egress}) + ctx, cancel := context.WithCancel(t.Context()) + t.Cleanup(cancel) + go mirror.Run(ctx) + + quarantine := model.Quarantine{Since: time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC), Reason: model.QuarantineReason{ + Kind: model.QuarantineUnmatched, + Traffic: Some(model.Traffic{Direction: model.Egress, Protocol: model.ProtocolHTTP, Route: "POST api.example/v1/charges"}), + Detail: "no matching reference call", + }} + egress <- quarantine + <-reports + + quarantined, err := spectre.ListDivergences(t.Context(), model.DivergenceFilter{}) + assert.NoError(t, err) + expected := []model.Divergence{{ + Service: "users", + Environment: "production", + Revision: "abc123", + Direction: model.Egress, + Quarantine: quarantine, + }} + assert.Equal(t, expected, quarantined) + assert.Equal(t, int32(2), attempts.Load()) +} diff --git a/internal/server/server_test.go b/internal/server/server_test.go index da6cf2c..e9893ad 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -1,20 +1,15 @@ package server_test import ( - "context" - "net/http" "net/http/httptest" - "sync/atomic" "testing" "time" "connectrpc.com/connect" "github.com/alecthomas/assert/v2" - . "github.com/alecthomas/types/optional" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" - "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" @@ -149,58 +144,3 @@ func TestQuarantineRejectsInvalidDivergence(t *testing.T) { }) } } - -func TestReporterRetriesQuarantine(t *testing.T) { - reports := make(chan struct{}, 8) - var attempts atomic.Int32 - 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) - return - } - service.ServeHTTP(w, request) - if request.URL.Path == spectrepbconnect.SpectreServiceQuarantineProcedure { - reports <- struct{}{} - } - })) - t.Cleanup(httpServer.Close) - ingress := make(chan model.Quarantine, 1) - egress := make(chan model.Quarantine, 1) - config := server.ReporterConfig{ - Client: client.Config{URL: httpServer.URL}, - Service: "users", - Environment: "production", - Revision: "abc123", - } - reporter := server.NewReporter(config, client.New(config.Client), server.Source{Direction: model.Ingress, Quarantined: ingress}, server.Source{Direction: model.Egress, Quarantined: egress}) - ctx, cancel := context.WithCancel(t.Context()) - t.Cleanup(cancel) - go reporter.Run(ctx) - - since := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) - egress <- model.Quarantine{Since: since, Reason: model.QuarantineReason{ - Kind: model.QuarantineUnmatched, - Traffic: Some(model.Traffic{Direction: model.Egress, Protocol: model.ProtocolHTTP, Route: "POST api.example/v1/charges"}), - Detail: "no matching reference call", - }} - <-reports - - rpc := spectrepbconnect.NewSpectreServiceClient(httpServer.Client(), httpServer.URL) - quarantined, err := rpc.ListDivergences(t.Context(), connect.NewRequest(&pb.ListDivergencesRequest{})) - assert.NoError(t, err) - expected := &pb.ListDivergencesResponse{Divergences: []*pb.Divergence{{ - Service: "users", - Environment: "production", - Revision: "abc123", - Direction: pb.Direction_DIRECTION_EGRESS, - Quarantine: &pb.Quarantine{ - Since: timestamppb.New(since), - Kind: pb.Quarantine_KIND_UNMATCHED, - Traffic: &pb.Traffic{Direction: pb.Direction_DIRECTION_EGRESS, Protocol: "http", Route: "POST api.example/v1/charges"}, - Detail: "no matching reference call", - }, - }}} - assert.True(t, proto.Equal(expected, quarantined.Msg), "%v", quarantined.Msg) - assert.Equal(t, int32(2), attempts.Load()) -} diff --git a/internal/server/service.go b/internal/server/service.go index e6a59b3..69a991e 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -14,6 +14,7 @@ import ( "github.com/block/spectre/internal/logger" "github.com/block/spectre/internal/netaddr" "github.com/block/spectre/internal/server/storage" + "github.com/block/spectre/internal/server/wire" pb "github.com/block/spectre/spectrepb" "github.com/block/spectre/spectrepb/spectrepbconnect" ) @@ -37,8 +38,8 @@ func (s *Service) Handler() http.Handler { // Quarantine records that a mirror quarantined its candidate. 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 { + divergence, err := wire.DivergenceFromMessage(request.Msg.GetDivergence()) + if err != nil { return nil, connect.NewError(connect.CodeInvalidArgument, err) } if err := s.store.Record(ctx, divergence); err != nil { @@ -53,7 +54,11 @@ func (s *Service) ListDivergences(ctx context.Context, request *connect.Request[ if err != nil { return nil, connect.NewError(connect.CodeUnavailable, err) } - return connect.NewResponse(&pb.ListDivergencesResponse{Divergences: filterDivergences(divergences, request.Msg)}), nil + response := &pb.ListDivergencesResponse{} + for _, divergence := range filterDivergences(divergences, wire.FilterFromMessage(request.Msg)) { + response.Divergences = append(response.Divergences, wire.DivergenceMessage(divergence)) + } + return connect.NewResponse(response), nil } // ClearDivergences removes the divergences of one service environment. diff --git a/internal/server/storage/memory.go b/internal/server/storage/memory.go index 3d02722..45e0ed3 100644 --- a/internal/server/storage/memory.go +++ b/internal/server/storage/memory.go @@ -4,14 +4,14 @@ import ( "context" "sync" - pb "github.com/block/spectre/spectrepb" + "github.com/block/spectre/internal/model" ) -// Memory keeps divergences in memory, so a restart loses them. Stored messages -// are never mutated, so they may be shared. +// Memory keeps divergences in memory, so a restart loses them. Stored +// divergences share their slices with callers, who never mutate them. type Memory struct { mu sync.Mutex - divergences map[sourceKey]*pb.Divergence + divergences map[sourceKey]model.Divergence } var _ Store = (*Memory)(nil) @@ -19,20 +19,20 @@ 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 + direction model.Direction } // NewMemory returns an empty in-memory store. func NewMemory() *Memory { - return &Memory{divergences: map[sourceKey]*pb.Divergence{}} + return &Memory{divergences: map[sourceKey]model.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()} +func (m *Memory) Record(_ context.Context, divergence model.Divergence) error { + key := sourceKey{service: divergence.Service, environment: divergence.Environment, direction: divergence.Direction} m.mu.Lock() defer m.mu.Unlock() - if existing, ok := m.divergences[key]; ok && quarantinedAfter(existing, divergence) { + if existing, ok := m.divergences[key]; ok && quarantinedAfter(existing.Quarantine, divergence.Quarantine) { return nil } m.divergences[key] = divergence @@ -52,10 +52,10 @@ func (m *Memory) Clear(_ context.Context, service, environment string) error { } // List returns every stored divergence, in no particular order. -func (m *Memory) List(_ context.Context) ([]*pb.Divergence, error) { +func (m *Memory) List(_ context.Context) ([]model.Divergence, error) { m.mu.Lock() defer m.mu.Unlock() - divergences := make([]*pb.Divergence, 0, len(m.divergences)) + divergences := make([]model.Divergence, 0, len(m.divergences)) for _, divergence := range m.divergences { divergences = append(divergences, divergence) } diff --git a/internal/server/storage/redis.go b/internal/server/storage/redis.go index 8e0f718..cfd11d0 100644 --- a/internal/server/storage/redis.go +++ b/internal/server/storage/redis.go @@ -1,17 +1,19 @@ package storage import ( + "bytes" "context" + "encoding/json" "strconv" "github.com/alecthomas/errors" "github.com/redis/go-redis/v9" - "google.golang.org/protobuf/proto" - pb "github.com/block/spectre/spectrepb" + "github.com/block/spectre/internal/model" ) -// divergencesKey is the Redis hash that maps each sourceField to its divergence. +// divergencesKey is the Redis hash that maps each sourceField to its divergence +// as JSON. const divergencesKey = "spectre:divergences" // Redis keeps divergences in a Redis hash, so they outlive the SPECTRE server @@ -37,9 +39,9 @@ func NewRedis(ctx context.Context, config Config) (*Redis, error) { } // 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) +func (r *Redis) Record(ctx context.Context, divergence model.Divergence) error { + field := sourceField(divergence.Service, divergence.Environment, divergence.Direction) + value, err := json.Marshal(divergence) if err != nil { return errors.Wrap(err, "encode divergence") } @@ -53,11 +55,11 @@ func (r *Redis) Record(ctx context.Context, divergence *pb.Divergence) error { 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") + stored, err := decodeDivergence(existing) + if err != nil { + return err } - if quarantinedAfter(stored, divergence) { + if quarantinedAfter(stored.Quarantine, divergence.Quarantine) { return nil } } @@ -75,23 +77,23 @@ func (r *Redis) Record(ctx context.Context, divergence *pb.Divergence) error { // 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), + sourceField(service, environment, model.Ingress), + sourceField(service, environment, model.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) { +func (r *Redis) List(ctx context.Context) ([]model.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)) + divergences := make([]model.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") + divergence, err := decodeDivergence([]byte(value)) + if err != nil { + return nil, err } divergences = append(divergences, divergence) } @@ -105,6 +107,18 @@ func (r *Redis) Close() error { // 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() +func sourceField(service, environment string, direction model.Direction) string { + return strconv.Quote(service) + " " + strconv.Quote(environment) + " " + string(direction) +} + +// decodeDivergence rejects unknown fields, so a stored format that has drifted +// fails loudly instead of losing data. +func decodeDivergence(data []byte) (model.Divergence, error) { + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + var divergence model.Divergence + if err := decoder.Decode(&divergence); err != nil { + return model.Divergence{}, errors.Wrap(err, "decode divergence") + } + return divergence, nil } diff --git a/internal/server/storage/storage.go b/internal/server/storage/storage.go index af4df08..94777ae 100644 --- a/internal/server/storage/storage.go +++ b/internal/server/storage/storage.go @@ -8,7 +8,7 @@ import ( "github.com/alecthomas/errors" "github.com/redis/go-redis/v9" - pb "github.com/block/spectre/spectrepb" + "github.com/block/spectre/internal/model" ) // Backend names where the SPECTRE server keeps divergences. @@ -47,11 +47,11 @@ func (c Config) Validate() error { // 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 + Record(ctx context.Context, divergence model.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) + List(ctx context.Context) ([]model.Divergence, error) Close() error } @@ -71,6 +71,6 @@ func Open(ctx context.Context, config Config) (Store, error) { // 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()) +func quarantinedAfter(existing, incoming model.Quarantine) bool { + return existing.Since.After(incoming.Since) } diff --git a/internal/server/storage/storage_test.go b/internal/server/storage/storage_test.go index 1af0e28..0751600 100644 --- a/internal/server/storage/storage_test.go +++ b/internal/server/storage/storage_test.go @@ -9,12 +9,11 @@ import ( "time" "github.com/alecthomas/assert/v2" + . "github.com/alecthomas/types/optional" "github.com/alicebob/miniredis/v2" - "google.golang.org/protobuf/proto" - "google.golang.org/protobuf/types/known/timestamppb" + "github.com/block/spectre/internal/model" "github.com/block/spectre/internal/server/storage" - pb "github.com/block/spectre/spectrepb" ) func TestStores(t *testing.T) { @@ -73,6 +72,17 @@ func TestOpen(t *testing.T) { } } +func TestRedisRejectsUnknownFields(t *testing.T) { + server := miniredis.RunT(t) + store, err := storage.NewRedis(t.Context(), storage.Config{Storage: storage.BackendRedis, RedisURL: "redis://" + server.Addr()}) + assert.NoError(t, err) + t.Cleanup(func() { assert.NoError(t, store.Close()) }) + server.HSet("spectre:divergences", `"users" "production" ingress`, + `{"service":"users","quarantine":{"reason":{"kind":"divergent","traffic":{"direction":"ingress","method":"GET"}}}}`) + _, err = store.List(t.Context()) + assert.Contains(t, err.Error(), `unknown field "method"`) +} + func openRedis(t *testing.T) storage.Store { t.Helper() server := miniredis.RunT(t) @@ -83,13 +93,19 @@ func openRedis(t *testing.T) storage.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 := divergence("production", model.Ingress, first.Add(time.Minute)) + latest.Quarantine.Reason = model.QuarantineReason{ + Kind: model.QuarantineDivergent, + Traffic: Some(model.Traffic{Direction: model.Ingress, Protocol: model.ProtocolHTTP, Route: "GET /v1/users"}), + Node: Some("users"), + Differences: []string{"body.name"}, + } + egress := divergence("production", model.Egress, first) + for _, reported := range []model.Divergence{ + divergence("production", model.Ingress, first), latest, // A stale retry from before the mirror restarted. - divergence("production", pb.Direction_DIRECTION_INGRESS, first.Add(-time.Minute)), + divergence("production", model.Ingress, first.Add(-time.Minute)), egress, } { assert.NoError(t, store.Record(t.Context(), reported)) @@ -99,10 +115,10 @@ func testKeepsLatestQuarantine(t *testing.T, store storage.Store) { 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 := divergence("staging", model.Ingress, since) + for _, reported := range []model.Divergence{ + divergence("production", model.Ingress, since), + divergence("production", model.Egress, since), staging, } { assert.NoError(t, store.Record(t.Context(), reported)) @@ -115,9 +131,9 @@ func testClearsServiceEnvironment(t *testing.T, store storage.Store) { // 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{} + expected := []model.Divergence{} for i := range 20 { - expected = append(expected, divergence("environment-"+strconv.Itoa(i), pb.Direction_DIRECTION_INGRESS, since)) + expected = append(expected, divergence("environment-"+strconv.Itoa(i), model.Ingress, since)) } var wg sync.WaitGroup for _, reported := range expected { @@ -127,25 +143,24 @@ func testRecordsConcurrently(t *testing.T, store storage.Store) { assertStored(t, store, expected...) } -func divergence(environment string, direction pb.Direction, since time.Time) *pb.Divergence { - return &pb.Divergence{ +func divergence(environment string, direction model.Direction, since time.Time) model.Divergence { + return model.Divergence{ Service: "users", Environment: environment, Revision: "abc123", Direction: direction, - Quarantine: &pb.Quarantine{Since: timestamppb.New(since), Kind: pb.Quarantine_KIND_CAPACITY}, + Quarantine: model.Quarantine{Since: since, Reason: model.QuarantineReason{Kind: model.QuarantineCapacity}}, } } // assertStored checks the store holds exactly the expected divergences, in any order. -func assertStored(t *testing.T, store storage.Store, expected ...*pb.Divergence) { +func assertStored(t *testing.T, store storage.Store, expected ...model.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())) + bySource := func(a, b model.Divergence) int { + return cmp.Or(cmp.Compare(a.Environment, b.Environment), cmp.Compare(a.Direction, b.Direction)) } 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) + assert.Equal(t, slices.SortedFunc(slices.Values(expected), bySource), stored) } diff --git a/internal/server/wire.go b/internal/server/wire.go deleted file mode 100644 index ff02801..0000000 --- a/internal/server/wire.go +++ /dev/null @@ -1,117 +0,0 @@ -package server - -import ( - "github.com/alecthomas/errors" - . "github.com/alecthomas/types/optional" - "google.golang.org/protobuf/types/known/timestamppb" - - "github.com/block/spectre/internal/model" - pb "github.com/block/spectre/spectrepb" -) - -func quarantineMessage(status model.Quarantine) *pb.Quarantine { - reason := status.Reason - message := &pb.Quarantine{ - Since: timestamppb.New(status.Since), - Kind: kindMessage(reason.Kind), - Differences: reason.Differences, - Detail: reason.Detail, - } - if traffic, ok := reason.Traffic.Get(); ok { - message.Traffic = &pb.Traffic{ - Direction: directionMessage(traffic.Direction), - Protocol: string(traffic.Protocol), - Route: string(traffic.Route), - } - } - if node, ok := reason.Node.Get(); ok { - message.Node = &node - } - return message -} - -// ReasonFromMessage restores the reason a mirror reported. Protobuf getters -// return plain strings, so the wire format is where named types are restored. -func ReasonFromMessage(message *pb.Quarantine) (model.QuarantineReason, error) { - kind, err := kindFromMessage(message.GetKind()) - if err != nil { - return model.QuarantineReason{}, err - } - reason := model.QuarantineReason{ - Kind: kind, - Node: Zero(message.GetNode()), - Differences: message.GetDifferences(), - Detail: message.GetDetail(), - } - if traffic := message.GetTraffic(); traffic != nil { - direction, err := DirectionFromMessage(traffic.GetDirection()) - if err != nil { - return model.QuarantineReason{}, err - } - reason.Traffic = Some(model.Traffic{ - Direction: direction, - Protocol: model.Protocol(traffic.GetProtocol()), //nolint:typeboundary // protobuf getters return plain strings - Route: model.Route(traffic.GetRoute()), //nolint:typeboundary // protobuf getters return plain strings - }) - } - return reason, nil -} - -func kindMessage(kind model.QuarantineKind) pb.Quarantine_Kind { - switch kind { - case model.QuarantineDivergent: - return pb.Quarantine_KIND_DIVERGENT - case model.QuarantineUnable: - return pb.Quarantine_KIND_UNABLE - case model.QuarantineUnmatched: - return pb.Quarantine_KIND_UNMATCHED - case model.QuarantineUnlisted: - return pb.Quarantine_KIND_UNLISTED - case model.QuarantineCapacity: - return pb.Quarantine_KIND_CAPACITY - case model.QuarantinePairing: - return pb.Quarantine_KIND_PAIRING - } - return pb.Quarantine_KIND_UNSPECIFIED -} - -func kindFromMessage(kind pb.Quarantine_Kind) (model.QuarantineKind, error) { - switch kind { - case pb.Quarantine_KIND_DIVERGENT: - return model.QuarantineDivergent, nil - case pb.Quarantine_KIND_UNABLE: - return model.QuarantineUnable, nil - case pb.Quarantine_KIND_UNMATCHED: - return model.QuarantineUnmatched, nil - case pb.Quarantine_KIND_UNLISTED: - return model.QuarantineUnlisted, nil - case pb.Quarantine_KIND_CAPACITY: - return model.QuarantineCapacity, nil - case pb.Quarantine_KIND_PAIRING: - return model.QuarantinePairing, nil - case pb.Quarantine_KIND_UNSPECIFIED: - } - return "", errors.Errorf("unknown quarantine kind %s", kind) -} - -func directionMessage(direction model.Direction) pb.Direction { - switch direction { - case model.Ingress: - return pb.Direction_DIRECTION_INGRESS - case model.Egress: - return pb.Direction_DIRECTION_EGRESS - } - return pb.Direction_DIRECTION_UNSPECIFIED -} - -// DirectionFromMessage restores the direction a mirror reported. -func DirectionFromMessage(direction pb.Direction) (model.Direction, error) { - switch direction { - case pb.Direction_DIRECTION_INGRESS: - return model.Ingress, nil - case pb.Direction_DIRECTION_EGRESS: - return model.Egress, nil - case pb.Direction_DIRECTION_UNSPECIFIED: - } - return "", errors.Errorf("unknown direction %s", direction) -} diff --git a/internal/server/wire/wire.go b/internal/server/wire/wire.go new file mode 100644 index 0000000..8a6f7c6 --- /dev/null +++ b/internal/server/wire/wire.go @@ -0,0 +1,187 @@ +// Package wire converts between the SPECTRE service protocol and the domain +// model, so only the server's API boundary handles protocol messages. +package wire + +import ( + "github.com/alecthomas/errors" + . "github.com/alecthomas/types/optional" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/types/known/timestamppb" + + "github.com/block/spectre/internal/model" + pb "github.com/block/spectre/spectrepb" +) + +// DivergenceMessage returns the message that reports divergence. +func DivergenceMessage(divergence model.Divergence) *pb.Divergence { + return &pb.Divergence{ + Service: divergence.Service, + Environment: divergence.Environment, + Revision: divergence.Revision, + Direction: directionMessage(divergence.Direction), + Quarantine: quarantineMessage(divergence.Quarantine), + } +} + +// DivergenceFromMessage restores a reported divergence, checking that it +// identifies its source and quarantine so every store can rely on them. +func DivergenceFromMessage(message *pb.Divergence) (model.Divergence, error) { + if message.GetService() == "" || message.GetEnvironment() == "" || message.GetRevision() == "" { + return model.Divergence{}, errors.New("divergence needs a service, environment, and revision") + } + direction, err := directionFromMessage(message.GetDirection()) + if err != nil { + return model.Divergence{}, err + } + quarantine := message.GetQuarantine() + if quarantine.GetSince() == nil { + return model.Divergence{}, errors.New("divergence needs a quarantine with its start time") + } + reason, err := reasonFromMessage(quarantine) + if err != nil { + return model.Divergence{}, errors.Wrap(err, "quarantine") + } + return model.Divergence{ + Service: message.GetService(), + Environment: message.GetEnvironment(), + Revision: message.GetRevision(), + Direction: direction, + Quarantine: model.Quarantine{Reason: reason, Since: quarantine.GetSince().AsTime()}, + }, nil +} + +// DivergencesJSON returns divergences as the protocol's JSON list response, +// which is what the command line prints. +func DivergencesJSON(divergences []model.Divergence) ([]byte, error) { + response := &pb.ListDivergencesResponse{Divergences: make([]*pb.Divergence, 0, len(divergences))} + for _, divergence := range divergences { + response.Divergences = append(response.Divergences, DivergenceMessage(divergence)) + } + data, err := protojson.MarshalOptions{Multiline: true}.Marshal(response) + return data, errors.Wrap(err, "encode JSON") +} + +// FilterMessage returns the request that lists the divergences filter selects. +func FilterMessage(filter model.DivergenceFilter) *pb.ListDivergencesRequest { + return &pb.ListDivergencesRequest{ + Service: filter.Service.Ptr(), + Environment: filter.Environment.Ptr(), + Revision: filter.Revision.Ptr(), + Node: filter.Node.Ptr(), + } +} + +// FilterFromMessage restores the filter of a list request. +func FilterFromMessage(request *pb.ListDivergencesRequest) model.DivergenceFilter { + // Getters hide whether an optional field is set, so presence is a nil check. + return model.DivergenceFilter{ + Service: From(request.GetService(), request.Service != nil), + Environment: From(request.GetEnvironment(), request.Environment != nil), + Revision: From(request.GetRevision(), request.Revision != nil), + Node: From(request.GetNode(), request.Node != nil), + } +} + +func quarantineMessage(status model.Quarantine) *pb.Quarantine { + reason := status.Reason + message := &pb.Quarantine{ + Since: timestamppb.New(status.Since), + Kind: kindMessage(reason.Kind), + Differences: reason.Differences, + Detail: reason.Detail, + Node: reason.Node.Ptr(), + } + if traffic, ok := reason.Traffic.Get(); ok { + message.Traffic = &pb.Traffic{ + Direction: directionMessage(traffic.Direction), + Protocol: string(traffic.Protocol), + Route: string(traffic.Route), + } + } + return message +} + +// reasonFromMessage restores the reason a mirror reported. Protobuf getters +// return plain strings, so the wire format is where named types are restored. +func reasonFromMessage(message *pb.Quarantine) (model.QuarantineReason, error) { + kind, err := kindFromMessage(message.GetKind()) + if err != nil { + return model.QuarantineReason{}, err + } + reason := model.QuarantineReason{ + Kind: kind, + Node: Zero(message.GetNode()), + Differences: message.GetDifferences(), + Detail: message.GetDetail(), + } + if traffic := message.GetTraffic(); traffic != nil { + direction, err := directionFromMessage(traffic.GetDirection()) + if err != nil { + return model.QuarantineReason{}, err + } + reason.Traffic = Some(model.Traffic{ + Direction: direction, + Protocol: model.Protocol(traffic.GetProtocol()), //nolint:typeboundary // protobuf getters return plain strings + Route: model.Route(traffic.GetRoute()), //nolint:typeboundary // protobuf getters return plain strings + }) + } + return reason, nil +} +func kindMessage(kind model.QuarantineKind) pb.Quarantine_Kind { + switch kind { + case model.QuarantineDivergent: + return pb.Quarantine_KIND_DIVERGENT + case model.QuarantineUnable: + return pb.Quarantine_KIND_UNABLE + case model.QuarantineUnmatched: + return pb.Quarantine_KIND_UNMATCHED + case model.QuarantineUnlisted: + return pb.Quarantine_KIND_UNLISTED + case model.QuarantineCapacity: + return pb.Quarantine_KIND_CAPACITY + case model.QuarantinePairing: + return pb.Quarantine_KIND_PAIRING + } + return pb.Quarantine_KIND_UNSPECIFIED +} + +func kindFromMessage(kind pb.Quarantine_Kind) (model.QuarantineKind, error) { + switch kind { + case pb.Quarantine_KIND_DIVERGENT: + return model.QuarantineDivergent, nil + case pb.Quarantine_KIND_UNABLE: + return model.QuarantineUnable, nil + case pb.Quarantine_KIND_UNMATCHED: + return model.QuarantineUnmatched, nil + case pb.Quarantine_KIND_UNLISTED: + return model.QuarantineUnlisted, nil + case pb.Quarantine_KIND_CAPACITY: + return model.QuarantineCapacity, nil + case pb.Quarantine_KIND_PAIRING: + return model.QuarantinePairing, nil + case pb.Quarantine_KIND_UNSPECIFIED: + } + return "", errors.Errorf("unknown quarantine kind %s", kind) +} + +func directionMessage(direction model.Direction) pb.Direction { + switch direction { + case model.Ingress: + return pb.Direction_DIRECTION_INGRESS + case model.Egress: + return pb.Direction_DIRECTION_EGRESS + } + return pb.Direction_DIRECTION_UNSPECIFIED +} + +// directionFromMessage restores the direction a mirror reported. +func directionFromMessage(direction pb.Direction) (model.Direction, error) { + switch direction { + case pb.Direction_DIRECTION_INGRESS: + return model.Ingress, nil + case pb.Direction_DIRECTION_EGRESS: + return model.Egress, nil + case pb.Direction_DIRECTION_UNSPECIFIED: + } + return "", errors.Errorf("unknown direction %s", direction) +} diff --git a/internal/server/wire/wire_test.go b/internal/server/wire/wire_test.go new file mode 100644 index 0000000..8a4a105 --- /dev/null +++ b/internal/server/wire/wire_test.go @@ -0,0 +1,67 @@ +package wire_test + +import ( + "testing" + "time" + + "github.com/alecthomas/assert/v2" + . "github.com/alecthomas/types/optional" + + "github.com/block/spectre/internal/model" + "github.com/block/spectre/internal/server/wire" +) + +func TestDivergenceRoundTrip(t *testing.T) { + since := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) + tests := []struct { + name string + divergence model.Divergence + }{ + {name: "Divergent", divergence: model.Divergence{ + Service: "users", + Environment: "production", + Revision: "abc123", + Direction: model.Ingress, + Quarantine: model.Quarantine{Since: since, Reason: model.QuarantineReason{ + Kind: model.QuarantineDivergent, + Traffic: Some(model.Traffic{Direction: model.Ingress, Protocol: model.ProtocolHTTP, Route: "GET /v1/users"}), + Node: Some("users"), + Differences: []string{"body.name"}, + }}, + }}, + {name: "WithoutTraffic", divergence: model.Divergence{ + Service: "users", + Environment: "staging", + Revision: "abc123", + Direction: model.Egress, + Quarantine: model.Quarantine{Since: since, Reason: model.QuarantineReason{ + Kind: model.QuarantineCapacity, + Detail: "candidate concurrency limit exceeded", + }}, + }}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + restored, err := wire.DivergenceFromMessage(wire.DivergenceMessage(test.divergence)) + assert.NoError(t, err) + assert.Equal(t, test.divergence, restored) + }) + } +} + +func TestFilterRoundTrip(t *testing.T) { + tests := []struct { + name string + filter model.DivergenceFilter + }{ + {name: "Unset"}, + {name: "Set", filter: model.DivergenceFilter{Service: Some("users"), Environment: Some("production"), Revision: Some("abc123"), Node: Some("users")}}, + // An empty filter value is set, and matches only empty values. + {name: "Empty", filter: model.DivergenceFilter{Node: Some("")}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.filter, wire.FilterFromMessage(wire.FilterMessage(test.filter))) + }) + } +} diff --git a/internal/supervisor/config.go b/internal/supervisor/config.go index 1332324..042402f 100644 --- a/internal/supervisor/config.go +++ b/internal/supervisor/config.go @@ -9,8 +9,8 @@ import ( "github.com/alecthomas/kong" "github.com/block/spectre/internal/descriptors" - "github.com/block/spectre/internal/server" "github.com/block/spectre/internal/server/client" + "github.com/block/spectre/internal/server/reporter" ) // Config contains the command-line configuration for the supervisor. @@ -47,7 +47,7 @@ type Config struct { // so it can only reach dependencies through the egress proxy. SeccompEgress bool `name:"seccomp-egress" negatable:"" default:"true" help:"Block the candidate from opening IPv4 and IPv6 sockets with a seccomp filter, leaving only the egress proxy's unix socket."` // Server reports the mirror's quarantines to the SPECTRE server. - Server server.ReporterConfig `embed:""` + Server reporter.Config `embed:""` // StartupTimeout bounds how long each process has to start listening. StartupTimeout time.Duration `default:"30s" help:"Maximum time for each process to begin accepting connections."` // ShutdownTimeout bounds graceful shutdown before processes are killed. @@ -79,7 +79,7 @@ func NewConfig() Config { config := Config{ Candidate: placeholder, Reference: placeholder, - Server: server.ReporterConfig{ + Server: reporter.Config{ Client: client.Config{URL: placeholder}, Service: placeholder, Environment: placeholder, @@ -92,7 +92,7 @@ func NewConfig() Config { config.Candidate = "" config.Reference = "" // Every reporter field is required, so none has a default to keep. - config.Server = server.ReporterConfig{} + config.Server = reporter.Config{} return config } diff --git a/internal/supervisor/supervisor.go b/internal/supervisor/supervisor.go index a335f7a..dfebdce 100644 --- a/internal/supervisor/supervisor.go +++ b/internal/supervisor/supervisor.go @@ -19,8 +19,8 @@ import ( "github.com/block/spectre/internal/model" "github.com/block/spectre/internal/netaddr" "github.com/block/spectre/internal/protocol" - "github.com/block/spectre/internal/server" "github.com/block/spectre/internal/server/client" + "github.com/block/spectre/internal/server/reporter" "github.com/block/spectre/internal/state" ) @@ -214,9 +214,9 @@ func (s *Supervisor) startAll(ctx context.Context) error { // startReporter reports both proxies' quarantines to the SPECTRE server. // Reporting never stops the mirror. func (s *Supervisor) startReporter(ctx context.Context) { - reporter := server.NewReporter(s.config.Server, client.New(s.config.Server.Client), - server.Source{Direction: model.Ingress, Quarantined: s.ingress.Candidates().Quarantined()}, - server.Source{Direction: model.Egress, Quarantined: s.egress.Candidates().Quarantined()}, + reporter := reporter.New(s.config.Server, client.New(s.config.Server.Client), + reporter.Source{Direction: model.Ingress, Quarantined: s.ingress.Candidates().Quarantined()}, + reporter.Source{Direction: model.Egress, Quarantined: s.egress.Candidates().Quarantined()}, ) go reporter.Run(logger.WithLogger(ctx, s.log)) s.log.InfoContext(ctx, "Reporting to the SPECTRE server", "url", s.config.Server.Client.URL)