go_loadgen

README

Go Loadgen

Go Loadgen is a protocol-agnostic, open-loop load generator for Go. Applications provide typed endpoint adapters; the library schedules offered load, distributes it across endpoints, and waits for issued requests to finish.

Semantics

• A phase's RPS is its total offered rate across all targets.
• Scheduling is open-loop: response latency never controls future arrivals.
• Endpoint selection is compiled before a run and uses O(1), lock-free weighted selection.
• Run stops issuing requests at phase boundaries and waits for in-flight requests by default.
• DrainTimeout is optional. When set, outstanding requests are cancelled after that period; arrivals are never blocked.
• MaxInFlight is optional. When full, new arrivals are dropped and reported, preserving open-loop semantics. Loader delays are reported as missed rather than replayed as a catch-up burst.

Scheduling Accuracy And Throughput

RPS is an offered-load target, not a guarantee that every scheduled arrival is issued. The scheduler uses 1 ms batches at rates of 1,000 RPS and above. If the loader is delayed by OS scheduling, Go runtime work, garbage collection, request goroutine creation, client-side serialization, or transport work, it can miss a batch deadline.

Go Loadgen intentionally does not replay overdue batches. Catching up would create a burst above the configured instantaneous rate, retain more work in memory, and hide loader saturation. Instead, the report records those arrivals in Missed; they were never sent. Dropped has a different meaning: an arrival was timely but rejected because MaxInFlight was full.

This is an explicit performance and measurement trade-off. The scheduling hot path avoids queues and blocking, and the report exposes whether the generator kept up. Benchmarks should report Scheduled, Issued, Missed, Dropped, and Completed, rather than treating configured RPS as achieved RPS.

Example

collector, err := go_loadgen.NewGobCollector[Result]("results.gob", time.Second)
if err != nil {
    log.Fatal(err)
}
defer collector.Close()

api, err := go_loadgen.NewEndpoint(client, provider, collector)
if err != nil {
    log.Fatal(err)
}
workload, err := go_loadgen.NewWorkload(go_loadgen.Spec{
    Duration: 60 * time.Second,
    Seed:     42,
    Endpoints: map[string]go_loadgen.Endpoint{
        "api": api,
    },
    Phases: []go_loadgen.Phase{
        {
            Duration: 60 * time.Second,
            RPS:      10_000,
            Targets:  []go_loadgen.Target{{Endpoint: "api", Weight: 1}},
        },
    },
})
if err != nil {
    log.Fatal(err)
}

report := workload.Run(context.Background())
log.Printf("scheduled=%d issued=%d dropped=%d missed=%d completed=%d", report.Scheduled, report.Issued, report.Dropped, report.Missed, report.Completed)

Client, DataProvider, and Collector implementations are called concurrently. Clients should reuse connections and honor their supplied context. For high result volume, prefer GobCollector; CSV conversion and its writer lock are deliberately not the low-overhead path.

Multi-Endpoint Workloads

Register every endpoint once, then split each phase's aggregate rate with integer weights:

Endpoints: map[string]go_loadgen.Endpoint{"read": readEndpoint, "write": writeEndpoint},
Phases: []go_loadgen.Phase{{
    Duration: time.Minute,
    RPS:      50_000,
    Targets: []go_loadgen.Target{
        {Endpoint: "read", Weight: 80},
        {Endpoint: "write", Weight: 20},
    },
}},

Each phase has its own deterministic random stream derived from Spec.Seed; no map lookup, mutex, or floating-point calculation occurs while choosing an endpoint.

License

Apache License 2.0. See LICENSE.

Documentation

Overview

Package go_loadgen provides a protocol-agnostic, open-loop load generator.

Applications provide typed Client, DataProvider, and Collector implementations. NewEndpoint adapts them into a heterogeneous endpoint set, while NewWorkload validates phases and compiles weighted target routing before a run begins.

Run stops scheduling at phase boundaries and drains issued requests by default. An optional drain timeout cancels requests that remain after scheduling ends.

Index

• type CSVCollector
  • func NewCSVCollector[R CSVSerializable](filePath string, flushInterval time.Duration) (*CSVCollector[R], error)
  • func (c *CSVCollector[R]) Close()
  • func (c *CSVCollector[R]) Collect(result R)
  • func (c *CSVCollector[R]) RunFlush(ctx context.Context)
• type CSVSerializable
• type Client
• type Collector
• type DataProvider
• type Endpoint
  • func NewEndpoint[C any, R any](client Client[C, R], provider DataProvider[C], collector Collector[R]) (Endpoint, error)
• type GobCollector
  • func NewGobCollector[R any](filePath string, flushInterval time.Duration, opts ...GobCollectorOption) (*GobCollector[R], error)
  • func (c *GobCollector[R]) Close()
  • func (c *GobCollector[R]) CloseAndErr() error
  • func (c *GobCollector[R]) Collect(result R)
  • func (c *GobCollector[R]) Err() error
• type GobCollectorOption
  • func WithGobCollectorBufferSize(size int) GobCollectorOption
  • func WithGobCollectorGzip(level int) GobCollectorOption
• type Phase
• type Ramp
• type Report
• type Spec
• type Target
• type Workload
  • func NewWorkload(spec Spec) (*Workload, error)
  • func (w *Workload) Run(ctx context.Context) Report

Types

type CSVCollector

type CSVCollector[R CSVSerializable] struct {
	// contains filtered or unexported fields
}

CSVCollector can collect results and write them to a CSV file. It requires result types to implement CSVSerializable. It will write the headers on the first collect and then every flushInterval. Note that headers will be rewritten if a new collector is created.

func NewCSVCollector

func NewCSVCollector[R CSVSerializable](filePath string, flushInterval time.Duration) (*CSVCollector[R], error)

NewCSVCollector creates a new CSV collector and starts a goroutine to flush the collector every flushInterval.

func (*CSVCollector[R]) Close

func (c *CSVCollector[R]) Close()

Close flushes the CSV collector and closes the file.

func (*CSVCollector[R]) Collect

func (c *CSVCollector[R]) Collect(result R)

Collect collects a result and writes it to the CSV file.

func (*CSVCollector[R]) RunFlush

func (c *CSVCollector[R]) RunFlush(ctx context.Context)

RunFlush flushes the CSV collector every flushInterval.

type CSVSerializable

type CSVSerializable interface {
	// CSVHeaders returns the headers that should be used for a CSV file.
	CSVHeaders() []string
	// CSVRecord returns the record that should be used to store the struct as a row in a CSV file.
	CSVRecord() []string
}

CSVSerializable is a struct that can be serialized to CSV

type Client

type Client[C any, R any] interface {
	CallEndpoint(context.Context, C) R
}

Client invokes one endpoint request. Implementations must be safe for concurrent use.

type Collector

type Collector[R any] interface {
	Collect(R)
	Close()
}

Collector receives one completed endpoint result. Implementations must be safe for concurrent use.

type DataProvider

type DataProvider[C any] interface {
	GetData() C
}

DataProvider creates one endpoint request. Implementations must be safe for concurrent use and should avoid blocking.

type Endpoint

type Endpoint interface {
	// contains filtered or unexported methods
}

Endpoint is a compiled unit of work. Endpoints are created with NewEndpoint.

func NewEndpoint

func NewEndpoint[C any, R any](client Client[C, R], provider DataProvider[C], collector Collector[R]) (Endpoint, error)

NewEndpoint adapts typed request generation, invocation, and result collection into an endpoint that can be used in a heterogeneous workload.

type GobCollector

type GobCollector[R any] struct {
	// contains filtered or unexported fields
}

GobCollector stores results as an async gob stream. It is a good default for very large experiments where CSV conversion and writer lock contention are too expensive.

func NewGobCollector

func NewGobCollector[R any](filePath string, flushInterval time.Duration, opts ...GobCollectorOption) (*GobCollector[R], error)

NewGobCollector creates a collector that writes results as a gob stream.

func (*GobCollector[R]) Close

func (c *GobCollector[R]) Close()

Close drains queued results, flushes the gob stream, and closes the file.

func (*GobCollector[R]) CloseAndErr

func (c *GobCollector[R]) CloseAndErr() error

CloseAndErr closes the collector and returns the first asynchronous write, flush, close, or post-close collection error observed by the collector.

func (*GobCollector[R]) Collect

func (c *GobCollector[R]) Collect(result R)

Collect queues a result to be written by the collector's writer goroutine.

func (*GobCollector[R]) Err

func (c *GobCollector[R]) Err() error

Err returns the first asynchronous write, flush, close, or post-close collection error observed by the collector.

type GobCollectorOption

type GobCollectorOption func(*gobCollectorConfig)

GobCollectorOption configures a GobCollector.

func WithGobCollectorBufferSize

func WithGobCollectorBufferSize(size int) GobCollectorOption

WithGobCollectorBufferSize configures how many results can queue before Collect blocks.

func WithGobCollectorGzip

func WithGobCollectorGzip(level int) GobCollectorOption

WithGobCollectorGzip enables gzip compression. Use gzip.BestSpeed for better write throughput, or gzip.BestCompression for lower storage cost.

type Phase

type Phase struct {
	StartAt  time.Duration
	Duration time.Duration
	RPS      uint64
	Ramp     *Ramp
	Targets  []Target
}

Phase schedules an open-loop offered rate. RPS is the total rate before target splitting.

type Ramp

type Ramp struct {
	To    uint64
	Step  uint64
	Every time.Duration
}

Ramp changes a phase's offered rate by Step every Every interval, ending at To. To may be lower than the phase RPS.

type Report

type Report struct {
	Scheduled     uint64
	Issued        uint64
	Dropped       uint64
	Missed        uint64
	Completed     uint64
	PeakInFlight  uint64
	DrainTimedOut bool
	// SchedulingDuration ends when no phase can issue another arrival.
	SchedulingDuration time.Duration
	// Duration includes the post-scheduling drain.
	Duration time.Duration
}

Report contains the actual load generator outcome. Scheduled is the number of arrivals requested by phases; Issued is the number passed to endpoint execution.

type Spec

type Spec struct {
	Duration  time.Duration
	Seed      uint64
	Endpoints map[string]Endpoint
	Phases    []Phase

	// MaxInFlight bounds outstanding requests. Zero leaves it unbounded.
	// When full, arrivals are dropped so the schedule remains open-loop.
	MaxInFlight uint64
	// DrainTimeout cancels outstanding requests after scheduling ends. Zero waits indefinitely.
	DrainTimeout time.Duration
}

Spec describes a workload before endpoint names and target weights are compiled.

type Target

type Target struct {
	Endpoint string
	Weight   uint32
}

Target assigns part of a phase's offered rate to an endpoint. Weight must be positive.

type Workload

type Workload struct {
	// contains filtered or unexported fields
}

Workload is an immutable, validated workload ready to run.

func NewWorkload

func NewWorkload(spec Spec) (*Workload, error)

NewWorkload validates a workload and compiles endpoint routing. It performs no allocation or endpoint lookup during request dispatch.

func (*Workload) Run

func (w *Workload) Run(ctx context.Context) Report

Run issues all phase arrivals, then waits for their completion. The supplied context is only external cancellation; phase deadlines never cancel requests.