e2e

Documentation

Index

• Constants
• type Config
  • func (c *Config) SetDefaults()
  • func (c *Config) Validate() error
• type EndToEndConsumerConfig
  • func (c *EndToEndConsumerConfig) SetDefaults()
  • func (c *EndToEndConsumerConfig) Validate() error
• type EndToEndMessage
• type EndToEndProducerConfig
  • func (c *EndToEndProducerConfig) SetDefaults()
  • func (c *EndToEndProducerConfig) Validate() error
• type EndToEndTopicConfig
  • func (c *EndToEndTopicConfig) SetDefaults()
  • func (c *EndToEndTopicConfig) Validate() error
• type Service
  • func NewService(ctx context.Context, cfg Config, logger *zap.Logger, kafkaSvc *kafka.Service, ...) (*Service, error)
  • func (s *Service) Start(ctx context.Context) error

Constants

const (
	EndToEndMessageStateCreated
	EndToEndMessageStateProducedSuccessfully
)

Types

type Config

type Config struct {
	Enabled         bool                   `koanf:"enabled"`
	TopicManagement EndToEndTopicConfig    `koanf:"topicManagement"`
	ProbeInterval   time.Duration          `koanf:"probeInterval"`
	Producer        EndToEndProducerConfig `koanf:"producer"`
	Consumer        EndToEndConsumerConfig `koanf:"consumer"`
}

func (*Config) SetDefaults

func (c *Config) SetDefaults()

func (*Config) Validate

func (c *Config) Validate() error

type EndToEndConsumerConfig

type EndToEndConsumerConfig struct {
	GroupIdPrefix             string `koanf:"groupIdPrefix"`
	DeleteStaleConsumerGroups bool   `koanf:"deleteStaleConsumerGroups"`

	// RoundtripSLA is the time duration from the moment where we try to produce until the moment where we consumed
	// the message. Therefore this should always be higher than the produceTimeout / SLA.
	RoundtripSla time.Duration `koanf:"roundtripSla"`
	CommitSla    time.Duration `koanf:"commitSla"`
}

func (*EndToEndConsumerConfig) SetDefaults

func (c *EndToEndConsumerConfig) SetDefaults()

func (*EndToEndConsumerConfig) Validate

func (c *EndToEndConsumerConfig) Validate() error

type EndToEndMessage

type EndToEndMessage struct {
	MinionID  string `json:"minionID"`     // unique for each running kminion instance
	MessageID string `json:"messageID"`    // unique for each message
	Timestamp int64  `json:"createdUtcNs"` // when the message was created, unix nanoseconds
	// contains filtered or unexported fields
}

type EndToEndProducerConfig

type EndToEndProducerConfig struct {
	AckSla       time.Duration `koanf:"ackSla"`
	RequiredAcks string        `koanf:"requiredAcks"`
}

func (*EndToEndProducerConfig) SetDefaults

func (c *EndToEndProducerConfig) SetDefaults()

func (*EndToEndProducerConfig) Validate

func (c *EndToEndProducerConfig) Validate() error

type EndToEndTopicConfig

type EndToEndTopicConfig struct {
	Enabled                bool          `koanf:"enabled"`
	Name                   string        `koanf:"name"`
	ReplicationFactor      int           `koanf:"replicationFactor"`
	PartitionsPerBroker    int           `koanf:"partitionsPerBroker"`
	ReconciliationInterval time.Duration `koanf:"reconciliationInterval"`
}

func (*EndToEndTopicConfig) SetDefaults

func (c *EndToEndTopicConfig) SetDefaults()

func (*EndToEndTopicConfig) Validate

func (c *EndToEndTopicConfig) Validate() error

type Service

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

func NewService

func NewService(ctx context.Context, cfg Config, logger *zap.Logger, kafkaSvc *kafka.Service, promRegisterer prometheus.Registerer) (*Service, error)

NewService creates a new instance of the e2e moinitoring service (wow)

func (*Service) Start

func (s *Service) Start(ctx context.Context) error

Start starts the service (wow)