trogdor

Documentation

Index

• Constants
• Variables
• func DeleteTask(coordinatorURL string, taskId string) (*http.Response, []byte, error)
• func TaskStatuses(coordinatorURL string, earliestStartMs int64, latestStartMs int64, ...) (map[string]*TaskStatus, error)
• type AdminConf
  • func (a *AdminConf) EnableGaussianPartitioner(partitionCount int)
  • func (r *AdminConf) ParseConfig(adminConfFile string) error
  • func (a *AdminConf) ToPartitionSpecConfig() *PartitionsSpecConfig
• type ConnectionStressSpec
• type ConnectionStressTestConfig
• type ConsumerOptions
• type ConsumerTestConfig
• type KeyGeneratorSpec
• type PartitionsSpec
• type PartitionsSpecConfig
• type ProducerOptions
  • func (po *ProducerOptions) MessagesPerSec(throughputMbPerSec float32) uint64
• type ProducerTestConfig
• type RecordBatchVerifierSpec
• type ScenarioConfig
• type ScenarioSpec
  • func (r *ScenarioSpec) CreateScenario(scenarioConfig ScenarioConfig)
• type SustainedConnectionSpec
• type SustainedConnectionTestConfig
• type TaskId
  • func (t *TaskId) Name() string
• type TaskSpec
  • func (r *TaskSpec) CreateTask(coordinatorURL string) (*http.Response, []byte, error)
  • func (r *TaskSpec) Status(coordinatorURL string) (*TaskStatus, error)
• type TaskStatus
• type TopicSpec
• type TransactionGeneratorSpec
• type ValueGeneratorSpec

Constants

const (
	PENDING_TASK_STATE  = "PENDING"
	RUNNING_TASK_STATE  = "RUNNING"
	STOPPING_TASK_STATE = "STOPPING"
	DONE_TASK_STATE     = "DONE"
)

const CONNECTION_STRESS_SPEC_CLASS = "org.apache.kafka.trogdor.workload.ConnectionStressSpec"

const CONSUME_BENCH_SPEC_CLASS = "org.apache.kafka.trogdor.workload.ConsumeBenchSpec"

const (
	GAUSSIAN_PARTITIONER = "org.apache.kafka.trogdor.workload.partitioner.GaussianPartitioner"
)

const PRODUCE_BENCH_SPEC_CLASS = "org.apache.kafka.trogdor.workload.ProduceBenchSpec"

const SUSTAINED_CONNECTION_SPEC_CLASS = "org.apache.kafka.trogdor.workload.SustainedConnectionSpec"

Variables

var DefaultKeyGeneratorSpec = KeyGeneratorSpec{Type: "sequential", Size: 4, StartOffset: 0}

var DefaultTransactionGeneratorSpec = TransactionGeneratorSpec{Type: "uniform", MessagesPerTransaction: 100}

var DefaultValueGeneratorSpec = ValueGeneratorSpec{ValueType: "uniformRandom", Size: 900, Padding: 100}

use mostly random values to best simulate real-life data compression

Functions

func DeleteTask

func DeleteTask(coordinatorURL string, taskId string) (*http.Response, []byte, error)

func TaskStatuses

func TaskStatuses(coordinatorURL string, earliestStartMs int64, latestStartMs int64, state string) (map[string]*TaskStatus, error)

Returns all the Trogdor Tasks from the coordinator that match the filter Note that earliestStartMs and latestStartMs denote the time when the task was actually started, not scheduled to start

Types

type AdminConf

type AdminConf struct {
	CompressionType                    string  `json:"compression.type,omitempty"`
	Acks                               string  `json:"acks,omitempty"`
	AutoOffsetReset                    string  `json:"auto.offset.reset,omitempty"`
	BatchSize                          int64   `json:"batch.size,omitempty"`
	EnableIdempotence                  string  `json:"enable.idempotence,omitempty"`
	LingerMs                           int64   `json:"linger.ms,omitempty"`
	RetentionMs                        int64   `json:"retention.ms,omitempty"`
	MaxInFlightRequestsPerConnection   uint64  `json:"max.in.flight.requests.per.connection,omitempty"`
	RetryBackoffMs                     int64   `json:"retry.backoff.ms,omitempty"`
	SaslJaasConfig                     string  `json:"sasl.jaas.config,omitempty"`
	SecurityProtocol                   string  `json:"security.protocol,omitempty"`
	SslEndpointIdentificationAlgorithm string  `json:"ssl.endpoint.identification.algorithm,omitempty"`
	SaslMechanism                      string  `json:"sasl.mechanism,omitempty"`
	SslProtocol                        string  `json:"ssl.protocol,omitempty"`
	SslKeyPassword                     string  `json:"ssl.key.password,omitempty"`
	SslKeystoreLocation                string  `json:"ssl.keystore.location,omitempty"`
	SslKeystorePassword                string  `json:"ssl.keystore.password,omitempty"`
	SslKeystoreType                    string  `json:"ssl.keystore.type,omitempty"`
	SslTruststoreLocation              string  `json:"ssl.truststore.location,omitempty"`
	SslTruststorePassword              string  `json:"ssl.truststore.password,omitempty"`
	SslTruststoreType                  string  `json:"ssl.truststore.type,omitempty"`
	Partitioner                        *string `json:"partitioner.class,omitempty"`
	GaussianPartitionerMean            *int64  `json:"confluent.gaussian.partitioner.mean,omitempty"`
	GaussianPartitionerStd             *int64  `json:"confluent.gaussian.partitioner.std,omitempty"`
}

func (*AdminConf) EnableGaussianPartitioner

func (a *AdminConf) EnableGaussianPartitioner(partitionCount int)

func (*AdminConf) ParseConfig

func (r *AdminConf) ParseConfig(adminConfFile string) error

func (*AdminConf) ToPartitionSpecConfig

func (a *AdminConf) ToPartitionSpecConfig() *PartitionsSpecConfig

type ConnectionStressSpec

type ConnectionStressSpec struct {
	Class                   string     `json:"class"`
	StartMs                 uint64     `json:"startMs"`
	DurationMs              uint64     `json:"durationMs"`
	ClientNode              string     `json:"clientNode,omitempty"`
	BootstrapServers        string     `json:"bootstrapServers"`
	TargetConnectionsPerSec int        `json:"targetConnectionsPerSec"`
	NumThreads              int        `json:"numThreads"`
	Action                  string     `json:"action"`
	CommonClientConf        *AdminConf `json:"commonClientConf,omitempty"`
}

type ConnectionStressTestConfig

type ConnectionStressTestConfig struct {
	TargetConnectionsPerSec int
	NumThreads              int
	Action                  string
}

type ConsumerOptions

type ConsumerOptions struct {
	ConsumerGroup       string
	RecordBatchVerifier *RecordBatchVerifierSpec
}

type ConsumerTestConfig

type ConsumerTestConfig struct {
	TopicSpec       TopicSpec
	MessagesPerSec  uint64 // the total messages per second we want this scenario to have
	ConsumerOptions ConsumerOptions
}

type KeyGeneratorSpec

type KeyGeneratorSpec struct {
	Type        string `json:"type"`
	Size        uint64 `json:"size"`
	StartOffset uint64 `json:"startOffset"`
}

type PartitionsSpec

type PartitionsSpec struct {
	NumPartitions        uint64                `json:"numPartitions"`
	ReplicationFactor    uint64                `json:"replicationFactor"`
	PartitionsSpecConfig *PartitionsSpecConfig `json:"configs"`
}

type PartitionsSpecConfig

type PartitionsSpecConfig struct {
	RetentionMs int64 `json:"retention.ms,omitempty"`
}

type ProducerOptions

type ProducerOptions struct {
	ValueGenerator       ValueGeneratorSpec
	TransactionGenerator TransactionGeneratorSpec
	KeyGenerator         KeyGeneratorSpec
}

func (*ProducerOptions) MessagesPerSec

func (po *ProducerOptions) MessagesPerSec(throughputMbPerSec float32) uint64

Returns the number of messages per second we would need in order to achieve the desired throughput in MBs

type ProducerTestConfig

type ProducerTestConfig struct {
	TopicSpec       TopicSpec
	MessagesPerSec  uint64 // the total messages per second we want this scenario to have
	ProducerOptions ProducerOptions
}

type RecordBatchVerifierSpec

type RecordBatchVerifierSpec struct {
	Type string `json:"type"`
}

type ScenarioConfig

type ScenarioConfig struct {
	ScenarioID         TaskId
	Class              string
	TaskCount          int
	DurationMs         uint64
	SlowStartPerStepMs uint64
	StartMs            uint64
	BootstrapServers   string
	AdminConf          AdminConf
	ClientNodes        []string // all the configured trogdor nodes

	ProducerTestConfig            ProducerTestConfig
	ConsumerTestConfig            ConsumerTestConfig
	ConnectionStressTestConfig    ConnectionStressTestConfig
	SustainedConnectionTestConfig SustainedConnectionTestConfig
}

type ScenarioSpec

type ScenarioSpec struct {
	TaskSpecs []TaskSpec
	UsedNames map[string]bool
}

func (*ScenarioSpec) CreateScenario

func (r *ScenarioSpec) CreateScenario(scenarioConfig ScenarioConfig)

type SustainedConnectionSpec

type SustainedConnectionSpec struct {
	Class                   string             `json:"class"`
	StartMs                 uint64             `json:"startMs"`
	DurationMs              uint64             `json:"durationMs"`
	BootstrapServers        string             `json:"bootstrapServers"`
	ProducerConnectionCount uint64             `json:"producerConnectionCount"`
	ConsumerConnectionCount uint64             `json:"consumerConnectionCount"`
	MetadataConnectionCount uint64             `json:"metadataConnectionCount"`
	NumThreads              uint64             `json:"numThreads"`
	RefreshRateMs           uint64             `json:"refreshRateMs"`
	ClientNode              string             `json:"clientNode,omitempty"`
	ConsumerConf            *AdminConf         `json:"consumerConf,omitempty"`
	ProducerConf            *AdminConf         `json:"producerConf,omitempty"`
	AdminClientConf         *AdminConf         `json:"adminClientConf,omitempty"`
	CommonClientConf        *AdminConf         `json:"commonClientConf,omitempty"`
	KeyGenerator            KeyGeneratorSpec   `json:"keyGenerator,omitempty"`
	ValueGenerator          ValueGeneratorSpec `json:"valueGenerator,omitempty"`
	TopicName               string             `json:"topicName,omitempty"`
}

type SustainedConnectionTestConfig

type SustainedConnectionTestConfig struct {
	ProducerConnectionCount uint64
	ConsumerConnectionCount uint64
	MetadataConnectionCount uint64
	NumThreads              uint64
	RefreshRateMs           uint64
	KeyGenerator            KeyGeneratorSpec
	ValueGenerator          ValueGeneratorSpec
	TopicName               string
}

type TaskId

type TaskId struct {
	TaskType string
	StartMs  uint64
	Desc     string // arbitrary task identifier
	// contains filtered or unexported fields
}

a structured name of a Trogdor task

func (*TaskId) Name

func (t *TaskId) Name() string

Returns the name of this Task in the following form: {TASK_TYPE}.{START_TIMESTAMP}.{AGENT_ID}.{OPTIONAL_TASK_DESCRIPTION}.{OPTIONAL_DUPLICATE_ID} e.g ShortLivedConsume.2014-07-16T20:55:46Z.cc-trogdor-service-agent-0.topic-1.2L.1

type TaskSpec

type TaskSpec struct {
	ID   string          `json:"id"`
	Spec json.RawMessage `json:"spec"`
}

func (*TaskSpec) CreateTask

func (r *TaskSpec) CreateTask(coordinatorURL string) (*http.Response, []byte, error)

func (*TaskSpec) Status

func (r *TaskSpec) Status(coordinatorURL string) (*TaskStatus, error)

type TaskStatus

type TaskStatus struct {
	State     string          `json:"state"`
	Spec      taskStatusSpec  `json:"spec"`
	StartedMs int64           `json:"startedMs"`
	DoneMs    int64           `json:"doneMs"`
	Cancelled bool            `json:"cancelled"`
	Error     string          `json:"error"`
	Status    json.RawMessage `json:"status"`
}

type TopicSpec

type TopicSpec struct {
	PartitionsSpec *PartitionsSpec
	TopicName      string
}

type TransactionGeneratorSpec

type TransactionGeneratorSpec struct {
	Type                   string `json:"type"`
	MessagesPerTransaction uint64 `json:"messagesPerTransaction"`
}

type ValueGeneratorSpec

type ValueGeneratorSpec struct {
	ValueType string `json:"type"`
	Size      uint64 `json:"size"`
	Padding   uint64 `json:"padding"`
}