m3nsch

README

m3nsch

m3nsch (pronounced mensch) is a load testing tool for M3DB. It has two components:

• m3nsch_server: long lived process which does the load generation
• m3nsch_client: cli wrapper which controls the myriad m3nsch_server(s)

A typical deploy will have multiple hosts, each running a single m3nsch_server instance, and a single m3nsch_client used to control them.

Build

$ make prod-m3nsch_server
$ make prod-m3nsch_client
$ cat <<EOF > server-conf.yaml
server:
  listenAddress: "0.0.0.0:12321"
  debugAddress: "0.0.0.0:13441"
  cpuFactor: 0.9

metrics:
  sampleRate: 0.1
  m3:
    hostPort: "<host-port-for-tally>"
    service: "m3nsch"
    includeHost: true
    env: development

m3nsch:
  concurrency: 2000
  numPointsPerDatum: 60

# any additional configs you may have
EOF

Deploy

(1) Transfer the m3nsch_server binary, and server-conf.yaml to all the hosts to be used to generate hosts, e.g. Ingesters

(2) On each host from (1), kick of the server process by running:

./m3nsch_server -f server-conf.yaml

(3) Transfer m3nsch_client binary to a host with network connectivity to all the hosts.

Sample Usage

# set an env var containing the host endpoints seperated by commas
$ export ENDPOINTS="host1:12321,host2:12321"

# get the status of the various endpoints, make sure all are healthy
$ ./m3nsch_client --endpoints $ENDPOINTS status

# investigate the various options available during initialization
$ ./m3nsch_client init --help
...
Flags:
  -b, --basetime-offset duration   offset from current time to use for load, e.g. -2m, -30s (default -2m0s)
  -c, --cardinality int            aggregate workload cardinality (default 10000)
  -f, --force                      force initialization, stop any running workload
  -i, --ingress-qps int            aggregate workload ingress qps (default 1000)
  -p, --metric-prefix string       prefix added to each metric (default "m3nsch_")
  -n, --namespace string           target namespace (default "testmetrics")
  -v, --target-env string          target env for load test (default "test")
  -z, --target-zone string         target zone for load test (default "sjc1")
  -t, --token string               [required] unique identifier required for all subsequent interactions on this workload

Global Flags:
  -e, --endpoints stringSlice   host:port for each of the agent process endpoints

# initialize the servers, set any workload parameters, target env/zone
# the command below targets the production sjc1 m3db cluster with each `m3nsch_server` attempting to
# sustain a load of 100K writes/s from a set of 1M metrics
$ ./m3nsch_client --endpoints $ENDPOINTS init \
  --token prateek-sample \
  --target-env prod      \
  --target-zone sjc1     \
  --ingress-qps 100000   \
  --cardinality 1000000  \

# start the load generation
$ ./m3nsch_client --endpoints $ENDPOINTS start

# modifying the running load uses many of the same options as `init`
$ ./m3nsch_client modify --help
...
Flags:
  -b, --basetime-offset duration   offset from current time to use for load, e.g. -2m, -30s (default -2m0s)
  -c, --cardinality int            aggregate workload cardinality (default 10000)
  -i, --ingress-qps int            aggregate workload ingress qps (default 1000)
  -p, --metric-prefix string       prefix added to each metric (default "m3nsch_")
  -n, --namespace string           target namespace (default "testmetrics")

Global Flags:
  -e, --endpoints stringSlice   host:port for each of the agent process endpoints

# the command below bumps up the workload on each `m3nsch_server`, to sustain
# a load of 1M writes/s from a set of 10M metrics
$ ./m3nsch_client --endpoints $ENDPOINTS
  --ingress-qps 1000000   \
  --cardinality 10000000  \

# finally, stop the load
$ ./m3nsch_client --endpoints $ENDPOINTS stop

# probably want to teardown the running server processes on the various hosts

Documentation

Index

• type Agent
• type AgentOptions
• type AgentStatus
• type Coordinator
• type CoordinatorOptions
• type NewSessionFn
• type Status
• type Workload

Types

type Agent

type Agent interface {
	// Status returns the status of the agent process.
	Status() AgentStatus

	// Workload returns Workload currently configured on the agent process.
	Workload() Workload

	// SetWorkload sets the Workload on the agent process.
	SetWorkload(Workload)

	// Init initializes resources required by the agent process.
	Init(token string, w Workload, force bool, targetZone string, targetEnv string) error

	// Start begins the load generation process if the agent is Initialized, or errors if it is not.
	Start() error

	// Stop ends the load generation process if the agent is Running.
	Stop() error

	// MaxQPS returns the maximum QPS this Agent is capable of driving.
	// MaxQPS := `AgentOptions.MaxWorkerQPS() * AgentOptions.Concurrency()`
	MaxQPS() int64
}

Agent refers to the process responsible for executing load generation.

type AgentOptions

type AgentOptions interface {
	// SetInstrumentOptions sets the InstrumentOptions
	SetInstrumentOptions(instrument.Options) AgentOptions

	// InstrumentOptions returns the set InstrumentOptions
	InstrumentOptions() instrument.Options

	// SetMaxWorkerQPS sets the maximum QPS per 'worker' go-routine used in
	// the load generation process.
	SetMaxWorkerQPS(int64) AgentOptions

	// SetMaxWorkerQPS returns the maximum QPS per 'worker' go-routine used in
	// the load generation process.
	MaxWorkerQPS() int64

	// SetConcurrency sets the number of concurrent go routines used.
	SetConcurrency(int) AgentOptions

	// Concurrency returns the number of concurrent go routines usable during
	// load generation.
	Concurrency() int

	// SetNewSessionFn sets the new session function
	SetNewSessionFn(fn NewSessionFn) AgentOptions

	// NewSessionFn returns the new session function
	NewSessionFn() NewSessionFn

	// SetTimeUnit sets the time unit to use during load operations.
	SetTimeUnit(xtime.Unit) AgentOptions

	// TimeUnit returns the time unit used during load operations.
	TimeUnit() xtime.Unit
}

AgentOptions is a collection of knobs to control Agent behavior.

type AgentStatus

type AgentStatus struct {
	// Status refers to the agent process' running status
	Status Status

	// Token (if non-empty) is the breadcrumb used to Init the agent process'
	Token string

	// MaxQPS is the maximum QPS attainable by the Agent process
	MaxQPS int64

	// Workload is the currently configured workload on the agent process
	Workload Workload
}

AgentStatus is a collection of attributes capturing the state of a agent process.

type Coordinator

type Coordinator interface {
	// Status returns the status of the agents known to this process.
	Status() (map[string]AgentStatus, error)

	// Init acquires resources required on known agent processes to be able to
	// generate load.
	// NB(prateek): Init takes ~30s to initialize m3db.Session objects
	Init(token string, w Workload, force bool, targetZone string, targetEnv string) error

	// Workload returns the aggregate workload currently initialized on the
	// agent processes known to the coordinator process.
	Workload() (Workload, error)

	// SetWorkload splits the specified workload into smaller chunks, and distributes them
	// across known agent processes.
	SetWorkload(Workload) error

	// Start begins the load generation process on known agent processes.
	Start() error

	// Stops ends the load generation process on known agent processes.
	Stop() error

	// Teardown releases resources held by the Coordinator process (connections, state tracking structures).
	Teardown() error
}

Coordinator refers to the process responsible for synchronizing load generation.

type CoordinatorOptions

type CoordinatorOptions interface {
	// SetInstrumentOptions sets the InstrumentOptions
	SetInstrumentOptions(instrument.Options) CoordinatorOptions

	// InstrumentOptions returns the set InstrumentOptions
	InstrumentOptions() instrument.Options

	// SetTimeout sets the timeout for rpc interaction
	SetTimeout(time.Duration) CoordinatorOptions

	// Timeout returns the timeout for rpc interaction
	Timeout() time.Duration

	// SetParallelOperations sets a flag determining if the operations
	// performed by the coordinator against various endpoints are to be
	// parallelized or not
	SetParallelOperations(bool) CoordinatorOptions

	// ParallelOperations returns a flag indicating if the operations
	// performed by the coordinator against various endpoints are to be
	// parallelized or not
	ParallelOperations() bool
}

CoordinatorOptions is a collection of the various knobs to control Coordinator behavior.

type NewSessionFn

type NewSessionFn func(targetZone string, targetEnv string) (client.Session, error)

NewSessionFn creates a new client.Session for the specified environment, zone.

type Status

type Status int

Status represents the various states the load generation processes may exist in.

const (
	// StatusUninitialized refers to a load generation process yet to be initialized.
	StatusUninitialized Status = iota

	// StatusInitialized refers to a load generation process which has been initialized.
	StatusInitialized

	// StatusRunning refers to a load generation process which is running.
	StatusRunning
)

type Workload

type Workload struct {
	// BaseTime is the epoch value for time used during load generation, all timestamps are
	// generated relative to it.
	BaseTime time.Time

	// Namespace is the target namespace to perform the writes against.
	Namespace string

	// MetricPrefix is the string prefixed to each metric used.
	MetricPrefix string

	// Cardinality is the number of unique metrics used.
	Cardinality int

	// IngressQPS is the number of metrics written per second.
	IngressQPS int

	// MetricStartIdx is an offset to control metric numbering. Can be safely ignored
	// by external callers.
	MetricStartIdx int

	// UniqueAmplifier is the percentage of unique metrics generated as a float
	// between 0.0 and 1.0 that will be unique. This allows for generating metrics
	// with steady cardinality rate over time.
	UniqueAmplifier float64
}

Workload is a collection of attributes required to define a load generation workload. TODO(prateek): add support for duration