sinker

Documentation

Index

• Constants
• Variables
• func RegisterMetrics()
• type ClickhouseSinker
  • func New(sink *sink.Sinker, loader *db.Loader, logger *zap.Logger, ...) (*ClickhouseSinker, error)
  • func (s *ClickhouseSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrpc.BlockScopedData, isLive *bool, ...) error
  • func (s *ClickhouseSinker) HandleBlockUndoSignal(ctx context.Context, data *pbsubstreamsrpc.BlockUndoSignal, ...) error
  • func (s *ClickhouseSinker) Run(ctx context.Context)
• type Stats
  • func NewStats(logger *zap.Logger) *Stats
  • func (s *Stats) Close()
  • func (s *Stats) LogNow()
  • func (s *Stats) RecordBlock(block bstream.BlockRef)
  • func (s *Stats) Start(each time.Duration, cursor *sink.Cursor)

Constants

const (
	HISTORICAL_BLOCK_FLUSH_EACH = 1000
	LIVE_BLOCK_FLUSH_EACH       = 1
)

Variables

var FlushCount = metrics.NewCounter("substreams_sink_clickhouse_store_flush_count", "The amount of flush that happened so far")

var FlushDuration = metrics.NewCounter("substreams_sink_clickhouse_store_flush_duration", "The amount of time spent flushing cache to db (in nanoseconds)")

var FlushedEntriesCount = metrics.NewGauge("substreams_sink_clickhouse_flushed_entries_count", "The number of flushed entries so far")

Functions

func RegisterMetrics

func RegisterMetrics()

Types

type ClickhouseSinker

type ClickhouseSinker struct {
	*shutter.Shutter
	*sink.Sinker
	// contains filtered or unexported fields
}

func New

func New(sink *sink.Sinker, loader *db.Loader, logger *zap.Logger, tracer logging.Tracer) (*ClickhouseSinker, error)

func (*ClickhouseSinker) HandleBlockScopedData

func (s *ClickhouseSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrpc.BlockScopedData, isLive *bool, cursor *sink.Cursor) error

func (*ClickhouseSinker) HandleBlockUndoSignal

func (s *ClickhouseSinker) HandleBlockUndoSignal(ctx context.Context, data *pbsubstreamsrpc.BlockUndoSignal, cursor *sink.Cursor) error

func (*ClickhouseSinker) Run

func (s *ClickhouseSinker) Run(ctx context.Context)

type Stats

type Stats struct {
	*shutter.Shutter
	// contains filtered or unexported fields
}

func NewStats

func NewStats(logger *zap.Logger) *Stats

func (*Stats) Close

func (s *Stats) Close()

func (*Stats) LogNow

func (s *Stats) LogNow()

func (*Stats) RecordBlock

func (s *Stats) RecordBlock(block bstream.BlockRef)

func (*Stats) Start

func (s *Stats) Start(each time.Duration, cursor *sink.Cursor)