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)