worker

Documentation

Index

• func NewQuerierWorker(cfg Config, handler RequestHandler, log log.Logger, reg prometheus.Registerer, ...) (services.Service, error)
• type Config
• type Processor
  • func NewFrontendProcessor(cfg Config, handler RequestHandler, log log.Logger, prcFunc ProcessorFunction) Processor
• type ProcessorFunction
• type ProcessorManager
  • func NewProcessorManager(ctx context.Context, p Processor, conn *grpc.ClientConn, address string) *ProcessorManager
  • func (pm *ProcessorManager) Concurrency(n int)
  • func (pm *ProcessorManager) Stop()
• type RequestHandler

Functions

func NewQuerierWorker

func NewQuerierWorker(cfg Config, handler RequestHandler, log log.Logger, reg prometheus.Registerer, processFunction ProcessorFunction) (services.Service, error)

Types

type Config

type Config struct {
	FrontendAddress string        `yaml:"frontend_address"`
	DNSLookupPeriod time.Duration `yaml:"dns_lookup_duration"`

	Parallelism           int  `yaml:"parallelism"`
	MatchMaxConcurrency   bool `yaml:"match_max_concurrent"`
	MaxConcurrentRequests int  `yaml:"-"`

	QuerierID string `yaml:"id"`

	GRPCClientConfig grpcclient.Config `yaml:"grpc_client_config"`
}

type Processor

type Processor interface {
	// Each invocation of processQueriesOnSingleStream starts new streaming operation to query-frontend
	// or query-scheduler to fetch queries and execute them.
	//
	// This method must react on context being finished, and stop when that happens.
	//
	// ProcessorManager (not Processor) is responsible for starting as many goroutines as needed for each connection.
	ProcessQueriesOnSingleStream(ctx context.Context, conn *grpc.ClientConn, address string)

	// notifyShutdown notifies the remote query-frontend or query-scheduler that the querier is
	// shutting down.
	NotifyShutdown(ctx context.Context, conn *grpc.ClientConn, address string)
}

Single processor handles all streaming operations to query-frontend or query-scheduler to fetch queries and process them.

func NewFrontendProcessor

func NewFrontendProcessor(cfg Config, handler RequestHandler, log log.Logger, prcFunc ProcessorFunction) Processor

type ProcessorFunction

type ProcessorFunction func(client frontendv1pb.FrontendClient, ctx context.Context) (frontendv1pb.Frontend_ProcessClient, error)

type ProcessorManager

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

ProcessorManager Manages processor goroutines for single grpc connection.

func NewProcessorManager

func NewProcessorManager(ctx context.Context, p Processor, conn *grpc.ClientConn, address string) *ProcessorManager

func (*ProcessorManager) Concurrency

func (pm *ProcessorManager) Concurrency(n int)

func (*ProcessorManager) Stop

func (pm *ProcessorManager) Stop()

type RequestHandler

type RequestHandler interface {
	Handle(context.Context, *httpgrpc.HTTPRequest) (*httpgrpc.HTTPResponse, error)
}

Handler for HTTP requests wrapped in protobuf messages.