sacura

README

sacura

Detect CloudEvents loss

Documentation

Index

• Constants
• func Main(ctx context.Context, config Config) error
• func NewTargeterGenerator(config Config, newUIID func() uuid.UUID, out chan<- ce.Event) vegeta.Targeter
• func StartReceiver(ctx context.Context, config ReceiverConfig, received chan<- ce.Event) error
• type Config
  • func FileConfig(r io.Reader) (Config, error)
• type Metrics
  • func StartSender(config Config, sentOut chan<- ce.Event) Metrics
• type OrderedConfig
• type ReceiverConfig
• type ReceiverFaultConfig
• type Report
• type SenderConfig
• type StateManager
  • func NewStateManager(config Config) *StateManager
  • func (s *StateManager) Diff() string
  • func (s *StateManager) GenerateReport() Report
  • func (s *StateManager) ReadReceived(received <-chan ce.Event) <-chan struct{}
  • func (s *StateManager) ReadSent(sent <-chan ce.Event) <-chan struct{}
  • func (s *StateManager) ReceivedCount() int
  • func (s *StateManager) Terminated(metrics Metrics)
• type StateManagerConfig

Constants

const (
	BenchmarkTimestampAttribute = "benchmarktimestamp"
)

const CloudEventIdHeader = "Cloudevent-Id"

Functions

func Main

func Main(ctx context.Context, config Config) error

func NewTargeterGenerator

func NewTargeterGenerator(config Config, newUIID func() uuid.UUID, out chan<- ce.Event) vegeta.Targeter

func StartReceiver

func StartReceiver(ctx context.Context, config ReceiverConfig, received chan<- ce.Event) error

Types

type Config

type Config struct {
	Sender SenderConfig `json:"sender" yaml:"sender"`

	Receiver ReceiverConfig `json:"receiver" yaml:"receiver"`

	Duration string         `json:"duration" yaml:"duration"`
	Ordered  *OrderedConfig `json:"ordered" yaml:"ordered"`

	ParsedDuration time.Duration
}

func FileConfig

func FileConfig(r io.Reader) (Config, error)

type Metrics

type Metrics struct {
	ProposedCount int            `json:"proposedCount"`
	AcceptedCount int            `json:"acceptedCount"`
	Metrics       vegeta.Metrics `json:"metrics"`
}

func StartSender

func StartSender(config Config, sentOut chan<- ce.Event) Metrics

type OrderedConfig

type OrderedConfig struct {
	NumPartitionKeys uint8 `json:"numPartitionKeys" yaml:"numPartitionKeys"`
}

type ReceiverConfig

type ReceiverConfig struct {
	Port                      int    `json:"port" yaml:"port"`
	Timeout                   string `json:"timeout" yaml:"timeout"`
	MaxDuplicatesPercentage   *int   `json:"maxDuplicatesPercentage" yaml:"maxDuplicatesPercentage"`
	IncludeRemoteAddressLabel *bool  `json:"includeRemoteAddressLabel" yaml:"includeRemoteAddressLabel"`

	ReceiverFaultConfig *ReceiverFaultConfig `json:"fault" yaml:"fault"`

	ParsedTimeout time.Duration
}

type ReceiverFaultConfig

type ReceiverFaultConfig struct {
	// MinSleepDuration is the minimum duration to sleep before sending the response.
	//
	// When MinSleepDuration is specified, MaxSleepDuration must be specified.
	MinSleepDuration *time.Duration `json:"minSleepDuration" yaml:"minSleepDuration"`
	// MaxSleepDuration is the maximum duration to sleep before sending the response.
	MaxSleepDuration *time.Duration `json:"maxSleepDuration" yaml:"maxSleepDuration"`
}

type Report

type Report struct {
	LostCount                int                 `json:"lostCount"`
	LostEventsByPartitionKey map[string][]string `json:"lostEvents"`
	DuplicateCount           int                 `json:"duplicateCount"`
	// DuplicateEventsByPartitionKey collects duplicate events by
	DuplicateEventsByPartitionKey map[string][]string `json:"duplicateEvents"`
	// ReceivedCount is the number of events received, including duplicates
	ReceivedCount int `json:"receivedCount"`
	// ReceivedEventsByPartitionKey collects all events by partition, including duplicates
	ReceivedEventsByPartitionKey map[string][]string `json:"-"`
	Terminated                   bool                `json:"terminated"`
	Metrics                      Metrics             `json:"metrics"`
}

type SenderConfig

type SenderConfig struct {
	Disabled           bool   `json:"disabled" yaml:"disabled"`
	Target             string `json:"target" yaml:"target"`
	FrequencyPerSecond int    `json:"frequency" yaml:"frequency"`
	Workers            uint64 `json:"workers" yaml:"workers"`
	KeepAlive          bool   `json:"keepAlive" yaml:"keepAlive"`
}

type StateManager

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

func NewStateManager

func NewStateManager(config Config) *StateManager

func (*StateManager) Diff

func (s *StateManager) Diff() string

func (*StateManager) GenerateReport

func (s *StateManager) GenerateReport() Report

func (*StateManager) ReadReceived

func (s *StateManager) ReadReceived(received <-chan ce.Event) <-chan struct{}

func (*StateManager) ReadSent

func (s *StateManager) ReadSent(sent <-chan ce.Event) <-chan struct{}

func (*StateManager) ReceivedCount

func (s *StateManager) ReceivedCount() int

func (*StateManager) Terminated

func (s *StateManager) Terminated(metrics Metrics)

type StateManagerConfig

type StateManagerConfig struct {
	Ordered bool
	OrderedConfig
}