sink

Documentation

Index

• type ClickHouseStorage
  • func NewClickhouseStorage(config ClickHouseStorageConfig) *ClickHouseStorage
  • func (c *ClickHouseStorage) BatchInsert(ctx context.Context, messages []sinkmodels.SinkMessage) error
• type ClickHouseStorageConfig
• type InsertEventsQuery
  • func (q InsertEventsQuery) ToSQL() (string, []interface{}, error)
• type MeterStore
• type NamespaceStore
  • func NewNamespaceStore() *NamespaceStore
  • func (n *NamespaceStore) AddMeter(meter models.Meter)
  • func (n *NamespaceStore) ValidateEvent(_ context.Context, m *sinkmodels.SinkMessage)
• type Sink
  • func NewSink(config SinkConfig) (*Sink, error)
  • func (s *Sink) Close() error
  • func (s *Sink) Run(ctx context.Context) error
• type SinkBuffer
  • func NewSinkBuffer() *SinkBuffer
  • func (b *SinkBuffer) Add(message sinkmodels.SinkMessage)
  • func (b *SinkBuffer) Dequeue() []sinkmodels.SinkMessage
  • func (b *SinkBuffer) RemoveByPartitions(partitions []kafka.TopicPartition)
  • func (b *SinkBuffer) Size() int
• type SinkConfig
• type Storage

Types

type ClickHouseStorage

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

func NewClickhouseStorage

func NewClickhouseStorage(config ClickHouseStorageConfig) *ClickHouseStorage

func (*ClickHouseStorage) BatchInsert

func (c *ClickHouseStorage) BatchInsert(ctx context.Context, messages []sinkmodels.SinkMessage) error

type ClickHouseStorageConfig

type ClickHouseStorageConfig struct {
	ClickHouse clickhouse.Conn
	Database   string
}

type InsertEventsQuery

type InsertEventsQuery struct {
	Database string
	Messages []sinkmodels.SinkMessage
}

func (InsertEventsQuery) ToSQL

func (q InsertEventsQuery) ToSQL() (string, []interface{}, error)

type MeterStore

type MeterStore struct {
	Meters []models.Meter
}

type NamespaceStore

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

func NewNamespaceStore

func NewNamespaceStore() *NamespaceStore

func (*NamespaceStore) AddMeter

func (n *NamespaceStore) AddMeter(meter models.Meter)

func (*NamespaceStore) ValidateEvent

func (n *NamespaceStore) ValidateEvent(_ context.Context, m *sinkmodels.SinkMessage)

ValidateEvent validates a single event by matching it with the corresponding meter if any

type Sink

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

func NewSink

func NewSink(config SinkConfig) (*Sink, error)

func (*Sink) Close

func (s *Sink) Close() error

func (*Sink) Run

func (s *Sink) Run(ctx context.Context) error

Run starts the Kafka consumer and sinks the events to Clickhouse

type SinkBuffer

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

func NewSinkBuffer

func NewSinkBuffer() *SinkBuffer

func (*SinkBuffer) Add

func (b *SinkBuffer) Add(message sinkmodels.SinkMessage)

func (*SinkBuffer) Dequeue

func (b *SinkBuffer) Dequeue() []sinkmodels.SinkMessage

func (*SinkBuffer) RemoveByPartitions

func (b *SinkBuffer) RemoveByPartitions(partitions []kafka.TopicPartition)

RemoveByPartitions removes messages from the buffer by partitions Useful when partitions are revoked.

func (*SinkBuffer) Size

func (b *SinkBuffer) Size() int

type SinkConfig

type SinkConfig struct {
	Logger          *slog.Logger
	Tracer          trace.Tracer
	MetricMeter     metric.Meter
	MeterRepository meter.Repository
	Storage         Storage
	Deduplicator    dedupe.Deduplicator
	Consumer        *kafka.Consumer
	// MinCommitCount is the minimum number of messages to wait before flushing the buffer.
	// Whichever happens earlier MinCommitCount or MaxCommitWait will trigger a flush.
	MinCommitCount int
	// MaxCommitWait is the maximum time to wait before flushing the buffer
	MaxCommitWait time.Duration
	// NamespaceRefetch is the interval to refetch exsisting namespaces and meters
	// this information is used to configure which topics the consumer subscribes and
	// the meter configs used in event validation.
	NamespaceRefetch time.Duration

	// FlushEventHandlers is an optional lifecycle hook, allowing to act on successful batch
	// flushes. To prevent blocking the main sink logic this is always called in a go routine.
	FlushEventHandler flushhandler.FlushEventHandler

	// FlushSuccessTimeout is the timeout for the OnFlushSuccess callback, after this period the context
	// of the callback will be canceled.
	FlushSuccessTimeout time.Duration
}

type Storage

type Storage interface {
	BatchInsert(ctx context.Context, messages []sinkmodels.SinkMessage) error
}