services

Documentation

Overview

Package services defines an interface for canary services and related implementations

Package services defines an interface for canary services and related implementations

Package services defines some canary related services

Package services defines some canary related services

Package services defines an interface for canary services and related implementations

Package services defines some canary related services

Package services defines some canary related services

Package services defines some canary related services

Index

• Constants
• Variables
• func LivenessHandler() http.Handler
• func ReadinessHandler() http.Handler
• type Backoff
  • func NewBackoff(maxAttempts int, scale time.Duration, max time.Duration) *Backoff
  • func (b *Backoff) Delay() (time.Duration, error)
• type BackoffDelayOverflow
  • func (e *BackoffDelayOverflow) Error() string
• type CanaryMessage
  • func NewCanaryMessage(bytes []byte) CanaryMessage
  • func (cm CanaryMessage) Json() string
  • func (cm CanaryMessage) String() string
• type ConnectionService
  • func NewConnectionService(canaryConfig *config.CanaryConfig, saramaConfig *sarama.Config) ConnectionService
• type ConsumerService
  • func NewConsumerService(canaryConfig *config.CanaryConfig, client sarama.Client) ConsumerService
• type ConsumingStatus
• type ErrExpectedClusterSize
  • func (e *ErrExpectedClusterSize) Error() string
• type MaxAttemptsExceeded
  • func (e *MaxAttemptsExceeded) Error() string
• type ProducerService
  • func NewProducerService(canaryConfig *config.CanaryConfig, client sarama.Client) ProducerService
• type Status
• type StatusService
  • func NewStatusServiceService(canaryConfig *config.CanaryConfig) StatusService
• type TopicReconcileResult
• type TopicService
  • func NewTopicService(canaryConfig *config.CanaryConfig, saramaConfig *sarama.Config) TopicService

Constants

const (
	ScaleDefault       = 200 * time.Millisecond
	MaxDefault         = 5 * time.Minute
	MaxAttemptsDefault = 6
)

Variables

var (
	RecordsConsumedCounter uint64 = 0
)

var (
	RecordsProducedCounter uint64 = 0
)

Functions

func LivenessHandler

func LivenessHandler() http.Handler

func ReadinessHandler

func ReadinessHandler() http.Handler

Types

type Backoff

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

Backoff encapsulates computing delays for an exponential back-off, when an operation has to be retried

func NewBackoff

func NewBackoff(maxAttempts int, scale time.Duration, max time.Duration) *Backoff

NewBackoff returns an instance of a Backoff struct

func (*Backoff) Delay

func (b *Backoff) Delay() (time.Duration, error)

Delay computes a delay in terms of Duration (nanoseconds) based on the current Backoff instance configuration Returns the delay in terms of Duration (nanoseconds) and an error if the max attempts is reached, otherwise it's nil

type BackoffDelayOverflow

type BackoffDelayOverflow struct{}

Overflow on computed delay

func (*BackoffDelayOverflow) Error

func (e *BackoffDelayOverflow) Error() string

type CanaryMessage

type CanaryMessage struct {
	ProducerID string `json:"producerId"`
	MessageID  int    `json:"messageId"`
	Timestamp  int64  `json:"timestamp"`
}

CanaryMessage defines the payload of a canary message

func NewCanaryMessage

func NewCanaryMessage(bytes []byte) CanaryMessage

func (CanaryMessage) Json

func (cm CanaryMessage) Json() string

func (CanaryMessage) String

func (cm CanaryMessage) String() string

type ConnectionService

type ConnectionService interface {
	Open()
	Close()
}

func NewConnectionService

func NewConnectionService(canaryConfig *config.CanaryConfig, saramaConfig *sarama.Config) ConnectionService

NewConnectionService returns an instance of ConnectionService

type ConsumerService

type ConsumerService interface {
	Consume()
	Refresh()
	Leaders() (map[int32]int32, error)
	Close()
}

ConsumerService defines the service for consuming messages

func NewConsumerService

func NewConsumerService(canaryConfig *config.CanaryConfig, client sarama.Client) ConsumerService

NewConsumerService returns an instance of ConsumerService

type ConsumingStatus

type ConsumingStatus struct {
	TimeWindow time.Duration
	Percentage float64
}

ConsumingStatus defines consuming related status information

type ErrExpectedClusterSize

type ErrExpectedClusterSize struct{}

ErrExpectedClusterSize defines the error raised when the expected cluster size is not met

func (*ErrExpectedClusterSize) Error

func (e *ErrExpectedClusterSize) Error() string

type MaxAttemptsExceeded

type MaxAttemptsExceeded struct{}

MaxAttemptsExceeded defines the error for the max attempts exceeded

func (*MaxAttemptsExceeded) Error

func (e *MaxAttemptsExceeded) Error() string

type ProducerService

type ProducerService interface {
	Send(partitionsAssignments map[int32][]int32)
	Refresh()
	Close()
}

func NewProducerService

func NewProducerService(canaryConfig *config.CanaryConfig, client sarama.Client) ProducerService

NewProducerService returns an instance of ProductService

type Status

type Status struct {
	Consuming ConsumingStatus
}

Status defines useful status related information

type StatusService

type StatusService interface {
	Open()
	Close()
	StatusHandler() http.Handler
}

func NewStatusServiceService

func NewStatusServiceService(canaryConfig *config.CanaryConfig) StatusService

NewStatusService returns an instance of StatusService

type TopicReconcileResult

type TopicReconcileResult struct {
	// new partitions assignments across brokers
	Assignments map[int32][]int32
	// partition to leader assignments
	Leaders map[int32]int32
	// if a refresh metadata is needed
	RefreshProducerMetadata bool
}

TopicReconcileResult contains the result of a topic reconcile

type TopicService

type TopicService interface {
	Reconcile() (TopicReconcileResult, error)
	Close()
}

func NewTopicService

func NewTopicService(canaryConfig *config.CanaryConfig, saramaConfig *sarama.Config) TopicService

NewTopicService returns an instance of TopicService