distributor

type Config struct {
	// Distributors ring
	DistributorRing RingConfig `yaml:"ring,omitempty"`
	PushWorkerCount int        `yaml:"push_worker_count"`

	// Request parser
	MaxRecvMsgSize      int   `yaml:"max_recv_msg_size"`
	MaxDecompressedSize int64 `yaml:"max_decompressed_size"`

	// RateStore customizes the rate storing used by stream sharding.
	RateStore RateStoreConfig `yaml:"rate_store"`

	// WriteFailuresLoggingCfg customizes write failures logging behavior.
	WriteFailuresLogging writefailures.Cfg `yaml:"write_failures_logging" doc:"description=Customize the logging of write failures."`

	OTLPConfig push.GlobalOTLPConfig `yaml:"otlp_config"`

	// DefaultPolicyStreamMappings contains the default policy stream mappings that are merged with per-tenant mappings.
	DefaultPolicyStreamMappings validation.PolicyStreamMapping `` /* 128-byte string literal not displayed */

	KafkaEnabled              bool `yaml:"kafka_writes_enabled"`
	IngesterEnabled           bool `yaml:"ingester_writes_enabled"`
	IngestLimitsEnabled       bool `yaml:"ingest_limits_enabled"`
	IngestLimitsDryRunEnabled bool `yaml:"ingest_limits_dry_run_enabled"`

	KafkaConfig kafka.Config `yaml:"-"`

	DataObjTeeConfig DataObjTeeConfig `yaml:"dataobj_tee"`
	// contains filtered or unexported fields
}

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

type DataObjTeeConfig struct {
	Enabled               bool          `yaml:"enabled"`
	Topic                 string        `yaml:"topic"`
	MaxBufferedBytes      int           `yaml:"max_buffered_bytes"`
	PerPartitionRateBytes int           `yaml:"per_partition_rate_bytes"`
	DebugMetricsEnabled   bool          `yaml:"debug_metrics_enabled"`
	RateBatchWindow       time.Duration `yaml:"rate_batch_window"`
}

type Distributor struct {
	services.Service

	RequestParserWrapper push.RequestParserWrapper
	// contains filtered or unexported fields
}

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

type KafkaProducer interface {
	ProduceSync(ctx context.Context, records []*kgo.Record) kgo.ProduceResults
	Close()
}

type KeyedStream struct {
	HashKey        uint32
	HashKeyNoShard uint64
	Stream         logproto.Stream
	Policy         string
}

type Limits interface {
	retention.Limits
	MaxLineSize(userID string) int
	MaxLineSizeTruncate(userID string) bool
	MaxLineSizeTruncateIdentifier(userID string) string

	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
	DiscoverServiceName(userID string) []string
	DiscoverGenericFields(userID string) map[string][]string
	DiscoverLogLevels(userID string) bool
	LogLevelFields(userID string) []string
	LogLevelFromJSONMaxDepth(userID string) int

	ShardStreams(userID string) shardstreams.Config
	IngestionRateStrategy() string
	IngestionRateBytes(userID string) float64
	IngestionBurstSizeBytes(userID string) int
	AllowStructuredMetadata(userID string) bool
	MaxStructuredMetadataSize(userID string) int
	MaxStructuredMetadataCount(userID string) int
	OTLPConfig(userID string) push.OTLPConfig

	BlockIngestionUntil(userID string) time.Time
	BlockIngestionStatusCode(userID string) int
	BlockIngestionPolicyUntil(userID string, policy string) time.Time
	EnforcedLabels(userID string) []string
	PolicyEnforcedLabels(userID string, policy string) []string

	IngestionPartitionsTenantShardSize(userID string) int

	SimulatedPushLatency(userID string) time.Duration
}

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

type RateBatcherConfig struct {
	// BatchWindow is the duration to accumulate rate updates before flushing.
	BatchWindow time.Duration
}

type RateStore interface {
	RateFor(tenantID string, streamHash uint64) (int64, float64)
}

type RateStoreConfig struct {
	MaxParallelism           int           `yaml:"max_request_parallelism"`
	StreamRateUpdateInterval time.Duration `yaml:"stream_rate_update_interval"`
	IngesterReqTimeout       time.Duration `yaml:"ingester_request_timeout"`
	Debug                    bool          `yaml:"debug"`
}

type ReadLifecycler interface {
	HealthyInstancesCount() int
}

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"`
	EnableIPv6             bool     `yaml:"instance_enable_ipv6" doc:"hidden"`

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

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

type Tee interface {
	Duplicate(ctx context.Context, tenant string, streams []KeyedStream, pushTracker *PushTracker)

	// Register is a prehook to allow Tee's to register its pending streams, allowing distributors to wait for them before concluding a push request.
	// If pending streams are registered, one should make sure `pushTracker.doneWithResult` is invoked for the same number of streams added.
	Register(ctx context.Context, tenant string, streams []KeyedStream, pushTracker *PushTracker)
}

type TeeErrorCodes int

type Validator struct {
	Limits
	// contains filtered or unexported fields
}