ingestor

Documentation

Index

• Variables
• func FinishWriteRequest(wr *prompb.WriteRequest)
• func NewPendingBuffer() *pendingBuffer
• func NewSeriesWriter(conn pgxconn.PgxConn, labelsCache *cache.InvertedLabelsCache) *seriesWriter
• func NewWriteRequest() *prompb.WriteRequest
• type Cfg
• type DBIngestor
  • func NewPgxIngestor(conn pgxconn.PgxConn, cache cache.MetricCache, sCache cache.SeriesCache, ...) (*DBIngestor, error)
  • func NewPgxIngestorForTests(conn pgxconn.PgxConn, cfg *Cfg) (*DBIngestor, error)
  • func (ingestor *DBIngestor) Close()
  • func (ingestor *DBIngestor) CompleteMetricCreation(ctx context.Context) error
  • func (ingestor *DBIngestor) Dispatcher() model.Dispatcher
  • func (ingestor *DBIngestor) IngestMetrics(ctx context.Context, r *prompb.WriteRequest) (numInsertablesIngested uint64, numMetadataIngested uint64, err error)
  • func (ingestor *DBIngestor) IngestTraces(ctx context.Context, traces ptrace.Traces) error
  • func (ingestor *DBIngestor) SeriesCache() cache.SeriesCache
• type DBInserter
• type ExemplarLabelFormatter
  • func NewExamplarLabelFormatter(conn pgxconn.PgxConn, exemplarKeyPosCache cache.PositionCache) *ExemplarLabelFormatter
• type ExemplarVisitor
• type ReadOnlyIngestor
  • func (ReadOnlyIngestor) Close()
  • func (ReadOnlyIngestor) IngestMetrics(context.Context, *prompb.WriteRequest) (numInsertablesIngested uint64, numMetadataIngested uint64, err error)
  • func (ReadOnlyIngestor) IngestTraces(context.Context, ptrace.Traces) error
• type SeriesVisitor

Variables

var ErrDispatcherClosed = fmt.Errorf("dispatcher is closed")

Functions

func FinishWriteRequest

func FinishWriteRequest(wr *prompb.WriteRequest)

FinishWriteRequest adds the *prompb.WriteRequest back into the pool after setting parameters to default.

func NewPendingBuffer

func NewPendingBuffer() *pendingBuffer

func NewSeriesWriter

func NewSeriesWriter(conn pgxconn.PgxConn, labelsCache *cache.InvertedLabelsCache) *seriesWriter

func NewWriteRequest

func NewWriteRequest() *prompb.WriteRequest

NewWriteRequest returns a new *prompb.WriteRequest from the pool.

Types

type Cfg

type Cfg struct {
	MetricsAsyncAcks        bool
	TracesAsyncAcks         bool
	NumCopiers              int
	DisableEpochSync        bool
	IgnoreCompressedChunks  bool
	InvertedLabelsCacheSize uint64
	TracesBatchTimeout      time.Duration
	TracesMaxBatchSize      int
	TracesBatchWorkers      int
}

type DBIngestor

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

DBIngestor ingest the TimeSeries data into Timescale database.

func NewPgxIngestor

func NewPgxIngestor(conn pgxconn.PgxConn, cache cache.MetricCache, sCache cache.SeriesCache, eCache cache.PositionCache, lCache *cache.InvertedLabelsCache, cfg *Cfg) (*DBIngestor, error)

NewPgxIngestor returns a new Ingestor that uses connection pool and a metrics cache for caching metric table names.

func NewPgxIngestorForTests

func NewPgxIngestorForTests(conn pgxconn.PgxConn, cfg *Cfg) (*DBIngestor, error)

NewPgxIngestorForTests returns a new Ingestor that write to PostgreSQL using PGX with an empty config, a new default size metrics cache and a non-ha-aware data parser

func (*DBIngestor) Close

func (ingestor *DBIngestor) Close()

Close closes the ingestor

func (*DBIngestor) CompleteMetricCreation

func (ingestor *DBIngestor) CompleteMetricCreation(ctx context.Context) error

Parts of metric creation not needed to insert data

func (*DBIngestor) Dispatcher

func (ingestor *DBIngestor) Dispatcher() model.Dispatcher

func (*DBIngestor) IngestMetrics

func (ingestor *DBIngestor) IngestMetrics(ctx context.Context, r *prompb.WriteRequest) (numInsertablesIngested uint64, numMetadataIngested uint64, err error)

IngestMetrics transforms and ingests the timeseries data into Timescale database. input:

req the WriteRequest backing tts. It will be added to our WriteRequest
    pool when it is no longer needed.

func (*DBIngestor) IngestTraces

func (ingestor *DBIngestor) IngestTraces(ctx context.Context, traces ptrace.Traces) error

func (*DBIngestor) SeriesCache

func (ingestor *DBIngestor) SeriesCache() cache.SeriesCache

type DBInserter

type DBInserter interface {
	// IngestMetrics takes an array of TimeSeries and attempts to store it into the database.
	// Returns the number of metrics ingested and any error encountered before finishing.
	IngestMetrics(context.Context, *prompb.WriteRequest) (uint64, uint64, error)
	IngestTraces(context.Context, ptrace.Traces) error
	Close()
}

DBInserter is responsible for ingesting the TimeSeries protobuf structs and storing them in the database.

type ExemplarLabelFormatter

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

func NewExamplarLabelFormatter

func NewExamplarLabelFormatter(conn pgxconn.PgxConn, exemplarKeyPosCache cache.PositionCache) *ExemplarLabelFormatter

type ExemplarVisitor

type ExemplarVisitor interface {
	VisitExemplar(func(info *model.MetricInfo, s *model.PromExemplars) error) error
}

type ReadOnlyIngestor

type ReadOnlyIngestor struct{}

func (ReadOnlyIngestor) Close

func (ReadOnlyIngestor) Close()

func (ReadOnlyIngestor) IngestMetrics

func (ReadOnlyIngestor) IngestMetrics(context.Context, *prompb.WriteRequest) (numInsertablesIngested uint64, numMetadataIngested uint64, err error)

func (ReadOnlyIngestor) IngestTraces

func (ReadOnlyIngestor) IngestTraces(context.Context, ptrace.Traces) error

type SeriesVisitor

type SeriesVisitor interface {
	VisitSeries(func(info *pgmodel.MetricInfo, s *model.Series) error) error
}