ingester

package
v1.9.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 14, 2021 License: Apache-2.0 Imports: 71 Imported by: 1

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidLengthWal = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowWal   = fmt.Errorf("proto: integer overflow")
)

Functions

This section is empty.

Types

type ActiveSeries added in v1.5.0 

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

ActiveSeries is keeping track of recently active series for a single tenant.

func NewActiveSeries added in v1.5.0 

func NewActiveSeries() *ActiveSeries

func (*ActiveSeries) Active added in v1.5.0 

func (c *ActiveSeries) Active() int

func (*ActiveSeries) Purge added in v1.5.0 

func (c *ActiveSeries) Purge(keepUntil time.Time)

Purge removes expired entries from the cache. This function should be called periodically to avoid memory leaks.

func (*ActiveSeries) UpdateSeries added in v1.5.0 

func (c *ActiveSeries) UpdateSeries(series labels.Labels, now time.Time, labelsCopy func(labels.Labels) labels.Labels)

Updates series timestamp to 'now'. Function is called to make a copy of labels if entry doesn't exist yet.

type ChunkStore 

type ChunkStore interface {
	Put(ctx context.Context, chunks []cortex_chunk.Chunk) error
}

ChunkStore is the interface we need to store chunks

type Config 

type Config struct {
	WALConfig        WALConfig             `` /* 165-byte string literal not displayed */
	LifecyclerConfig ring.LifecyclerConfig `yaml:"lifecycler"`

	// Config for transferring chunks. Zero or negative = no retries.
	MaxTransferRetries int `yaml:"max_transfer_retries"`

	// Config for chunk flushing.
	FlushCheckPeriod  time.Duration `yaml:"flush_period"`
	RetainPeriod      time.Duration `yaml:"retain_period"`
	MaxChunkIdle      time.Duration `yaml:"max_chunk_idle_time"`
	MaxStaleChunkIdle time.Duration `yaml:"max_stale_chunk_idle_time"`
	FlushOpTimeout    time.Duration `yaml:"flush_op_timeout"`
	MaxChunkAge       time.Duration `yaml:"max_chunk_age"`
	ChunkAgeJitter    time.Duration `yaml:"chunk_age_jitter"`
	ConcurrentFlushes int           `yaml:"concurrent_flushes"`
	SpreadFlushes     bool          `yaml:"spread_flushes"`

	// Config for metadata purging.
	MetadataRetainPeriod time.Duration `yaml:"metadata_retain_period"`

	RateUpdatePeriod time.Duration `yaml:"rate_update_period"`

	ActiveSeriesMetricsEnabled      bool          `yaml:"active_series_metrics_enabled"`
	ActiveSeriesMetricsUpdatePeriod time.Duration `yaml:"active_series_metrics_update_period"`
	ActiveSeriesMetricsIdleTimeout  time.Duration `yaml:"active_series_metrics_idle_timeout"`

	// Use blocks storage.
	BlocksStorageEnabled        bool                     `yaml:"-"`
	BlocksStorageConfig         tsdb.BlocksStorageConfig `yaml:"-"`
	StreamChunksWhenUsingBlocks bool                     `yaml:"-"`
	// Runtime-override for type of streaming query to use (chunks or samples).
	StreamTypeFn func() QueryStreamType `yaml:"-"`

	// Injected at runtime and read from the distributor config, required
	// to accurately apply global limits.
	DistributorShardingStrategy string `yaml:"-"`
	DistributorShardByAllLabels bool   `yaml:"-"`

	DefaultLimits    InstanceLimits         `yaml:"instance_limits"`
	InstanceLimitsFn func() *InstanceLimits `yaml:"-"`
	// contains filtered or unexported fields
}

Config for an Ingester.

func (*Config) RegisterFlags 

func (cfg *Config) RegisterFlags(f *flag.FlagSet)

RegisterFlags adds the flags required to config this to the given FlagSet

type Ingester 

type Ingester struct {
	*services.BasicService

	// Prometheus block storage
	TSDBState TSDBState
	// contains filtered or unexported fields
}

Ingester deals with "in flight" chunks. Based on Prometheus 1.x MemorySeriesStorage.

func New 

func New(cfg Config, clientConfig client.Config, limits *validation.Overrides, chunkStore ChunkStore, registerer prometheus.Registerer, logger log.Logger) (*Ingester, error)

New constructs a new Ingester.

func NewForFlusher added in v1.0.0 

func NewForFlusher(cfg Config, chunkStore ChunkStore, limits *validation.Overrides, registerer prometheus.Registerer, logger log.Logger) (*Ingester, error)

NewForFlusher constructs a new Ingester to be used by flusher target. Compared to the 'New' method:

  • Always replays the WAL.
  • Does not start the lifecycler.

func NewV2 added in v0.4.0 

func NewV2(cfg Config, clientConfig client.Config, limits *validation.Overrides, registerer prometheus.Registerer, logger log.Logger) (*Ingester, error)

NewV2 returns a new Ingester that uses Cortex block storage instead of chunks storage.

func NewV2ForFlusher added in v1.3.0 

func NewV2ForFlusher(cfg Config, limits *validation.Overrides, registerer prometheus.Registerer, logger log.Logger) (*Ingester, error)

NewV2ForFlusher is a special version of ingester used by Flusher. This ingester is not ingesting anything, its only purpose is to react on Flush method and flush all openened TSDBs when called.

func (*Ingester) AllUserStats 

func (i *Ingester) AllUserStats(ctx context.Context, req *client.UserStatsRequest) (*client.UsersStatsResponse, error)

AllUserStats returns ingestion statistics for all users known to this ingester.

func (*Ingester) CheckReady added in v0.7.0 

func (i *Ingester) CheckReady(ctx context.Context) error

CheckReady is the readiness handler used to indicate to k8s when the ingesters are ready for the addition or removal of another ingester.

func (*Ingester) Flush 

func (i *Ingester) Flush()

Flush triggers a flush of all the chunks and closes the flush queues. Called from the Lifecycler as part of the ingester shutdown.

func (*Ingester) FlushHandler 

func (i *Ingester) FlushHandler(w http.ResponseWriter, r *http.Request)

FlushHandler triggers a flush of all in memory chunks. Mainly used for local testing.

func (*Ingester) LabelNames 

func (i *Ingester) LabelNames(ctx context.Context, req *client.LabelNamesRequest) (*client.LabelNamesResponse, error)

LabelNames return all the label names.

func (*Ingester) LabelValues 

func (i *Ingester) LabelValues(ctx context.Context, req *client.LabelValuesRequest) (*client.LabelValuesResponse, error)

LabelValues returns all label values that are associated with a given label name.

func (*Ingester) MetricsForLabelMatchers 

func (i *Ingester) MetricsForLabelMatchers(ctx context.Context, req *client.MetricsForLabelMatchersRequest) (*client.MetricsForLabelMatchersResponse, error)

MetricsForLabelMatchers returns all the metrics which match a set of matchers.

func (*Ingester) MetricsMetadata added in v1.1.0 

func (i *Ingester) MetricsMetadata(ctx context.Context, req *client.MetricsMetadataRequest) (*client.MetricsMetadataResponse, error)

MetricsMetadata returns all the metric metadata of a user.

func (*Ingester) Push 

func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error)

Push implements client.IngesterServer

func (*Ingester) Query 

func (i *Ingester) Query(ctx context.Context, req *client.QueryRequest) (*client.QueryResponse, error)

Query implements service.IngesterServer

func (*Ingester) QueryStream 

func (i *Ingester) QueryStream(req *client.QueryRequest, stream client.Ingester_QueryStreamServer) error

QueryStream implements service.IngesterServer

func (*Ingester) ShutdownHandler added in v0.4.0 

func (i *Ingester) ShutdownHandler(w http.ResponseWriter, r *http.Request)

ShutdownHandler triggers the following set of operations in order:

  • Change the state of ring to stop accepting writes.
  • Flush all the chunks.

func (*Ingester) TransferChunks 

func (i *Ingester) TransferChunks(stream client.Ingester_TransferChunksServer) error

TransferChunks receives all the chunks from another ingester.

func (*Ingester) TransferOut 

func (i *Ingester) TransferOut(ctx context.Context) error

TransferOut finds an ingester in PENDING state and transfers our chunks to it. Called as part of the ingester shutdown process.

func (*Ingester) UserStats 

func (i *Ingester) UserStats(ctx context.Context, req *client.UserStatsRequest) (*client.UserStatsResponse, error)

UserStats returns ingestion statistics for the current user.

type InstanceLimits added in v1.9.0 

type InstanceLimits struct {
	MaxIngestionRate        float64 `yaml:"max_ingestion_rate"`
	MaxInMemoryTenants      int64   `yaml:"max_tenants"`
	MaxInMemorySeries       int64   `yaml:"max_series"`
	MaxInflightPushRequests int64   `yaml:"max_inflight_push_requests"`
}

InstanceLimits describes limits used by ingester. Reaching any of these will result in Push method to return (internal) error.

func (*InstanceLimits) UnmarshalYAML added in v1.9.0 

func (l *InstanceLimits) UnmarshalYAML(unmarshal func(interface{}) error) error

UnmarshalYAML implements the yaml.Unmarshaler interface. If give

type Limiter added in v1.1.0 

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

Limiter implements primitives to get the maximum number of series an ingester can handle for a specific tenant

func NewLimiter added in v1.1.0 

func NewLimiter(
	limits *validation.Overrides,
	ring RingCount,
	shardingStrategy string,
	shardByAllLabels bool,
	replicationFactor int,
	zoneAwarenessEnabled bool,
) *Limiter

NewLimiter makes a new in-memory series limiter

func (*Limiter) AssertMaxMetadataPerMetric added in v1.1.0 

func (l *Limiter) AssertMaxMetadataPerMetric(userID string, metadata int) error

AssertMaxMetadataPerMetric limit has not been reached compared to the current number of metadata per metric in input and returns an error if so.

func (*Limiter) AssertMaxMetricsWithMetadataPerUser added in v1.1.0 

func (l *Limiter) AssertMaxMetricsWithMetadataPerUser(userID string, metrics int) error

AssertMaxMetricsWithMetadataPerUser limit has not been reached compared to the current number of metrics with metadata in input and returns an error if so.

func (*Limiter) AssertMaxSeriesPerMetric added in v1.1.0 

func (l *Limiter) AssertMaxSeriesPerMetric(userID string, series int) error

AssertMaxSeriesPerMetric limit has not been reached compared to the current number of series in input and returns an error if so.

func (*Limiter) AssertMaxSeriesPerUser added in v1.1.0 

func (l *Limiter) AssertMaxSeriesPerUser(userID string, series int) error

AssertMaxSeriesPerUser limit has not been reached compared to the current number of series in input and returns an error if so.

func (*Limiter) FormatError added in v1.9.0 

func (l *Limiter) FormatError(userID string, err error) error

FormatError returns the input error enriched with the actual limits for the given user. It acts as pass-through if the input error is unknown.

func (*Limiter) MaxSeriesPerQuery added in v1.1.0 

func (l *Limiter) MaxSeriesPerQuery(userID string) int

MaxSeriesPerQuery returns the maximum number of series a query is allowed to hit.

type QueryStreamType added in v1.8.0 

type QueryStreamType int

QueryStreamType defines type of function to use when doing query-stream operation. 

const (
	QueryStreamDefault QueryStreamType = iota // Use default configured value.
	QueryStreamSamples                        // Stream individual samples.
	QueryStreamChunks                         // Stream entire chunks.
)

type RecordType added in v1.1.0 

type RecordType byte

RecordType represents the type of the WAL/Checkpoint record. 

const (
	// WALRecordSeries is the type for the WAL record on Prometheus TSDB record for series.
	WALRecordSeries RecordType = 1
	// WALRecordSamples is the type for the WAL record based on Prometheus TSDB record for samples.
	WALRecordSamples RecordType = 2

	// CheckpointRecord is the type for the Checkpoint record based on protos.
	CheckpointRecord RecordType = 3
)

type RingCount added in v0.4.0 

type RingCount interface {
	HealthyInstancesCount() int
	ZonesCount() int
}

RingCount is the interface exposed by a ring implementation which allows to count members

type Series added in v0.6.0 

type Series struct {
	UserId      string                                                      `protobuf:"bytes,1,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"`
	Fingerprint uint64                                                      `protobuf:"varint,2,opt,name=fingerprint,proto3" json:"fingerprint,omitempty"`
	Labels      []github_com_cortexproject_cortex_pkg_cortexpb.LabelAdapter `protobuf:"bytes,3,rep,name=labels,proto3,customtype=github.com/cortexproject/cortex/pkg/cortexpb.LabelAdapter" json:"labels"`
	Chunks      []client.Chunk                                              `protobuf:"bytes,4,rep,name=chunks,proto3" json:"chunks"`
}

func (*Series) Descriptor added in v0.6.0 

func (*Series) Descriptor() ([]byte, []int)

func (*Series) Equal added in v0.6.0 

func (this *Series) Equal(that interface{}) bool

func (*Series) GetChunks added in v0.6.0 

func (m *Series) GetChunks() []client.Chunk

func (*Series) GetFingerprint added in v0.6.0 

func (m *Series) GetFingerprint() uint64

func (*Series) GetUserId added in v0.6.0 

func (m *Series) GetUserId() string

func (*Series) GoString added in v0.6.0 

func (this *Series) GoString() string

func (*Series) Marshal added in v0.6.0 

func (m *Series) Marshal() (dAtA []byte, err error)

func (*Series) MarshalTo added in v0.6.0 

func (m *Series) MarshalTo(dAtA []byte) (int, error)

func (*Series) MarshalToSizedBuffer added in v0.7.0 

func (m *Series) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Series) ProtoMessage added in v0.6.0 

func (*Series) ProtoMessage()

func (*Series) Reset added in v0.6.0 

func (m *Series) Reset()

func (*Series) Size added in v0.6.0 

func (m *Series) Size() (n int)

func (*Series) String added in v0.6.0 

func (this *Series) String() string

func (*Series) Unmarshal added in v0.6.0 

func (m *Series) Unmarshal(dAtA []byte) error

func (*Series) XXX_DiscardUnknown added in v0.6.0 

func (m *Series) XXX_DiscardUnknown()

func (*Series) XXX_Marshal added in v0.6.0 

func (m *Series) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Series) XXX_Merge added in v0.6.0 

func (m *Series) XXX_Merge(src proto.Message)

func (*Series) XXX_Size added in v0.6.0 

func (m *Series) XXX_Size() int

func (*Series) XXX_Unmarshal added in v0.6.0 

func (m *Series) XXX_Unmarshal(b []byte) error

type Shipper added in v0.7.0 

type Shipper interface {
	Sync(ctx context.Context) (uploaded int, err error)
}

Shipper interface is used to have an easy way to mock it in tests.

type TSDBState added in v0.4.0 

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

TSDBState holds data structures used by the TSDB storage engine

type WAL added in v0.6.0 

type WAL interface {
	// Log marshalls the records and writes it into the WAL.
	Log(*WALRecord) error
	// Stop stops all the WAL operations.
	Stop()
}

WAL interface allows us to have a no-op WAL when the WAL is disabled.

type WALConfig added in v0.6.0 

type WALConfig struct {
	WALEnabled         bool          `yaml:"wal_enabled"`
	CheckpointEnabled  bool          `yaml:"checkpoint_enabled"`
	Recover            bool          `yaml:"recover_from_wal"`
	Dir                string        `yaml:"wal_dir"`
	CheckpointDuration time.Duration `yaml:"checkpoint_duration"`
	FlushOnShutdown    bool          `yaml:"flush_on_shutdown_with_wal_enabled"`
	// contains filtered or unexported fields
}

WALConfig is config for the Write Ahead Log.

func (*WALConfig) RegisterFlags added in v0.6.0 

func (cfg *WALConfig) RegisterFlags(f *flag.FlagSet)

RegisterFlags adds the flags required to config this to the given FlagSet

type WALRecord added in v1.1.0 

type WALRecord struct {
	UserID  string
	Series  []tsdb_record.RefSeries
	Samples []tsdb_record.RefSample
}

WALRecord is a struct combining the series and samples record.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.
JackTT - Gopher 🇻🇳