distributor

Documentation

Index

• Constants
• func TokenFor(tenantID, labels string) uint32
• type Config
  • func (cfg *Config) RegisterFlags(fs *flag.FlagSet, logger log.Logger)
• type Distributor
  • func New(cfg Config, ingestersRing ring.ReadRing, factory ring_client.PoolFactory, ...) (*Distributor, error)
  • func (d *Distributor) HealthyInstancesCount() int
  • func (d *Distributor) Push(ctx context.Context, grpcReq *connect.Request[pushv1.PushRequest]) (*connect.Response[pushv1.PushResponse], error)
  • func (d *Distributor) PushParsed(ctx context.Context, req *distributormodel.PushRequest) (*connect.Response[pushv1.PushResponse], error)
  • func (d *Distributor) ServeHTTP(w http.ResponseWriter, req *http.Request)
• type Limits
• type PushClient
• type ReadLifecycler

Constants

const (
	ProfileName = "__name__"
)

Functions

func TokenFor

func TokenFor(tenantID, labels string) uint32

TokenFor generates a token used for finding ingesters from ring

Types

type Config

type Config struct {
	PushTimeout time.Duration
	PoolConfig  clientpool.PoolConfig `yaml:"pool_config,omitempty"`

	// Distributors ring
	DistributorRing util.CommonRingConfig `yaml:"ring" doc:"hidden"`
}

Config for a Distributor.

func (*Config) RegisterFlags

func (cfg *Config) RegisterFlags(fs *flag.FlagSet, logger log.Logger)

RegisterFlags registers distributor-related flags.

type Distributor

type Distributor struct {
	services.Service
	// contains filtered or unexported fields
}

Distributor coordinates replicates and distribution of log streams.

func New

func New(cfg Config, ingestersRing ring.ReadRing, factory ring_client.PoolFactory, limits Limits, reg prometheus.Registerer, logger log.Logger, clientsOptions ...connect.ClientOption) (*Distributor, error)

func (*Distributor) HealthyInstancesCount

func (d *Distributor) HealthyInstancesCount() int

HealthyInstancesCount implements the ReadLifecycler interface

We use a ring lifecycler delegate to count the number of members of the ring. The count is then used to enforce rate limiting correctly for each distributor. $EFFECTIVE_RATE_LIMIT = $GLOBAL_RATE_LIMIT / $NUM_INSTANCES

func (*Distributor) Push

func (d *Distributor) Push(ctx context.Context, grpcReq *connect.Request[pushv1.PushRequest]) (*connect.Response[pushv1.PushResponse], error)

func (*Distributor) PushParsed

func (d *Distributor) PushParsed(ctx context.Context, req *distributormodel.PushRequest) (*connect.Response[pushv1.PushResponse], error)

func (*Distributor) ServeHTTP

func (d *Distributor) ServeHTTP(w http.ResponseWriter, req *http.Request)

type Limits

type Limits interface {
	IngestionRateBytes(tenantID string) float64
	IngestionBurstSizeBytes(tenantID string) int
	IngestionTenantShardSize(tenantID string) int
	MaxLabelNameLength(tenantID string) int
	MaxLabelValueLength(tenantID string) int
	MaxLabelNamesPerSeries(tenantID string) int
	MaxProfileSizeBytes(tenantID string) int
	MaxProfileStacktraceSamples(tenantID string) int
	MaxProfileStacktraceSampleLabels(tenantID string) int
	MaxProfileStacktraceDepth(tenantID string) int
	MaxProfileSymbolValueLength(tenantID string) int
	MaxSessionsPerSeries(tenantID string) int
	validation.ProfileValidationLimits
}

type PushClient

type PushClient interface {
	Push(context.Context, *connect.Request[pushv1.PushRequest]) (*connect.Response[pushv1.PushResponse], error)
}

type ReadLifecycler

type ReadLifecycler interface {
	HealthyInstancesCount() int
}

ReadLifecycler represents the read interface to the lifecycler.