distributor

Documentation

Index

• type Config
  • func (cfg *Config) RegisterFlags(fs *flag.FlagSet)
• type Distributor
  • func New(cfg Config, clientCfg client.Config, configs *runtime.TenantConfigs, ...) (*Distributor, error)
  • func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*logproto.PushResponse, error)
  • func (d *Distributor) PushHandler(w http.ResponseWriter, r *http.Request)
  • func (d *Distributor) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type Limits
• type RateStore
• type ReadLifecycler
• type RingConfig
  • func (cfg *RingConfig) RegisterFlags(f *flag.FlagSet)
  • func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig
  • func (cfg *RingConfig) ToRingConfig() ring.Config
• type Validator
  • func NewValidator(l Limits) (*Validator, error)
  • func (v Validator) ValidateEntry(ctx validationContext, labels string, entry logproto.Entry) error
  • func (v Validator) ValidateLabels(ctx validationContext, ls labels.Labels, stream logproto.Stream) error

Types

type Config

type Config struct {
	// Distributors ring
	DistributorRing RingConfig `yaml:"ring,omitempty"`
	// contains filtered or unexported fields
}

Config for a Distributor.

func (*Config) RegisterFlags

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

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,
	clientCfg client.Config,
	configs *runtime.TenantConfigs,
	ingestersRing ring.ReadRing,
	overrides *validation.Overrides,
	registerer prometheus.Registerer,
) (*Distributor, error)

New a distributor creates.

func (*Distributor) Push

func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*logproto.PushResponse, error)

Push a set of streams. The returned error is the last one seen.

func (*Distributor) PushHandler

func (d *Distributor) PushHandler(w http.ResponseWriter, r *http.Request)

PushHandler reads a snappy-compressed proto from the HTTP body.

func (*Distributor) ServeHTTP

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

ServeHTTP implements the distributor ring status page.

If the rate limiting strategy is local instead of global, no ring is used by the distributor and as such, no ring status is returned from this function.

type Limits

type Limits interface {
	MaxLineSize(userID string) int
	MaxLineSizeTruncate(userID string) bool
	EnforceMetricName(userID string) bool
	MaxLabelNamesPerSeries(userID string) int
	MaxLabelNameLength(userID string) int
	MaxLabelValueLength(userID string) int

	CreationGracePeriod(userID string) time.Duration
	RejectOldSamples(userID string) bool
	RejectOldSamplesMaxAge(userID string) time.Duration

	IncrementDuplicateTimestamps(userID string) bool

	ShardStreams(userID string) *shardstreams.Config
}

Limits is an interface for distributor limits/related configs

type RateStore

type RateStore interface {
	RateFor(stream *logproto.Stream) (int, error)
}

RateStore manages the ingestion rate of streams, populated by data fetched from ingesters.

type ReadLifecycler

type ReadLifecycler interface {
	HealthyInstancesCount() int
}

ReadLifecycler represents the read interface to the lifecycler.

type RingConfig

type RingConfig struct {
	KVStore          kv.Config     `yaml:"kvstore"`
	HeartbeatPeriod  time.Duration `yaml:"heartbeat_period"`
	HeartbeatTimeout time.Duration `yaml:"heartbeat_timeout"`

	// Instance details
	InstanceID             string   `yaml:"instance_id" doc:"hidden"`
	InstanceInterfaceNames []string `yaml:"instance_interface_names" doc:"default=[<private network interfaces>]"`
	InstancePort           int      `yaml:"instance_port" doc:"hidden"`
	InstanceAddr           string   `yaml:"instance_addr" doc:"hidden"`

	// Injected internally
	ListenPort int `yaml:"-"`
}

RingConfig masks the ring lifecycler config which contains many options not really required by the distributors ring. This config is used to strip down the config to the minimum, and avoid confusion to the user.

func (*RingConfig) RegisterFlags

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

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

func (*RingConfig) ToLifecyclerConfig

func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig

ToLifecyclerConfig returns a LifecyclerConfig based on the distributor ring config.

func (*RingConfig) ToRingConfig

func (cfg *RingConfig) ToRingConfig() ring.Config

type Validator

type Validator struct {
	Limits
}

func NewValidator

func NewValidator(l Limits) (*Validator, error)

func (Validator) ValidateEntry

func (v Validator) ValidateEntry(ctx validationContext, labels string, entry logproto.Entry) error

ValidateEntry returns an error if the entry is invalid and report metrics for invalid entries accordingly.

func (Validator) ValidateLabels

func (v Validator) ValidateLabels(ctx validationContext, ls labels.Labels, stream logproto.Stream) error

Validate labels returns an error if the labels are invalid