diff --git a/coderd/x/nats/natsbench/doc.go b/coderd/x/nats/natsbench/doc.go new file mode 100644 index 0000000000..65042706c5 --- /dev/null +++ b/coderd/x/nats/natsbench/doc.go @@ -0,0 +1,42 @@ +// Command natsbench benchmarks Coder's NATS-backed pubsub +// (github.com/coder/coder/v2/coderd/x/nats) under high fan-out load. +// +// A run publishes a configurable total number of messages across a set +// of publishers, subjects, subscribers, and replica nodes, then reports +// two logical throughput metrics: +// +// - Pubs/sec: messages published divided by the publish duration. +// - Deliveries/sec: messages delivered divided by the delivery +// duration. Every message is delivered to every subscriber on its +// subject, so fan-out makes deliveries exceed publishes. This is +// logical throughput, not physical bandwidth. +// +// Correctness rules: +// +// - Production defaults: the benchmark measures the pubsub as Coder +// configures it. It does not autotune queue or pending limits to +// avoid drops; sane defaults are the thing under test. +// - Drops as a metric: each subscriber's expected delivery count is +// computed up front, so the exact, complete loss is Expected minus +// Delivered. Dropped messages are reported in the Drops column, not +// treated as a failure; a run that drops still reports the +// throughput it achieved. Only a real error (publish failure, +// cancellation, hard timeout) invalidates a run. +// - Delivery completion: a zero-drop run finishes precisely when every +// subscriber reaches its expected count. A run that dropped messages +// can never reach that count, so the deliver phase instead completes +// by quiescence: once the delivery counter stays flat for a fixed +// settle window, the phase ends, and the rate is measured to the +// last observed delivery so the idle wait stays out of it. +// - Readiness gate: in multi-replica runs, cross-route delivery is +// silent when subscription interest has not yet propagated. Before +// the measured phase, in-band probe messages prove that every +// publisher node can reach every subscriber on its subjects. +// - Bounded phases: every wait is bounded by Config.Timeout and fails +// with diagnostics (per-subscriber shortfalls, per-node server +// stats, goroutine dump) instead of hanging. +// +// The command runs the default scenario matrix, one named scenario, or +// a custom shape, and renders the grouped markdown report to stdout. +// The heavy benchmarks only run through this command, never in CI. +package main diff --git a/coderd/x/nats/natsbench/main.go b/coderd/x/nats/natsbench/main.go new file mode 100644 index 0000000000..373eb4bfc7 --- /dev/null +++ b/coderd/x/nats/natsbench/main.go @@ -0,0 +1,199 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "io" + "os" + "os/signal" + "syscall" + "time" + + "golang.org/x/xerrors" + + "cdr.dev/slog/v3" + "cdr.dev/slog/v3/sloggers/sloghuman" +) + +func main() { + // Cancel running scenarios on SIGINT/SIGTERM so the per-phase + // selects unwind cleanly instead of requiring kill -9. stop() is + // called explicitly rather than deferred so os.Exit below does not + // skip it. + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + err := runCLI(ctx, os.Args, os.Stdout, os.Stderr) + stop() + if err != nil { + _, _ = fmt.Fprintf(os.Stderr, "natsbench: %v\n", err) + os.Exit(1) + } +} + +// runCLI runs the natsbench command-line interface. args is the full +// argument vector (args[0] is the program name). The grouped markdown +// report is written to stdout and progress and logs to stderr. With no +// flags it runs the default scenario matrix; -scenario runs one named +// scenario; any custom shape flag runs a single custom configuration. +// It returns an error when the flags are invalid or any run fails. +func runCLI(ctx context.Context, args []string, stdout, stderr io.Writer) error { + fs := flag.NewFlagSet(args[0], flag.ContinueOnError) + fs.SetOutput(stderr) + var ( + run cliRun + list = fs.Bool("list", false, "list default scenarios and exit") + verbose = fs.Bool("v", false, "enable debug logging") + ) + fs.StringVar(&run.scenarioName, "scenario", "", "run one named default scenario (see -list)") + fs.IntVar(&run.messages, "messages", 0, "total messages across all publishers (0 keeps scenario defaults, or 100000 for custom runs)") + fs.IntVar(&run.payload, "payload", Payload8KB, "payload size in bytes (custom run)") + fs.IntVar(&run.subjects, "subjects", 10, "number of subjects (custom run)") + fs.IntVar(&run.publishers, "publishers", 10, "number of publishers (custom run)") + fs.IntVar(&run.subscribers, "subscribers", 50, "number of subscribers (custom run)") + fs.IntVar(&run.replicas, "replicas", 1, "number of embedded pubsub nodes (custom run)") + fs.IntVar(&run.publishConns, "publish-conns", DefaultConns, "publisher connection pool size (applies to every run)") + fs.IntVar(&run.subscribeConns, "subscribe-conns", DefaultConns, "subscriber connection pool size (applies to every run)") + fs.Int64Var(&run.seed, "seed", DefaultSeed, "seed for pseudorandom node placement (applies to every run); same seed reproduces the same placement") + fs.DurationVar(&run.timeout, "timeout", 2*time.Minute, "per-phase timeout") + if err := fs.Parse(args[1:]); err != nil { + if errors.Is(err, flag.ErrHelp) { + return nil + } + return err + } + fs.Visit(func(f *flag.Flag) { + switch f.Name { + case "payload", "subjects", "publishers", "subscribers", "replicas": + run.shapeFlagSet = true + } + }) + + if *list { + for _, sc := range DefaultScenarios() { + c := sc.Config + _, _ = fmt.Fprintf(stdout, "%s: messages=%d payload=%d subjects=%d publishers=%d subscribers=%d replicas=%d\n", + sc.Name, c.Messages, c.PayloadSize, c.Subjects, c.Publishers, c.Subscribers, c.Replicas) + } + return nil + } + + scenarios, err := run.scenarios() + if err != nil { + return err + } + + level := slog.LevelWarn + if *verbose { + level = slog.LevelDebug + } + logger := slog.Make(sloghuman.Sink(stderr)).Leveled(level) + + // Scenarios run sequentially so they never compete for CPU, memory, + // or the network stack and skew each other's numbers. + results := make([]ScenarioResult, 0, len(scenarios)) + failed := false + for _, sc := range scenarios { + // Stop launching scenarios once interrupted, rather than + // letting each remaining Run fail with a confusing topology + // error from the canceled context. + if err := ctx.Err(); err != nil { + return xerrors.Errorf("interrupted before %s: %w", sc.Name, err) + } + _, _ = fmt.Fprintf(stderr, "running %s...\n", sc.Name) + res, runErr := Run(ctx, logger, sc.Config) + results = append(results, ScenarioResult{Scenario: sc, Result: res, Err: runErr}) + if runErr != nil { + failed = true + _, _ = fmt.Fprintf(stderr, "%s failed: %v\n", sc.Name, runErr) + continue + } + _, _ = fmt.Fprintf(stderr, "%s: pubs/sec=%.0f deliveries/sec=%.0f\n", + sc.Name, res.PubsPerSec, res.DeliveriesPerSec) + } + + if err := RenderMarkdown(stdout, results); err != nil { + return xerrors.Errorf("render report: %w", err) + } + if failed { + return xerrors.New("one or more runs failed; see report") + } + return nil +} + +// cliRun holds the parsed flag values that select the scenarios to run. +type cliRun struct { + scenarioName string + // shapeFlagSet is true when any custom-shape flag was passed, which + // selects a single custom run. + shapeFlagSet bool + messages int + payload int + subjects int + publishers int + subscribers int + replicas int + publishConns int + subscribeConns int + seed int64 + timeout time.Duration +} + +// scenarios resolves the parsed flags into the scenarios to run: one +// named default scenario, a single custom shape when any shape flag was +// set, or the full default matrix. The per-phase timeout is applied to +// every scenario, and -messages overrides the message count when set. +func (c cliRun) scenarios() ([]Scenario, error) { + switch { + case c.scenarioName != "": + if c.shapeFlagSet { + return nil, xerrors.New("-scenario and custom shape flags are mutually exclusive") + } + for _, sc := range DefaultScenarios() { + if sc.Name != c.scenarioName { + continue + } + if c.messages > 0 { + sc.Config.Messages = c.messages + } + sc.Config.PublishConns = c.publishConns + sc.Config.SubscribeConns = c.subscribeConns + sc.Config.Seed = c.seed + sc.Config.Timeout = c.timeout + return []Scenario{sc}, nil + } + return nil, xerrors.Errorf("unknown scenario %q; use -list to see available scenarios", c.scenarioName) + case c.shapeFlagSet: + messages := c.messages + if messages <= 0 { + messages = DefaultMessages + } + return []Scenario{{ + Name: "custom", + Config: Config{ + Messages: messages, + PayloadSize: c.payload, + Subjects: c.subjects, + Publishers: c.publishers, + Subscribers: c.subscribers, + Replicas: c.replicas, + PublishConns: c.publishConns, + SubscribeConns: c.subscribeConns, + Seed: c.seed, + Timeout: c.timeout, + }, + }}, nil + default: + scenarios := DefaultScenarios() + for i := range scenarios { + if c.messages > 0 { + scenarios[i].Config.Messages = c.messages + } + scenarios[i].Config.PublishConns = c.publishConns + scenarios[i].Config.SubscribeConns = c.subscribeConns + scenarios[i].Config.Seed = c.seed + scenarios[i].Config.Timeout = c.timeout + } + return scenarios, nil + } +} diff --git a/coderd/x/nats/natsbench/main_internal_test.go b/coderd/x/nats/natsbench/main_internal_test.go new file mode 100644 index 0000000000..ddd975aed6 --- /dev/null +++ b/coderd/x/nats/natsbench/main_internal_test.go @@ -0,0 +1,88 @@ +package main + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/coder/coder/v2/testutil" +) + +func TestCLIScenarios(t *testing.T) { + t.Parallel() + + t.Run("DefaultMatrix", func(t *testing.T) { + t.Parallel() + got, err := cliRun{timeout: testutil.WaitShort, publishConns: DefaultConns, subscribeConns: DefaultConns}.scenarios() + require.NoError(t, err) + require.Len(t, got, len(DefaultScenarios())) + for _, sc := range got { + require.Equal(t, testutil.WaitShort, sc.Config.Timeout) + require.Equal(t, DefaultConns, sc.Config.PublishConns) + require.Equal(t, DefaultConns, sc.Config.SubscribeConns) + } + }) + + t.Run("ConnOverride", func(t *testing.T) { + t.Parallel() + got, err := cliRun{timeout: testutil.WaitShort, publishConns: 1, subscribeConns: 1}.scenarios() + require.NoError(t, err) + for _, sc := range got { + require.Equal(t, 1, sc.Config.PublishConns) + require.Equal(t, 1, sc.Config.SubscribeConns) + } + }) + + t.Run("MessageOverride", func(t *testing.T) { + t.Parallel() + got, err := cliRun{messages: 5, timeout: testutil.WaitShort}.scenarios() + require.NoError(t, err) + for _, sc := range got { + require.Equal(t, 5, sc.Config.Messages) + } + }) + + t.Run("NamedScenario", func(t *testing.T) { + t.Parallel() + got, err := cliRun{scenarioName: "8KiB-1r", messages: 9, timeout: testutil.WaitShort}.scenarios() + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, "8KiB-1r", got[0].Name) + require.Equal(t, 9, got[0].Config.Messages) + }) + + t.Run("UnknownScenario", func(t *testing.T) { + t.Parallel() + _, err := cliRun{scenarioName: "nope", timeout: testutil.WaitShort}.scenarios() + require.Error(t, err) + }) + + t.Run("CustomShape", func(t *testing.T) { + t.Parallel() + got, err := cliRun{ + shapeFlagSet: true, + payload: Payload64KB, + subjects: 3, + publishers: 4, + subscribers: 8, + replicas: 2, + publishConns: DefaultConns, + subscribeConns: DefaultConns, + timeout: testutil.WaitShort, + }.scenarios() + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, "custom", got[0].Name) + // Custom runs default to the standard message count. + require.Equal(t, DefaultMessages, got[0].Config.Messages) + require.Equal(t, Payload64KB, got[0].Config.PayloadSize) + require.Equal(t, 2, got[0].Config.Replicas) + require.Equal(t, DefaultConns, got[0].Config.PublishConns) + }) + + t.Run("ScenarioAndShapeConflict", func(t *testing.T) { + t.Parallel() + _, err := cliRun{scenarioName: "8KiB-1r", shapeFlagSet: true, timeout: testutil.WaitShort}.scenarios() + require.Error(t, err) + }) +} diff --git a/coderd/x/nats/natsbench/natsbench.go b/coderd/x/nats/natsbench/natsbench.go new file mode 100644 index 0000000000..798b802639 --- /dev/null +++ b/coderd/x/nats/natsbench/natsbench.go @@ -0,0 +1,171 @@ +package main + +import ( + "context" + "time" + + natsserver "github.com/nats-io/nats-server/v2/server" + "golang.org/x/xerrors" + + "cdr.dev/slog/v3" +) + +const ( + // DefaultMessages is the default total message count for a run. + DefaultMessages = 100000 + // Payload8KB and Payload64KB are the standard matrix payload sizes. + Payload8KB = 8 << 10 + Payload64KB = 64 << 10 +) + +// Config describes one benchmark run. Run validates a fully populated +// config and applies no defaults of its own (the CLI fills defaults +// before calling Run), so every required field must be set. The +// optional tuning fields below are passed straight through to +// nats.Options, where the nats package applies its own zero-value +// defaults. The benchmark deliberately does NOT autotune these knobs: +// it measures the pubsub as Coder configures it in production, so any +// dropped messages are reported as a metric rather than tuned away. +type Config struct { + // Messages is the TOTAL number of messages across all publishers, + // split evenly with the remainder assigned to publisher 0. Must be + // at least 1. + Messages int + // PayloadSize is the benchmark message size in bytes. + PayloadSize int + // Subjects is the number of distinct subjects ("bench."). + Subjects int + // Publishers is the number of concurrent publishers. Publisher i + // publishes to subject i % Subjects from node i % Replicas. + Publishers int + // Subscribers is the number of subscribers. Subscriber j listens on + // subject j % Subjects from node j % Replicas. + Subscribers int + // Replicas is the number of embedded pubsub nodes. 1 runs a single + // node; greater values run a fully meshed cluster. + Replicas int + // Seed selects the pseudorandom node placement. The same seed + // reproduces the same placement for a given shape; change it to + // sample a different placement. Zero is a valid, deterministic + // seed. + Seed int64 + + // InProcess uses in-process server connections instead of TCP + // loopback. + InProcess bool + // PublishConns and SubscribeConns configure the pubsub connection + // pools. Zero passes through to nats.Options, which defaults each + // pool to a single connection (the production default). + PublishConns int + SubscribeConns int + // LocalQueueMsgs overrides the per-subscription NATS pending + // message limit. Zero uses the nats package production default; + // set it only for sensitivity analysis. + LocalQueueMsgs int + // LocalQueueBytes overrides the per-subscription NATS pending byte + // limit. Zero uses the nats package production default. + LocalQueueBytes int + // MaxPending overrides the embedded server's per-client outbound + // pending byte budget. Zero uses the nats package production + // default. + MaxPending int64 + // Timeout bounds each phase (readiness, publish, deliver). It must + // be positive. + Timeout time.Duration +} + +// Result reports one run's exact accounting and throughput. +type Result struct { + // Config is the fully resolved configuration the run used, + // including derived sizing. + Config Config + + // Published is the number of successfully published messages. + Published int64 + // Delivered is the number of benchmark messages observed across all + // subscribers. Fan-out makes this exceed Published whenever a + // subject has multiple subscribers; it is logical throughput, not + // physical bandwidth. + Delivered int64 + // Expected is the total number of benchmark deliveries the plan + // requires: the sum of every subscriber's expected count. It is the + // exact denominator for the drop rate. + Expected int64 + // Drops is Expected minus Delivered: the number of deliveries that + // never arrived. It is the authoritative, complete loss count, since + // the ErrDroppedMessages signal coalesces and cross-node routed loss + // is silent. Drops are a reported metric, not a failure: a run with + // drops still produces trustworthy throughput numbers for what it did + // deliver. + Drops int64 + + // ConvergenceDuration is how long the readiness gate took to + // propagate subscription interest across the cluster, measured from + // the first probe (right after all subscriptions are registered) to + // full propagation. Zero for single-node runs, which need no gate. + ConvergenceDuration time.Duration + + // PublishDuration spans the hot start to the last publisher + // finishing, including the final flush. DeliverDuration spans the + // hot start to the last delivery: for a zero-drop run that is the + // last subscriber reaching its expected count; for a run that dropped + // messages it is the last observed counter progress before the + // settle window elapsed, so the settle delay stays out of the rate. + PublishDuration time.Duration + DeliverDuration time.Duration + + // PubsPerSec is Published / PublishDuration. DeliveriesPerSec is + // Delivered / DeliverDuration. + PubsPerSec float64 + DeliveriesPerSec float64 +} + +// validate rejects configurations the engine cannot run. +func (c Config) validate() error { + if c.Messages < 1 { + return xerrors.Errorf("messages must be at least 1, got %d", c.Messages) + } + if c.PayloadSize < 1 { + return xerrors.Errorf("payload size must be at least 1, got %d", c.PayloadSize) + } + if c.PayloadSize > natsserver.MAX_PAYLOAD_SIZE { + return xerrors.Errorf("payload size %d exceeds the NATS max payload %d", c.PayloadSize, natsserver.MAX_PAYLOAD_SIZE) + } + if c.Subjects < 1 { + return xerrors.Errorf("subjects must be at least 1, got %d", c.Subjects) + } + if c.Publishers < 1 { + return xerrors.Errorf("publishers must be at least 1, got %d", c.Publishers) + } + if c.Subscribers < 1 { + return xerrors.Errorf("subscribers must be at least 1, got %d", c.Subscribers) + } + if c.Replicas < 1 { + return xerrors.Errorf("replicas must be at least 1, got %d", c.Replicas) + } + if c.Timeout <= 0 { + return xerrors.Errorf("timeout must be positive, got %s", c.Timeout) + } + return nil +} + +// Run executes one benchmark run: build the deterministic plan, start +// the topology, and drive the workload. The config must be fully +// populated; Run applies no defaults. On failure it returns any partial +// Result alongside the error for diagnostics. Dropped messages are not +// a failure: a successful Result may still report a nonzero Drops count. +func Run(ctx context.Context, logger slog.Logger, cfg Config) (*Result, error) { + if err := cfg.validate(); err != nil { + return nil, xerrors.Errorf("validate config: %w", err) + } + + pl := buildPlan(cfg) + + top, err := buildTopology(ctx, logger, cfg) + if err != nil { + return nil, xerrors.Errorf("build topology: %w", err) + } + defer top.closeAll() + + return runWorkload(ctx, logger, top, pl, cfg) +} diff --git a/coderd/x/nats/natsbench/natsbench_internal_test.go b/coderd/x/nats/natsbench/natsbench_internal_test.go new file mode 100644 index 0000000000..9c4322bdd1 --- /dev/null +++ b/coderd/x/nats/natsbench/natsbench_internal_test.go @@ -0,0 +1,106 @@ +package main + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/coder/coder/v2/testutil" +) + +func TestConfigValidate(t *testing.T) { + t.Parallel() + + base := Config{ + Messages: 100, PayloadSize: 1024, Subjects: 1, Publishers: 1, Subscribers: 1, Replicas: 1, + Timeout: testutil.WaitShort, + } + require.NoError(t, base.validate()) + + cases := []struct { + name string + mutate func(*Config) + }{ + {"NoPayload", func(c *Config) { c.PayloadSize = 0 }}, + {"OversizedPayload", func(c *Config) { c.PayloadSize = 2 << 20 }}, + {"NoSubjects", func(c *Config) { c.Subjects = 0 }}, + {"NoPublishers", func(c *Config) { c.Publishers = 0 }}, + {"NoSubscribers", func(c *Config) { c.Subscribers = 0 }}, + {"NoReplicas", func(c *Config) { c.Replicas = 0 }}, + {"NegativeMessages", func(c *Config) { c.Messages = -1 }}, + {"NoTimeout", func(c *Config) { c.Timeout = 0 }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + cfg := base + tc.mutate(&cfg) + require.Error(t, cfg.validate()) + }) + } +} + +func TestRunSingleNode(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitLong) + logger := testutil.Logger(t) + + cfg := Config{ + Messages: 1000, + PayloadSize: 1024, + Subjects: 2, + Publishers: 4, + Subscribers: 8, + Replicas: 1, + InProcess: true, + Timeout: testutil.WaitLong, + } + res, err := Run(ctx, logger, cfg) + require.NoError(t, err) + + pl := buildPlan(cfg) + require.EqualValues(t, cfg.Messages, res.Published) + require.EqualValues(t, pl.totalExpected, res.Expected) + require.EqualValues(t, pl.totalExpected, res.Delivered) + require.Greater(t, res.Delivered, res.Published, "fan-out must exceed publishes") + require.Zero(t, res.Drops) + require.Greater(t, res.PubsPerSec, 0.0) + require.Greater(t, res.DeliveriesPerSec, 0.0) + require.Greater(t, res.DeliverDuration, time.Duration(0)) + require.GreaterOrEqual(t, res.DeliverDuration, res.PublishDuration) +} + +func TestRunCluster(t *testing.T) { + t.Parallel() + if testing.Short() { + t.SkipNow() + } + + ctx := testutil.Context(t, testutil.WaitSuperLong) + logger := testutil.Logger(t) + + // Random node placement across 3 replicas makes cross-node + // delivery the common case, exercising route propagation through + // the readiness gate. + cfg := Config{ + Messages: 600, + PayloadSize: 512, + Subjects: 2, + Publishers: 4, + Subscribers: 6, + Replicas: 3, + Timeout: testutil.WaitLong, + } + res, err := Run(ctx, logger, cfg) + require.NoError(t, err) + + pl := buildPlan(cfg) + require.EqualValues(t, cfg.Messages, res.Published) + require.EqualValues(t, pl.totalExpected, res.Expected) + require.EqualValues(t, pl.totalExpected, res.Delivered) + require.Zero(t, res.Drops) + require.Greater(t, res.PubsPerSec, 0.0) + require.Greater(t, res.DeliveriesPerSec, 0.0) +} diff --git a/coderd/x/nats/natsbench/plan.go b/coderd/x/nats/natsbench/plan.go new file mode 100644 index 0000000000..3725e0f41d --- /dev/null +++ b/coderd/x/nats/natsbench/plan.go @@ -0,0 +1,122 @@ +package main + +import ( + "fmt" + "math/rand/v2" + "slices" +) + +// plan assigns each publisher and subscriber to a subject round-robin by +// index (index % Subjects), places each one on a pseudorandom replica +// node (seeded for reproducibility), and precomputes each subscriber's +// expected delivery count so the workload can do exact accounting. +// +// Node placement is random to model a real deployment, where each +// client connects to an arbitrary replica. Round-robin node placement +// would instead co-locate publisher i and subscriber i on the same +// node and, whenever Replicas divides Subjects, keep a subject's entire +// traffic local, which understates cross-node routing cost. +// +// Worked example with Subjects=2, Publishers=3, Subscribers=4, +// Messages=100. Subject assignment is round-robin (index % Subjects) +// and the 100 messages split 34/33/33 (the remainder lands on publisher +// 0): +// +// publisher 0 -> subject 0, sends 34 +// publisher 1 -> subject 1, sends 33 +// publisher 2 -> subject 0, sends 33 +// +// subject 0 receives 34+33 = 67 (publishers 0 and 2) +// subject 1 receives 33 (publisher 1) +// +// subscriber 0 -> subject 0, expects 67 +// subscriber 1 -> subject 1, expects 33 +// subscriber 2 -> subject 0, expects 67 +// subscriber 3 -> subject 1, expects 33 +// +// totalExpected = 67+33+67+33 = 200. It exceeds the 100 published +// messages because each subject has two subscribers, so every message +// is delivered twice. That fan-out is the logical delivery count the +// benchmark measures. Each publisher's and subscriber's node is drawn +// independently from [0, Replicas). +type plan struct { + // perPubMsgs[i] is the number of messages publisher i sends. + perPubMsgs []int + // pubSubject[i] is publisher i's subject index. + pubSubject []int + // pubNode[i] is publisher i's node index. + pubNode []int + // subSubject[j] is subscriber j's subject index. + subSubject []int + // subNode[j] is subscriber j's node index. + subNode []int + // expectPerSub[j] is how many benchmark messages subscriber j must + // observe: the total sent to its subject. + expectPerSub []int + // totalExpected is the sum of expectPerSub. + totalExpected int + // pubNodes and subNodes are the sorted distinct node indexes that + // host at least one publisher or subscriber. Several publishers or + // subscribers can share a node, so these dedupe pubNode / subNode + // for callers that act once per node (for example flushing). + pubNodes []int + subNodes []int +} + +// buildPlan computes the workload assignment for cfg. Subject and +// message assignment is deterministic; node placement is pseudorandom +// but fully determined by cfg.Seed, so a given config reproduces the +// same plan. cfg must already be validated: all counts are at least 1. +func buildPlan(cfg Config) plan { + pl := plan{ + perPubMsgs: make([]int, cfg.Publishers), + pubSubject: make([]int, cfg.Publishers), + pubNode: make([]int, cfg.Publishers), + subSubject: make([]int, cfg.Subscribers), + subNode: make([]int, cfg.Subscribers), + expectPerSub: make([]int, cfg.Subscribers), + } + + // Seed both PCG streams from cfg.Seed so placement is reproducible + // for a given seed. Node placement is benchmark randomness, not + // security-sensitive. + seed := uint64(cfg.Seed) //nolint:gosec // G115: deterministic benchmark seed, sign irrelevant. + rng := rand.New(rand.NewPCG(seed, seed)) //nolint:gosec // G404: placement randomness, not security-sensitive. + + base := cfg.Messages / cfg.Publishers + remainder := cfg.Messages % cfg.Publishers + perSubjectMsgs := make([]int, cfg.Subjects) + for i := range cfg.Publishers { + msgs := base + if i == 0 { + msgs += remainder + } + pl.perPubMsgs[i] = msgs + pl.pubSubject[i] = i % cfg.Subjects + pl.pubNode[i] = rng.IntN(cfg.Replicas) + perSubjectMsgs[pl.pubSubject[i]] += msgs + } + + for j := range cfg.Subscribers { + pl.subSubject[j] = j % cfg.Subjects + pl.subNode[j] = rng.IntN(cfg.Replicas) + pl.expectPerSub[j] = perSubjectMsgs[pl.subSubject[j]] + pl.totalExpected += pl.expectPerSub[j] + } + + pl.pubNodes = uniqueInts(pl.pubNode) + pl.subNodes = uniqueInts(pl.subNode) + return pl +} + +// uniqueInts returns the sorted distinct values of a slice. +func uniqueInts(values []int) []int { + out := slices.Clone(values) + slices.Sort(out) + return slices.Compact(out) +} + +// subjectName returns the NATS subject for subject index i. +func subjectName(i int) string { + return fmt.Sprintf("bench.%d", i) +} diff --git a/coderd/x/nats/natsbench/plan_internal_test.go b/coderd/x/nats/natsbench/plan_internal_test.go new file mode 100644 index 0000000000..04445f084f --- /dev/null +++ b/coderd/x/nats/natsbench/plan_internal_test.go @@ -0,0 +1,165 @@ +package main + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestBuildPlan(t *testing.T) { + t.Parallel() + + // Node placement is random, so these cases assert only the + // deterministic subject and message assignment. + cases := []struct { + name string + cfg Config + wantPerPubMsgs []int + wantPubSubject []int + wantSubSubject []int + wantExpectPerSub []int + wantTotalExpected int + }{ + { + name: "EvenSplit", + cfg: Config{ + Messages: 100, Publishers: 4, Subjects: 2, Subscribers: 4, Replicas: 2, + }, + wantPerPubMsgs: []int{25, 25, 25, 25}, + wantPubSubject: []int{0, 1, 0, 1}, + wantSubSubject: []int{0, 1, 0, 1}, + wantExpectPerSub: []int{50, 50, 50, 50}, + wantTotalExpected: 200, + }, + { + name: "RemainderToPublisherZero", + cfg: Config{ + Messages: 10, Publishers: 3, Subjects: 3, Subscribers: 3, Replicas: 1, + }, + wantPerPubMsgs: []int{4, 3, 3}, + wantPubSubject: []int{0, 1, 2}, + wantSubSubject: []int{0, 1, 2}, + wantExpectPerSub: []int{4, 3, 3}, + wantTotalExpected: 10, + }, + { + name: "MorePublishersThanSubjects", + cfg: Config{ + Messages: 30, Publishers: 5, Subjects: 2, Subscribers: 2, Replicas: 3, + }, + // Publisher 0 gets 6 + remainder 0; 30/5 = 6 each. + wantPerPubMsgs: []int{6, 6, 6, 6, 6}, + wantPubSubject: []int{0, 1, 0, 1, 0}, + wantSubSubject: []int{0, 1}, + // Subject 0 receives from publishers 0, 2, 4 (18 msgs); + // subject 1 from publishers 1, 3 (12 msgs). + wantExpectPerSub: []int{18, 12}, + wantTotalExpected: 30, + }, + { + name: "SubscriberOnSubjectWithoutPublishers", + cfg: Config{ + Messages: 9, Publishers: 1, Subjects: 2, Subscribers: 4, Replicas: 1, + }, + wantPerPubMsgs: []int{9}, + wantPubSubject: []int{0}, + wantSubSubject: []int{0, 1, 0, 1}, + // Subscribers on subject 1 expect nothing. + wantExpectPerSub: []int{9, 0, 9, 0}, + wantTotalExpected: 18, + }, + { + name: "FanOutExceedsPublishes", + cfg: Config{ + Messages: 100, Publishers: 2, Subjects: 1, Subscribers: 10, Replicas: 5, + }, + wantPerPubMsgs: []int{50, 50}, + wantPubSubject: []int{0, 0}, + wantSubSubject: []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, + wantExpectPerSub: []int{100, 100, 100, 100, 100, 100, 100, 100, 100, 100}, + wantTotalExpected: 1000, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + pl := buildPlan(tc.cfg) + require.Equal(t, tc.wantPerPubMsgs, pl.perPubMsgs) + require.Equal(t, tc.wantPubSubject, pl.pubSubject) + require.Equal(t, tc.wantSubSubject, pl.subSubject) + require.Equal(t, tc.wantExpectPerSub, pl.expectPerSub) + require.Equal(t, tc.wantTotalExpected, pl.totalExpected) + + // Every node index is a valid replica, and pubNodes/subNodes + // are the sorted distinct sets of those indexes. + for _, n := range append(append([]int{}, pl.pubNode...), pl.subNode...) { + require.GreaterOrEqual(t, n, 0) + require.Less(t, n, tc.cfg.Replicas) + } + require.Equal(t, uniqueInts(pl.pubNode), pl.pubNodes) + require.Equal(t, uniqueInts(pl.subNode), pl.subNodes) + }) + } +} + +func TestBuildPlanNodePlacement(t *testing.T) { + t.Parallel() + + cfg := Config{Messages: 1000, Publishers: 50, Subjects: 10, Subscribers: 200, Replicas: 7} + + // The same seed reproduces the same placement. + require.Equal(t, buildPlan(cfg).pubNode, buildPlan(cfg).pubNode) + require.Equal(t, buildPlan(cfg).subNode, buildPlan(cfg).subNode) + + // A different seed (very likely) produces a different placement. + other := cfg + other.Seed = cfg.Seed + 1 + require.NotEqual(t, buildPlan(cfg).subNode, buildPlan(other).subNode) + + // With many clients and few replicas, placement spreads across + // every node rather than collapsing onto one. + pl := buildPlan(cfg) + require.Len(t, pl.pubNodes, cfg.Replicas) + require.Len(t, pl.subNodes, cfg.Replicas) +} + +// TestDefaultSeedSpreadsEvenly guards the DefaultSeed comment's claim: +// with the matrix shape (10 publishers) it places publishers perfectly +// evenly across nodes at every matrix replica count (1, 5, and 10), so +// no node is left publisher-less and cross-node routing is not skewed. +func TestDefaultSeedSpreadsEvenly(t *testing.T) { + t.Parallel() + + const publishers = 10 + for _, replicas := range []int{1, 5, 10} { + cfg := Config{ + Messages: 1000, + Subjects: 10, + Publishers: publishers, + // Subscribers do not affect publisher placement, but a plan + // needs at least one. + Subscribers: 1, + Replicas: replicas, + Seed: DefaultSeed, + } + pl := buildPlan(cfg) + + perNode := make([]int, replicas) + for _, n := range pl.pubNode { + perNode[n]++ + } + want := publishers / replicas + for node, got := range perNode { + require.Equalf(t, want, got, + "replicas=%d node=%d expected %d publishers, got %d (placement %v)", + replicas, node, want, got, perNode) + } + } +} + +func TestSubjectName(t *testing.T) { + t.Parallel() + require.Equal(t, "bench.0", subjectName(0)) + require.Equal(t, "bench.17", subjectName(17)) +} diff --git a/coderd/x/nats/natsbench/readiness.go b/coderd/x/nats/natsbench/readiness.go new file mode 100644 index 0000000000..f87b1d1b8b --- /dev/null +++ b/coderd/x/nats/natsbench/readiness.go @@ -0,0 +1,199 @@ +package main + +import ( + "bytes" + "context" + "fmt" + "slices" + "strconv" + "strings" + "sync" + "time" + + "golang.org/x/xerrors" +) + +const ( + // probePrefix tags readiness probes so they are distinguishable + // from benchmark payloads. Benchmark payloads are all zeros, so any + // payload starting with this non-zero ASCII prefix can only be a + // probe. + probePrefix = "natsbench-probe:" + // probeInterval is how often probes are re-published while waiting + // for cross-route subscription interest to propagate. + probeInterval = 25 * time.Millisecond +) + +// probePrefixBytes avoids per-call allocation in probeNode, which runs +// on every delivered message. +var probePrefixBytes = []byte(probePrefix) + +// probePayload encodes a readiness probe identifying the publishing +// node: the probe prefix followed by the node index in decimal ASCII. +func probePayload(node int) []byte { + return []byte(probePrefix + strconv.Itoa(node)) +} + +// probeNode decodes a readiness probe. It reports false for benchmark +// payloads, which are all zeros and never match the probe prefix. +func probeNode(payload []byte) (int, bool) { + rest, ok := bytes.CutPrefix(payload, probePrefixBytes) + if !ok { + return 0, false + } + node, err := strconv.Atoi(string(rest)) + if err != nil { + return 0, false + } + return node, true +} + +// probeTracker records which publisher nodes a subscriber has observed +// probes from. +type probeTracker struct { + mu sync.Mutex + seen map[int]struct{} +} + +func newProbeTracker() *probeTracker { + return &probeTracker{seen: make(map[int]struct{})} +} + +func (t *probeTracker) observe(node int) { + t.mu.Lock() + defer t.mu.Unlock() + t.seen[node] = struct{}{} +} + +// missing returns the required node indexes not yet observed, sorted. +func (t *probeTracker) missing(required map[int]struct{}) []int { + t.mu.Lock() + defer t.mu.Unlock() + var out []int + for node := range required { + if _, ok := t.seen[node]; !ok { + out = append(out, node) + } + } + slices.Sort(out) + return out +} + +// subjectNodes maps each subject index to the set of publisher node +// indexes that publish to it. This single mapping drives the whole +// gate: it is both the probe schedule (each node probes its subjects) +// and, looked up by a subscriber's subject, that subscriber's required +// probe set. Subjects without publishers are absent (nothing required). +func subjectNodes(pl plan) map[int]map[int]struct{} { + out := make(map[int]map[int]struct{}) + for i, node := range pl.pubNode { + subject := pl.pubSubject[i] + if out[subject] == nil { + out[subject] = make(map[int]struct{}) + } + out[subject][node] = struct{}{} + } + return out +} + +// readinessCheckInterval is how often the gate polls for convergence. +// It is finer than probeInterval so the measured convergence time +// reflects when probes actually propagated rather than the probe +// republish cadence. +const readinessCheckInterval = time.Millisecond + +// awaitTopologyReady proves that subscription interest has propagated to +// every publisher node before the measured phase, and returns how long +// that took. Each publisher node repeatedly publishes an in-band probe +// on every subject it will publish to; the gate converges when every +// subscriber has observed a probe from every publisher node targeting +// its subject. Without this, routed deliveries silently undercount on +// fresh clusters. +// +// The returned duration is the cluster convergence time: from the first +// probe to the moment interest has propagated everywhere. It is +// measured from gate entry, which is immediately after all subscriptions +// are registered. +func awaitTopologyReady(ctx context.Context, top *topology, pl plan, timeout time.Duration, trackers []*probeTracker) (time.Duration, error) { + bySubject := subjectNodes(pl) + required := make([]map[int]struct{}, len(pl.subSubject)) + for j, subject := range pl.subSubject { + required[j] = bySubject[subject] + } + + publishProbes := func() error { + for subject, nodes := range bySubject { + for node := range nodes { + if err := top.nodes[node].Publish(subjectName(subject), probePayload(node)); err != nil { + return xerrors.Errorf("publish probe from node %d on %s: %w", node, subjectName(subject), err) + } + } + } + for _, node := range pl.pubNodes { + if err := top.nodes[node].Flush(); err != nil { + return xerrors.Errorf("flush probes from node %d: %w", node, err) + } + } + return nil + } + + started := time.Now() + deadline := time.NewTimer(timeout) + defer deadline.Stop() + publishTicker := time.NewTicker(probeInterval) + defer publishTicker.Stop() + checkTicker := time.NewTicker(readinessCheckInterval) + defer checkTicker.Stop() + + if err := publishProbes(); err != nil { + return 0, err + } + if isReady(trackers, required) { + return time.Since(started), nil + } + + for { + select { + case <-ctx.Done(): + return 0, xerrors.Errorf("readiness gate canceled: %w", ctx.Err()) + case <-deadline.C: + return 0, xerrors.Errorf("readiness gate timed out after %s: %s", timeout, unreadySubscribers(trackers, required)) + case <-publishTicker.C: + if err := publishProbes(); err != nil { + return 0, err + } + case <-checkTicker.C: + if isReady(trackers, required) { + return time.Since(started), nil + } + } + } +} + +func isReady(trackers []*probeTracker, required []map[int]struct{}) bool { + for j, tracker := range trackers { + if len(tracker.missing(required[j])) > 0 { + return false + } + } + return true +} + +// unreadySubscribers describes which subscribers are still missing +// probes from which publisher nodes. +func unreadySubscribers(trackers []*probeTracker, required []map[int]struct{}) string { + const maxEntries = 20 + var entries []string + for j, tracker := range trackers { + missing := tracker.missing(required[j]) + if len(missing) == 0 { + continue + } + if len(entries) >= maxEntries { + entries = append(entries, "...") + break + } + entries = append(entries, fmt.Sprintf("subscriber %d missing probes from nodes %v", j, missing)) + } + return strings.Join(entries, "; ") +} diff --git a/coderd/x/nats/natsbench/readiness_internal_test.go b/coderd/x/nats/natsbench/readiness_internal_test.go new file mode 100644 index 0000000000..bb49e9a3ae --- /dev/null +++ b/coderd/x/nats/natsbench/readiness_internal_test.go @@ -0,0 +1,82 @@ +package main + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestProbeRoundTrip(t *testing.T) { + t.Parallel() + + for _, node := range []int{0, 1, 7, 1 << 30} { + got, ok := probeNode(probePayload(node)) + require.True(t, ok) + require.Equal(t, node, got) + } +} + +func TestProbeNodeRejectsBenchmarkPayloads(t *testing.T) { + t.Parallel() + + // Benchmark payloads are all zeros, at any size. + for _, size := range []int{1, 2, 9, Payload8KB} { + _, ok := probeNode(make([]byte, size)) + require.False(t, ok) + } + // The bare prefix has no node index. + _, ok := probeNode([]byte(probePrefix)) + require.False(t, ok) + // A trailing non-digit byte fails decoding. + _, ok = probeNode(append(probePayload(3), 0)) + require.False(t, ok) +} + +func TestProbeTracker(t *testing.T) { + t.Parallel() + + tracker := newProbeTracker() + required := map[int]struct{}{0: {}, 2: {}} + + require.Equal(t, []int{0, 2}, tracker.missing(required)) + tracker.observe(0) + tracker.observe(1) // Not required; ignored by missing. + require.Equal(t, []int{2}, tracker.missing(required)) + tracker.observe(2) + require.Empty(t, tracker.missing(required)) + require.Empty(t, tracker.missing(nil)) +} + +func TestSubjectNodes(t *testing.T) { + t.Parallel() + + pl := buildPlan(Config{ + Messages: 30, Publishers: 5, Subjects: 2, Subscribers: 3, Replicas: 3, + }) + + // subjectNodes must map each subject to exactly the set of nodes + // hosting its publishers, derived independently from the plan. + want := make(map[int]map[int]struct{}) + for i, subject := range pl.pubSubject { + if want[subject] == nil { + want[subject] = make(map[int]struct{}) + } + want[subject][pl.pubNode[i]] = struct{}{} + } + require.Equal(t, want, subjectNodes(pl)) +} + +func TestReadinessConverged(t *testing.T) { + t.Parallel() + + trackers := []*probeTracker{newProbeTracker(), newProbeTracker()} + required := []map[int]struct{}{{0: {}}, {0: {}, 1: {}}} + + require.False(t, isReady(trackers, required)) + trackers[0].observe(0) + trackers[1].observe(0) + require.False(t, isReady(trackers, required)) + require.Contains(t, unreadySubscribers(trackers, required), "subscriber 1") + trackers[1].observe(1) + require.True(t, isReady(trackers, required)) +} diff --git a/coderd/x/nats/natsbench/report.go b/coderd/x/nats/natsbench/report.go new file mode 100644 index 0000000000..50135240b0 --- /dev/null +++ b/coderd/x/nats/natsbench/report.go @@ -0,0 +1,234 @@ +package main + +import ( + "fmt" + "io" + "math" + "strconv" + "strings" + "time" + + "github.com/dustin/go-humanize" +) + +// ScenarioResult pairs a scenario with its run outcome. +type ScenarioResult struct { + Scenario Scenario + Result *Result + Err error +} + +// valid reports whether the run produced trustworthy numbers: it +// completed without error. Dropped messages do NOT invalidate a run; +// they are reported in the Drops column alongside the throughput the +// run did achieve. +func (r ScenarioResult) valid() bool { + return r.Err == nil && r.Result != nil +} + +// RenderMarkdown writes grouped markdown tables, one per payload size +// in first-seen order. Failed runs (a non-nil error) render INVALID +// instead of throughput numbers; a Status column appears only for +// groups that contain a failed run, so clean matrices stay compact. +// Dropped messages are a normal metric in the Drops column, not a +// failure. +func RenderMarkdown(w io.Writer, results []ScenarioResult) error { + var b strings.Builder + for gi, group := range groupByPayload(results) { + if gi > 0 { + _, _ = b.WriteString("\n") + } + _, _ = fmt.Fprintf(&b, "### Payload %s\n\n", formatPayload(group.payload)) + renderGroup(&b, group.rows) + } + _, err := io.WriteString(w, b.String()) + return err +} + +func renderGroup(b *strings.Builder, rows []ScenarioResult) { + withStatus := false + for _, row := range rows { + if !row.valid() { + withStatus = true + break + } + } + + // The Status column is left-aligned (free text); the rest are + // numeric and right-aligned. + headers := []string{"Replicas", "Subjects", "Publishers", "Subscribers", "Messages", "Converge", "Pubs/sec", "Deliveries/sec", "Drops"} + aligns := []alignment{alignRight, alignRight, alignRight, alignRight, alignRight, alignRight, alignRight, alignRight, alignRight} + if withStatus { + headers = append(headers, "Status") + aligns = append(aligns, alignLeft) + } + + table := [][]string{headers} + for _, row := range rows { + cells, status := rowCells(row) + if withStatus { + cells = append(cells, status) + } + table = append(table, cells) + } + writeAlignedTable(b, table, aligns) +} + +type alignment int + +const ( + alignLeft alignment = iota + alignRight +) + +// writeAlignedTable writes a GitHub-flavored markdown table whose raw +// text also lines up in a fixed-width terminal: every cell is padded to +// its column's widest value. table[0] is the header row. +func writeAlignedTable(b *strings.Builder, table [][]string, aligns []alignment) { + widths := make([]int, len(table[0])) + for _, rowCells := range table { + for col, cell := range rowCells { + widths[col] = max(widths[col], len(cell)) + } + } + + writeRow := func(cells []string) { + _, _ = b.WriteString("|") + for col, cell := range cells { + _, _ = fmt.Fprintf(b, " %s |", pad(cell, widths[col], aligns[col])) + } + _, _ = b.WriteString("\n") + } + + writeRow(table[0]) + + // Separator row: plain dashes sized to each column. + _, _ = b.WriteString("|") + for _, width := range widths { + _, _ = fmt.Fprintf(b, " %s |", strings.Repeat("-", width)) + } + _, _ = b.WriteString("\n") + + for _, cells := range table[1:] { + writeRow(cells) + } +} + +// pad widens s to width, on the left for right-aligned columns and on +// the right otherwise. +func pad(s string, width int, align alignment) string { + gap := width - len(s) + if gap <= 0 { + return s + } + if align == alignRight { + return strings.Repeat(" ", gap) + s + } + return s + strings.Repeat(" ", gap) +} + +type payloadGroup struct { + payload int + rows []ScenarioResult +} + +func groupByPayload(results []ScenarioResult) []payloadGroup { + var groups []payloadGroup + index := make(map[int]int) + for _, result := range results { + size := result.Scenario.Config.PayloadSize + gi, ok := index[size] + if !ok { + gi = len(groups) + index[size] = gi + groups = append(groups, payloadGroup{payload: size}) + } + groups[gi].rows = append(groups[gi].rows, result) + } + return groups +} + +// rowCells returns the shape and measured cells for one result, plus a +// status string. The cells are: replicas, subjects, publishers, +// subscribers, messages, cluster convergence time, pubs/sec, +// deliveries/sec, drops. Failed runs render INVALID rates and a status +// describing why; valid runs render rates and an "ok" status that +// callers may drop for all-clean groups. A valid run that dropped +// messages still renders its rates, with the loss in the Drops column. +func rowCells(row ScenarioResult) (cells []string, status string) { + cfg := row.Scenario.Config + if row.Result != nil { + cfg = row.Result.Config + } + cells = []string{ + strconv.Itoa(cfg.Replicas), + strconv.Itoa(cfg.Subjects), + strconv.Itoa(cfg.Publishers), + strconv.Itoa(cfg.Subscribers), + humanize.Comma(int64(cfg.Messages)), + formatConvergence(row), + } + + if row.valid() { + return append(cells, + formatRate(row.Result.PubsPerSec), + formatRate(row.Result.DeliveriesPerSec), + formatDrops(row.Result), + ), "ok" + } + return append(cells, "INVALID", "INVALID", formatDrops(row.Result)), shortReason(row) +} + +// formatDrops renders the drop count and its percentage of expected +// deliveries, or "-" when there is no result to measure. +func formatDrops(res *Result) string { + if res == nil { + return "-" + } + if res.Drops == 0 { + return "0" + } + pct := 0.0 + if res.Expected > 0 { + pct = 100 * float64(res.Drops) / float64(res.Expected) + } + return fmt.Sprintf("%s (%.2f%%)", humanize.Comma(res.Drops), pct) +} + +// shortReason summarizes why a run failed in one table-safe line. +func shortReason(row ScenarioResult) string { + if row.Err == nil { + return "no result" + } + msg := row.Err.Error() + if i := strings.IndexByte(msg, '\n'); i >= 0 { + msg = msg[:i] + } + const maxLen = 80 + if len(msg) > maxLen { + msg = msg[:maxLen] + "..." + } + return strings.ReplaceAll(msg, "|", "\\|") +} + +// formatRate renders a rate as a comma-separated integer. +func formatRate(rate float64) string { + return humanize.Comma(int64(math.Round(rate))) +} + +// formatConvergence renders the cluster convergence time, or "-" for +// single-node runs and runs without a result, which have no gate. +func formatConvergence(row ScenarioResult) string { + if row.Result == nil || row.Result.Config.Replicas <= 1 { + return "-" + } + return row.Result.ConvergenceDuration.Round(100 * time.Microsecond).String() +} + +// formatPayload renders a payload size as KiB when it divides evenly. +func formatPayload(size int) string { + if size >= 1024 && size%1024 == 0 { + return fmt.Sprintf("%d KiB", size/1024) + } + return fmt.Sprintf("%d B", size) +} diff --git a/coderd/x/nats/natsbench/report_internal_test.go b/coderd/x/nats/natsbench/report_internal_test.go new file mode 100644 index 0000000000..570fa17557 --- /dev/null +++ b/coderd/x/nats/natsbench/report_internal_test.go @@ -0,0 +1,168 @@ +package main + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + "golang.org/x/xerrors" + + "github.com/coder/coder/v2/testutil" +) + +func TestFormatPayload(t *testing.T) { + t.Parallel() + + require.Equal(t, "8 KiB", formatPayload(Payload8KB)) + require.Equal(t, "64 KiB", formatPayload(Payload64KB)) + require.Equal(t, "100 B", formatPayload(100)) +} + +func TestRenderMarkdown(t *testing.T) { + t.Parallel() + + valid := ScenarioResult{ + Scenario: Scenario{ + Name: "8KiB-1r", + Config: Config{Messages: 100000, PayloadSize: Payload8KB, Replicas: 1}, + }, + Result: &Result{ + Config: Config{Messages: 100000, PayloadSize: Payload8KB, Replicas: 1}, + Published: 100000, + Delivered: 500000, + PublishDuration: time.Second, + DeliverDuration: 2 * time.Second, + PubsPerSec: 100000, + DeliveriesPerSec: 250000, + }, + } + // A run that dropped messages is still valid: it renders real rates + // plus the loss in the Drops column. + dropped := ScenarioResult{ + Scenario: Scenario{ + Name: "8KiB-5r", + Config: Config{Messages: 100000, PayloadSize: Payload8KB, Replicas: 5}, + }, + Result: &Result{ + Config: Config{Messages: 100000, PayloadSize: Payload8KB, Replicas: 5}, + Expected: 100000, + Published: 100000, + Delivered: 97500, + Drops: 2500, + PublishDuration: time.Second, + DeliverDuration: time.Second, + PubsPerSec: 80000, + DeliveriesPerSec: 195000, + }, + } + failed := ScenarioResult{ + Scenario: Scenario{ + Name: "64KiB-10r", + Config: Config{Messages: 20000, PayloadSize: Payload64KB, Replicas: 10}, + }, + Err: xerrors.New("readiness gate: timed out\nsecond line is omitted"), + } + + var b strings.Builder + require.NoError(t, RenderMarkdown(&b, []ScenarioResult{valid, dropped, failed})) + out := b.String() + + require.Contains(t, out, "### Payload 8 KiB") + require.Contains(t, out, "### Payload 64 KiB") + require.Contains(t, out, "Drops") + require.Contains(t, out, "100,000") + require.Contains(t, out, "250,000") + // The dropped run renders its throughput and its loss percentage, + // not INVALID. + require.Contains(t, out, "195,000") + require.Contains(t, out, "2,500 (2.50%)") + // Only the failed run (a non-nil error) renders INVALID and a Status + // column. + require.Contains(t, out, "Status") + require.Contains(t, out, "INVALID") + require.Contains(t, out, "readiness gate: timed out") + require.NotContains(t, out, "second line is omitted") + + // Every body row has the same width as the header, so columns line + // up in a terminal. + assertAlignedTable(t, out) +} + +// assertAlignedTable checks that all table rows in a rendered report +// (lines starting with "|") within each contiguous block share the same +// rune width. +func assertAlignedTable(t *testing.T, out string) { + t.Helper() + width := -1 + for _, line := range strings.Split(out, "\n") { + if !strings.HasPrefix(line, "|") { + width = -1 + continue + } + if width < 0 { + width = len([]rune(line)) + continue + } + require.Equal(t, width, len([]rune(line)), "row width mismatch:\n%s", line) + } +} + +func TestRenderMarkdownCleanGroupOmitsStatus(t *testing.T) { + t.Parallel() + + result := func(replicas int, pubs, dels float64, converge time.Duration) ScenarioResult { + cfg := Config{ + Messages: 100000, PayloadSize: Payload8KB, Replicas: replicas, + Subjects: 10, Publishers: 10, Subscribers: 50, + } + return ScenarioResult{ + Scenario: Scenario{Config: cfg}, + Result: &Result{ + Config: cfg, PubsPerSec: pubs, DeliveriesPerSec: dels, + ConvergenceDuration: converge, + }, + } + } + + var b strings.Builder + require.NoError(t, RenderMarkdown(&b, []ScenarioResult{ + result(1, 100000, 250000, 0), + result(5, 90000, 220000, 25*time.Millisecond), + })) + out := b.String() + + // The shape and measured columns are reported alongside throughput. + for _, header := range []string{"Replicas", "Subjects", "Publishers", "Subscribers", "Messages", "Converge", "Pubs/sec", "Deliveries/sec"} { + require.Contains(t, out, header) + } + // A clean group omits the conditional Status column. + require.NotContains(t, out, "Status") + require.Contains(t, out, "250,000") + require.Contains(t, out, "220,000") + // Single-replica rows have no gate; multi-replica rows show the + // convergence time. + require.Contains(t, out, "25ms") + assertAlignedTable(t, out) +} + +func TestDefaultScenarios(t *testing.T) { + t.Parallel() + + scenarios := DefaultScenarios() + require.Len(t, scenarios, 6) + seen := make(map[string]struct{}) + for _, sc := range scenarios { + seen[sc.Name] = struct{}{} + cfg := sc.Config + cfg.Timeout = testutil.WaitShort + require.NoError(t, cfg.validate()) + if sc.Config.PayloadSize == Payload64KB && sc.Config.Replicas > 1 { + require.Less(t, sc.Config.Messages, DefaultMessages, + "64KiB cluster runs must reduce the message count") + } else { + require.Equal(t, DefaultMessages, sc.Config.Messages) + } + } + require.Len(t, seen, 6, "scenario names must be unique") +} diff --git a/coderd/x/nats/natsbench/scenario.go b/coderd/x/nats/natsbench/scenario.go new file mode 100644 index 0000000000..bb2902ef9b --- /dev/null +++ b/coderd/x/nats/natsbench/scenario.go @@ -0,0 +1,76 @@ +package main + +import "fmt" + +// Scenario names one benchmark configuration in the standard matrix. +type Scenario struct { + Name string + Config Config +} + +// DefaultConns is the default size of the publisher and subscriber +// connection pools in the standard matrix. Production ships a single +// connection each, but the benchmarks use three to match the prior +// natsbench harness and to exercise cross-subject parallelism. +const DefaultConns = 3 + +// DefaultSeed seeds the pseudorandom node placement. With the matrix +// shape (10 publishers) and 10 replicas, most seeds leave several nodes +// publisher-less (only ~1 in 2750 seeds covers every node), which would +// skew cross-node routing measurements. 2068 was selected because it +// spreads publishers perfectly evenly across nodes at every matrix +// replica count (1, 5, and 10): one publisher per node at 10 replicas, +// two per node at 5, as TestDefaultSeedSpreadsEvenly verifies. Override +// with -seed to sample a different, more uneven (and arguably more +// realistic) placement. +const DefaultSeed int64 = 2068 + +// DefaultScenarios returns the standard matrix: payloads of 8 KiB and +// 64 KiB at 1, 5, and 10 replicas. The 64 KiB cluster runs use a +// reduced total message count because the fan-out byte volume is +// memory-heavy. +// Random node placement (see buildPlan) makes the multi-replica runs +// exercise cross-node routing. +// +// The returned configs leave Timeout (and Seed) unset; callers set them +// before passing the configs to Run (the CLI does this from its +// -timeout and -seed flags). +func DefaultScenarios() []Scenario { + const ( + subjects = 10 + publishers = 10 + subscribers = 50 + reducedMessages = 20000 + ) + payloads := []struct { + label string + size int + }{ + {"8KiB", Payload8KB}, + {"64KiB", Payload64KB}, + } + + var scenarios []Scenario + for _, payload := range payloads { + for _, replicas := range []int{1, 5, 10} { + messages := DefaultMessages + if payload.size == Payload64KB && replicas > 1 { + messages = reducedMessages + } + scenarios = append(scenarios, Scenario{ + Name: fmt.Sprintf("%s-%dr", payload.label, replicas), + Config: Config{ + Messages: messages, + PayloadSize: payload.size, + Subjects: subjects, + Publishers: publishers, + Subscribers: subscribers, + Replicas: replicas, + PublishConns: DefaultConns, + SubscribeConns: DefaultConns, + }, + }) + } + } + return scenarios +} diff --git a/coderd/x/nats/natsbench/topology.go b/coderd/x/nats/natsbench/topology.go new file mode 100644 index 0000000000..f44744743c --- /dev/null +++ b/coderd/x/nats/natsbench/topology.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "fmt" + "slices" + "sync" + "time" + + natsserver "github.com/nats-io/nats-server/v2/server" + "golang.org/x/xerrors" + + "cdr.dev/slog/v3" + "github.com/coder/coder/v2/coderd/x/nats" +) + +// topology owns the embedded pubsub nodes for one benchmark run. +type topology struct { + nodes []*nats.Pubsub +} + +// closeAll shuts down every node. Pubsub.Close is idempotent and +// always returns nil. +func (t *topology) closeAll() { + for _, node := range t.nodes { + _ = node.Close() + } +} + +// staticPeerFetcher is a mutable nats.PeerFetcher seeded after every +// node in the cluster has started. +type staticPeerFetcher struct { + mu sync.Mutex + addrs []string +} + +var _ nats.PeerFetcher = (*staticPeerFetcher)(nil) + +func (f *staticPeerFetcher) PrimaryPeerAddresses() []string { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.addrs) +} + +func (f *staticPeerFetcher) set(addrs []string) { + f.mu.Lock() + defer f.mu.Unlock() + f.addrs = slices.Clone(addrs) +} + +// buildTopology starts cfg.Replicas embedded pubsub nodes. For +// multi-replica runs it wires a full mesh: every node learns the other +// nodes' route addresses through its peer fetcher and refreshes its +// routes. Route convergence is not assumed here; the readiness gate +// proves it before the measured phase. +func buildTopology(ctx context.Context, logger slog.Logger, cfg Config) (*topology, error) { + // The route auth token is unique per run so concurrent benchmark + // processes on one host can never mesh with each other. + token := fmt.Sprintf("natsbench-%d", time.Now().UnixNano()) + + // One fetcher shared by every node: each node filters its own + // address out of the route list, so they can all be handed the + // full set of addresses. + fetcher := &staticPeerFetcher{} + top := &topology{nodes: make([]*nats.Pubsub, 0, cfg.Replicas)} + for i := range cfg.Replicas { + opts := pubsubOptions(cfg) + opts.ClusterAuthToken = token + opts.PeerFetcher = fetcher + node, err := nats.New(ctx, logger.Named(fmt.Sprintf("node%d", i)), opts) + if err != nil { + top.closeAll() + return nil, xerrors.Errorf("create node %d: %w", i, err) + } + top.nodes = append(top.nodes, node) + } + + if cfg.Replicas > 1 { + addrs := make([]string, len(top.nodes)) + for i, node := range top.nodes { + addr, err := routeAddress(node) + if err != nil { + top.closeAll() + return nil, xerrors.Errorf("node %d route address: %w", i, err) + } + addrs[i] = addr + } + fetcher.set(addrs) + for _, node := range top.nodes { + node.RefreshPeers() + } + } + return top, nil +} + +// pubsubOptions maps the benchmark config onto nats.Options. The +// cluster route listener always uses a random port: a zero ClusterPort +// means the production default 6222, which both collides across nodes +// and triggers peer-address port rewriting. +func pubsubOptions(cfg Config) nats.Options { + opts := nats.Options{ + InProcess: cfg.InProcess, + PublishConns: cfg.PublishConns, + SubscribeConns: cfg.SubscribeConns, + ClusterHost: "127.0.0.1", + ClusterPort: natsserver.RANDOM_PORT, + } + if cfg.LocalQueueMsgs > 0 { + opts.PendingLimits.Msgs = cfg.LocalQueueMsgs + } + if cfg.LocalQueueBytes > 0 { + opts.PendingLimits.Bytes = cfg.LocalQueueBytes + } + if cfg.MaxPending > 0 { + opts.MaxPending = cfg.MaxPending + } + return opts +} + +// routeAddress returns the node's cluster route address as a NATS URL. +func routeAddress(node *nats.Pubsub) (string, error) { + addr := node.Server.ClusterAddr() + if addr == nil { + return "", xerrors.New("server has no cluster listener") + } + return "nats://" + addr.String(), nil +} diff --git a/coderd/x/nats/natsbench/workload.go b/coderd/x/nats/natsbench/workload.go new file mode 100644 index 0000000000..d0af735af3 --- /dev/null +++ b/coderd/x/nats/natsbench/workload.go @@ -0,0 +1,383 @@ +package main + +import ( + "context" + "errors" + "fmt" + "runtime" + "strings" + "sync" + "sync/atomic" + "time" + + natsserver "github.com/nats-io/nats-server/v2/server" + "golang.org/x/xerrors" + + "cdr.dev/slog/v3" + "github.com/coder/coder/v2/coderd/database/pubsub" +) + +// subscriberState tracks one subscriber's exact delivery accounting. +type subscriberState struct { + expect int + delivered atomic.Int64 + tracker *probeTracker +} + +// workload wires subscribers and publishers onto a topology and runs +// the measured phase with exact accounting and bounded waits. +type workload struct { + logger slog.Logger + top *topology + pl plan + cfg Config + + subs []*subscriberState + cancels []func() + // dropSignals counts ErrDroppedMessages deliveries. It is a + // diagnostic signal only: the authoritative loss count is Expected + // minus Delivered, because these signals coalesce and cross-node + // routed loss never signals at all. + dropSignals atomic.Int64 + + published atomic.Int64 + // outstanding counts subscribers that have not yet reached their + // expected delivery count; allDone closes when it hits zero. + outstanding atomic.Int64 + allDone chan struct{} + // convergence is how long the readiness gate took to propagate + // interest across the cluster; zero for single-node runs. + convergence time.Duration + // settleWindow is the delivery quiescence window used when a run + // dropped messages and can never reach its exact count. It defaults + // to defaultSettleWindow; tests override it to stay fast. + settleWindow time.Duration +} + +// runWorkload executes one benchmark run on an already-built topology: +// subscribe everywhere, prove cluster readiness, release all publishers +// together, and account for every expected delivery. +func runWorkload(ctx context.Context, logger slog.Logger, top *topology, pl plan, cfg Config) (*Result, error) { + w := &workload{ + logger: logger, + top: top, + pl: pl, + cfg: cfg, + allDone: make(chan struct{}), + settleWindow: defaultSettleWindow, + } + defer w.cancelAll() + if err := w.subscribe(); err != nil { + return nil, err + } + + if len(top.nodes) > 1 { + trackers := make([]*probeTracker, len(w.subs)) + for j, st := range w.subs { + trackers[j] = st.tracker + } + convergence, err := awaitTopologyReady(ctx, top, pl, cfg.Timeout, trackers) + if err != nil { + return nil, xerrors.Errorf("readiness gate: %w", err) + } + w.convergence = convergence + logger.Debug(ctx, "cluster converged", slog.F("duration", convergence)) + } + + // Both durations are measured from this instant. Goroutine spawn + // cost is negligible against the publish phase, so the publishers + // start as they are launched rather than behind a barrier. + hot := time.Now() + pubDone, pubErrCh := w.startPublishers() + + if err := w.awaitPhase(ctx, "publish", pubDone); err != nil { + return w.buildResult(time.Since(hot), time.Since(hot)), err + } + // pubDone closing implies pubErrCh is already closed (both happen + // after the publishers' WaitGroup), so this drain terminates. + var pubErrs []error + for err := range pubErrCh { + pubErrs = append(pubErrs, err) + } + if err := errors.Join(pubErrs...); err != nil { + return w.buildResult(time.Since(hot), time.Since(hot)), xerrors.Errorf("publish: %w", err) + } + for _, idx := range pl.pubNodes { + if err := top.nodes[idx].Flush(); err != nil { + return w.buildResult(time.Since(hot), time.Since(hot)), xerrors.Errorf("flush publisher node %d: %w", idx, err) + } + } + publishDur := time.Since(hot) + + deliverDur, err := w.awaitDelivery(ctx, hot) + if err != nil { + return w.buildResult(publishDur, deliverDur), err + } + + res := w.buildResult(publishDur, deliverDur) + if res.Drops > 0 { + logger.Warn(ctx, "run dropped messages", + slog.F("expected", res.Expected), + slog.F("delivered", res.Delivered), + slog.F("drops", res.Drops), + slog.F("drop_signals", w.dropSignals.Load()), + ) + } + return res, nil +} + +// subscribe registers every subscriber. SubscribeWithErr flushes the +// SUB on its subscribe connection before returning, so once this +// returns every subscriber's interest is registered with its local +// server; cross-node interest propagation is proven separately by the +// readiness gate. +func (w *workload) subscribe() error { + for j := range w.pl.subSubject { + st := &subscriberState{ + expect: w.pl.expectPerSub[j], + tracker: newProbeTracker(), + } + w.subs = append(w.subs, st) + if st.expect > 0 { + w.outstanding.Add(1) + } + node := w.top.nodes[w.pl.subNode[j]] + cancel, err := node.SubscribeWithErr(subjectName(w.pl.subSubject[j]), w.listener(st)) + if err != nil { + return xerrors.Errorf("register subscriber %d: %w", j, err) + } + w.cancels = append(w.cancels, cancel) + } + return nil +} + +// listener builds the delivery callback for one subscriber: probes feed +// the readiness tracker, errors poison the run, and benchmark payloads +// count toward the exact expected total. +func (w *workload) listener(st *subscriberState) pubsub.ListenerWithErr { + return func(ctx context.Context, message []byte, err error) { + if err != nil { + if !xerrors.Is(err, pubsub.ErrDroppedMessages) { + w.logger.Error(ctx, "unexpected subscriber error", slog.Error(err)) + } + w.dropSignals.Add(1) + return + } + if node, ok := probeNode(message); ok { + st.tracker.observe(node) + return + } + // Each subscriber decrements outstanding exactly once, when its + // delivered count first equals its expectation, so the atomic + // reaches zero exactly once and closes allDone exactly once. + if st.delivered.Add(1) == int64(st.expect) { + if w.outstanding.Add(-1) == 0 { + close(w.allDone) + } + } + } +} + +func (w *workload) cancelAll() { + for _, cancel := range w.cancels { + cancel() + } +} + +// startPublishers launches one goroutine per publisher and returns a +// channel closed when every publisher finished and a channel of publish +// errors that is closed at the same time. The error channel is buffered +// to the publisher count so a failing publisher never blocks, and each +// publisher sends at most one error. +func (w *workload) startPublishers() (<-chan struct{}, <-chan error) { + payload := make([]byte, w.cfg.PayloadSize) + errCh := make(chan error, len(w.pl.perPubMsgs)) + var wg sync.WaitGroup + for i := range w.pl.perPubMsgs { + wg.Go(func() { + node := w.top.nodes[w.pl.pubNode[i]] + subject := subjectName(w.pl.pubSubject[i]) + for range w.pl.perPubMsgs[i] { + if err := node.Publish(subject, payload); err != nil { + errCh <- xerrors.Errorf("publisher %d on %s: %w", i, subject, err) + return + } + w.published.Add(1) + } + }) + } + done := make(chan struct{}) + go func() { + wg.Wait() + close(errCh) + close(done) + }() + return done, errCh +} + +// awaitPhase blocks until the phase signal fires, failing on context +// cancellation or the per-phase timeout. Timeouts carry full +// diagnostics so a stuck run is debuggable. Dropped messages do not +// fail a phase: they are accounted as a metric, not a failure. +func (w *workload) awaitPhase(ctx context.Context, phase string, signal <-chan struct{}) error { + timer := time.NewTimer(w.cfg.Timeout) + defer timer.Stop() + select { + case <-signal: + return nil + case <-ctx.Done(): + return xerrors.Errorf("%s phase canceled: %w", phase, ctx.Err()) + case <-timer.C: + return xerrors.Errorf("%s phase timed out after %s:\n%s", phase, w.cfg.Timeout, w.diagnostics()) + } +} + +// deliveryPollInterval is how often awaitDelivery samples the delivery +// counter to detect quiescence. It only bounds the precision of the +// quiescence path; the exact-count fast path fires immediately on +// allDone regardless of this cadence. +const deliveryPollInterval = 100 * time.Millisecond + +// defaultSettleWindow is how long the delivery counter must stay flat +// before a run that dropped messages is declared complete. It is a fixed +// internal constant rather than a knob: only a run that drops messages +// ever waits it out, and on loopback a backlogged subscriber catches up +// in milliseconds, so five seconds is ample headroom. Too short would +// overcount drops; too long only slows runs that already dropped. +const defaultSettleWindow = 5 * time.Second + +// awaitDelivery waits for the deliver phase to finish and returns its +// duration measured from hot. +// +// A zero-drop run reaches its exact expected count, closing allDone, and +// finishes precisely with no settle delay. A run that dropped messages +// can never reach that count, so it instead completes by quiescence: the +// total delivered counter is polled, and once it has not advanced for +// settleWindow the phase is declared complete. The duration is measured +// to the last observed progress, not to the end of the settle window, so +// the idle wait never inflates the delivery rate. +// +// The counter is read by summing the per-subscriber delivery atomics, so +// no global counter or per-delivery timestamp is added to the hot +// delivery path. +func (w *workload) awaitDelivery(ctx context.Context, hot time.Time) (time.Duration, error) { + timer := time.NewTimer(w.cfg.Timeout) + defer timer.Stop() + poll := time.NewTicker(deliveryPollInterval) + defer poll.Stop() + + lastCount := w.totalDelivered() + lastProgress := time.Now() + for { + select { + case <-w.allDone: + return time.Since(hot), nil + case <-ctx.Done(): + return time.Since(hot), xerrors.Errorf("deliver phase canceled: %w", ctx.Err()) + case <-timer.C: + return time.Since(hot), xerrors.Errorf("deliver phase timed out after %s:\n%s", w.cfg.Timeout, w.diagnostics()) + case now := <-poll.C: + count := w.totalDelivered() + if count != lastCount { + lastCount = count + lastProgress = now + continue + } + if now.Sub(lastProgress) >= w.settleWindow { + return lastProgress.Sub(hot), nil + } + } + } +} + +// totalDelivered sums every subscriber's delivered count. It is called +// from the quiescence poller, off the hot delivery path. +func (w *workload) totalDelivered() int64 { + var total int64 + for _, st := range w.subs { + total += st.delivered.Load() + } + return total +} + +// diagnostics renders subscriber shortfalls, per-node server stats, and +// a goroutine dump for timeout errors. +func (w *workload) diagnostics() string { + var b strings.Builder + const maxShortfalls = 20 + short := 0 + for j, st := range w.subs { + got := st.delivered.Load() + if got >= int64(st.expect) { + continue + } + short++ + if short <= maxShortfalls { + _, _ = fmt.Fprintf(&b, "subscriber %d (subject %s, node %d): delivered %d of %d\n", + j, subjectName(w.pl.subSubject[j]), w.pl.subNode[j], got, st.expect) + } + } + if short > maxShortfalls { + _, _ = fmt.Fprintf(&b, "... and %d more subscribers short\n", short-maxShortfalls) + } + _, _ = fmt.Fprintf(&b, "published: %d, drop signals: %d\n", w.published.Load(), w.dropSignals.Load()) + + for i, node := range w.top.nodes { + varz, err := node.Server.Varz(&natsserver.VarzOptions{}) + if err != nil { + _, _ = fmt.Fprintf(&b, "node %d: varz error: %v\n", i, err) + continue + } + _, _ = fmt.Fprintf(&b, "node %d: connections=%d routes=%d subscriptions=%d slow_consumers=%d in_msgs=%d out_msgs=%d\n", + i, varz.Connections, varz.Routes, varz.Subscriptions, varz.SlowConsumers, varz.InMsgs, varz.OutMsgs) + } + + _, _ = b.WriteString("goroutine dump:\n") + _, _ = b.WriteString(goroutineDump()) + return b.String() +} + +// buildResult snapshots counters into a Result. Drops is the exact +// shortfall between expected and observed deliveries, the authoritative +// loss count. Rates are computed for valid and dropping runs alike, +// since a dropping run still has a meaningful throughput for what it +// delivered. +func (w *workload) buildResult(publishDur, deliverDur time.Duration) *Result { + delivered := w.totalDelivered() + expected := int64(w.pl.totalExpected) + res := &Result{ + Config: w.cfg, + Expected: expected, + Published: w.published.Load(), + Delivered: delivered, + Drops: max(0, expected-delivered), + ConvergenceDuration: w.convergence, + PublishDuration: publishDur, + DeliverDuration: deliverDur, + PubsPerSec: ratePerSec(w.published.Load(), publishDur), + DeliveriesPerSec: ratePerSec(delivered, deliverDur), + } + return res +} + +// ratePerSec computes count/dur, returning 0 for non-positive +// durations. +func ratePerSec(count int64, dur time.Duration) float64 { + if dur <= 0 { + return 0 + } + return float64(count) / dur.Seconds() +} + +// goroutineDump captures all goroutine stacks, growing the buffer until +// the dump fits or a hard cap is reached. +func goroutineDump() string { + buf := make([]byte, 1<<20) + for { + n := runtime.Stack(buf, true) + if n < len(buf) || len(buf) >= 16<<20 { + return string(buf[:n]) + } + buf = make([]byte, len(buf)*2) + } +} diff --git a/coderd/x/nats/natsbench/workload_internal_test.go b/coderd/x/nats/natsbench/workload_internal_test.go new file mode 100644 index 0000000000..c80e342392 --- /dev/null +++ b/coderd/x/nats/natsbench/workload_internal_test.go @@ -0,0 +1,85 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/coder/coder/v2/coderd/database/pubsub" + "github.com/coder/coder/v2/testutil" +) + +func TestListenerRecordsDropsAsMetric(t *testing.T) { + t.Parallel() + + w := &workload{ + logger: testutil.Logger(t), + cfg: Config{Timeout: testutil.WaitShort}, + } + st := &subscriberState{expect: 5, tracker: newProbeTracker()} + + // A dropped-message delivery is counted as a signal but does not + // advance delivery progress; drops no longer fail the run. + w.listener(st)(context.Background(), nil, pubsub.ErrDroppedMessages) + require.EqualValues(t, 1, w.dropSignals.Load()) + require.EqualValues(t, 0, st.delivered.Load()) +} + +func TestAwaitDeliveryExactCount(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitShort) + w := &workload{ + logger: testutil.Logger(t), + cfg: Config{Timeout: testutil.WaitShort}, + allDone: make(chan struct{}), + settleWindow: testutil.WaitShort, + } + st := &subscriberState{expect: 3, tracker: newProbeTracker()} + w.subs = []*subscriberState{st} + w.outstanding.Store(1) + + // Deliver exactly the expected count: allDone fires and the deliver + // phase finishes precisely, without waiting out the settle window. + listener := w.listener(st) + go func() { + for range 3 { + listener(ctx, []byte("x"), nil) + } + }() + + dur, err := w.awaitDelivery(ctx, time.Now()) + require.NoError(t, err) + require.Positive(t, dur) + require.EqualValues(t, 3, w.totalDelivered()) +} + +func TestAwaitDeliveryQuiescence(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitShort) + w := &workload{ + logger: testutil.Logger(t), + cfg: Config{Timeout: testutil.WaitShort}, + // A short settle window keeps the test fast; the timeout is far + // larger so quiescence, not the hard timeout, ends the phase. + allDone: make(chan struct{}), + settleWindow: deliveryPollInterval / 2, + } + // A subscriber that never reaches its expectation models a dropped + // message: allDone can never close, so the phase must complete by + // quiescence once the delivery counter stays flat. + w.subs = []*subscriberState{{expect: 100}} + + dur, err := w.awaitDelivery(ctx, time.Now()) + require.NoError(t, err) + require.GreaterOrEqual(t, dur, time.Duration(0)) + + select { + case <-w.allDone: + t.Fatal("allDone closed despite a permanent shortfall") + default: + } +} diff --git a/coderd/x/nats/pubsub.go b/coderd/x/nats/pubsub.go index b1fea0b5b8..d6e20d6327 100644 --- a/coderd/x/nats/pubsub.go +++ b/coderd/x/nats/pubsub.go @@ -17,8 +17,42 @@ import ( "github.com/coder/coder/v2/coderd/database/pubsub" ) -// DefaultMaxPending is the per-client outbound pending byte budget. -const DefaultMaxPending int64 = 128 << 20 +// DefaultServerMaxPendingBytes caps how many bytes the embedded NATS server will +// hold in memory for a single client connection while waiting to write +// them out to that connection. Each message the server needs to deliver +// to a connection is queued in that connection's outbound buffer until +// the socket can accept it; if the consumer reads slower than messages +// arrive, the buffer grows. When it exceeds this cap, NATS declares the +// connection a slow consumer and drops messages rather than buffer +// without bound. +// +// The connection that fills this buffer in practice is the subscribe +// connection: the server writes every message bound for a replica's +// local subscribers out over its subscribe connection pool, which +// defaults to a single connection. In a cluster, cross-node fan-out +// therefore concentrates all of a replica's inbound deliveries on that +// one connection's outbound buffer. Benchmarking high-fanout cluster +// workloads (10 subjects, 10 publishers, 50 subscribers) showed the 128 +// MiB default overflowing and dropping 10-15% of deliveries, while 256 +// MiB and above dropped none; 512 MiB is chosen for headroom. +// +// This is a ceiling, not a reservation: the buffer grows only with +// actual backlog, so a connection that keeps up holds nearly nothing and +// raising the cap costs memory only during the overload bursts it +// absorbs. +const DefaultServerMaxPendingBytes int64 = 512 << 20 + +// DefaultClientMaxPendingBytes is the pending byte limit applied via +// SetPendingLimits to each coalesced *natsgo.Subscription when +// PendingLimits.Bytes is zero. Unlike DefaultServerMaxPendingBytes, +// which bounds the server-side outbound buffer for a whole connection, +// this bounds the nats.go client's per-subscription pending buffer: +// messages the client has received from the server but the +// subscription's async message handler has not yet dispatched. When +// that handler falls behind and the buffer overflows, NATS marks the +// subscription a slow consumer and drops messages, which the package +// surfaces as pubsub.ErrDroppedMessages. +const DefaultClientMaxPendingBytes = 512 * 1024 * 1024 const ( defaultClusterName = "coder" @@ -31,9 +65,9 @@ var errClosed = xerrors.New("nats pubsub closed") // PendingLimits configures per-subscription NATS pending limits set // via SetPendingLimits on each *natsgo.Subscription. type PendingLimits struct { - // Msgs is the per-subscription pending message limit. Positive - // values also set each local listener queue capacity. - // Zero uses the package default. Negative disables this limit. + // Msgs is the per-subscription pending message limit. Zero or + // negative disables the message limit, leaving the byte limit + // (PendingLimits.Bytes) as the only per-subscription bound. Msgs int // Bytes is the per-subscription pending byte limit. @@ -202,7 +236,7 @@ func defaultPendingLimits(in PendingLimits) PendingLimits { out.Msgs = -1 } if out.Bytes == 0 { - out.Bytes = 512 * 1024 * 1024 + out.Bytes = DefaultClientMaxPendingBytes } return out } diff --git a/coderd/x/nats/server.go b/coderd/x/nats/server.go index 47194c8a75..3575cd33b8 100644 --- a/coderd/x/nats/server.go +++ b/coderd/x/nats/server.go @@ -20,7 +20,7 @@ func buildServerOptions(opts Options) (*natsserver.Options, error) { } maxPending := opts.MaxPending if maxPending <= 0 { - maxPending = DefaultMaxPending + maxPending = DefaultServerMaxPendingBytes } sopts := &natsserver.Options{