ingesterkafka

Documentation

Index

• Constants
• Variables
• type Config
  • func (cfg *Config) RegisterFlags(f *flag.FlagSet)
• type Ingester
  • func New(cfg Config, registerer prometheus.Registerer, metricsNamespace string, ...) (*Ingester, error)
  • func (i *Ingester) CheckReady(ctx context.Context) error
  • func (i *Ingester) Flush()
  • func (i *Ingester) PrepareShutdown(w http.ResponseWriter, r *http.Request)
  • func (i *Ingester) Push(_ context.Context, _ *logproto.PushRequest) (*logproto.PushResponse, error)
  • func (i *Ingester) ServeHTTP(w http.ResponseWriter, r *http.Request)
  • func (i *Ingester) ShutdownHandler(w http.ResponseWriter, r *http.Request)
  • func (i *Ingester) TransferOut(_ context.Context) error
  • func (*Ingester) Watch(*grpc_health_v1.HealthCheckRequest, grpc_health_v1.Health_WatchServer) error
• type Interface
• type Storage
• type Wrapper

Constants

const (
	RingName          = "kafka-ingester"
	PartitionRingName = "kafka-partition"
)

Variables

var (
	ErrReadOnly = errors.New("Ingester is shutting down")
)

ErrReadOnly is returned when the ingester is shutting down and a push was attempted.

Types

type Config

type Config struct {
	Enabled bool `yaml:"enabled" doc:"description=Whether the kafka ingester is enabled."`

	LifecyclerConfig   ring.LifecyclerConfig `` /* 145-byte string literal not displayed */
	ShutdownMarkerPath string                `yaml:"shutdown_marker_path"`

	// Used for the kafka ingestion path
	PartitionRingConfig partitionring.Config `yaml:"partition_ring" category:"experimental"`
	KafkaConfig         kafka.Config
}

Config for an ingester.

func (*Config) RegisterFlags

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

RegisterFlags registers the flags.

type Ingester

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

Ingester builds chunks for incoming log streams.

func New

func New(cfg Config,
	registerer prometheus.Registerer,
	metricsNamespace string,
	logger log.Logger,
) (*Ingester, error)

New makes a new Ingester.

func (*Ingester) CheckReady

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

ReadinessHandler is used to indicate to k8s when the ingesters are ready for the addition removal of another ingester. Returns 204 when the ingester is ready, 500 otherwise.

func (*Ingester) Flush

func (i *Ingester) Flush()

Flush implements ring.FlushTransferer 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) PrepareShutdown

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

PrepareShutdown will handle the /ingester/prepare_shutdown endpoint.

Internally, when triggered, this handler will configure the ingester service to release their resources whenever a SIGTERM is received. Releasing resources meaning flushing data, deleting tokens, and removing itself from the ring.

It also creates a file on disk which is used to re-apply the configuration if the ingester crashes and restarts before being permanently shutdown.

* `GET` shows the status of this configuration * `POST` enables this configuration * `DELETE` disables this configuration

func (*Ingester) Push

func (i *Ingester) Push(_ context.Context, _ *logproto.PushRequest) (*logproto.PushResponse, error)

Push implements logproto.Pusher.

func (*Ingester) ServeHTTP

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

ServeHTTP implements the pattern ring status page.

func (*Ingester) ShutdownHandler

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

ShutdownHandler handles a graceful shutdown of the ingester service and termination of the Loki process.

func (*Ingester) TransferOut

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

func (*Ingester) Watch

func (*Ingester) Watch(*grpc_health_v1.HealthCheckRequest, grpc_health_v1.Health_WatchServer) error

Watch implements grpc_health_v1.HealthCheck.

type Interface

type Interface interface {
	services.Service
	http.Handler

	logproto.PusherServer

	CheckReady(ctx context.Context) error
	FlushHandler(w http.ResponseWriter, _ *http.Request)
	ShutdownHandler(w http.ResponseWriter, r *http.Request)
	PrepareShutdown(w http.ResponseWriter, r *http.Request)
}

Interface is an interface for the Ingester

type Storage

type Storage interface {
	PutObject(ctx context.Context, objectKey string, object io.Reader) error
	Stop()
}

Storage is the store interface we need on the ingester.

type Wrapper

type Wrapper interface {
	Wrap(wrapped Interface) Interface
}