writepath

Documentation

Index

• Constants
• Variables
• type Config
  • func (o *Config) RegisterFlags(f *flag.FlagSet)
• type IngesterClient
• type IngesterFunc
  • func (f IngesterFunc) Push(ctx context.Context, req *distributormodel.PushRequest) (*connect.Response[pushv1.PushResponse], error)
• type Overrides
• type Router
  • func NewRouter(logger log.Logger, registerer prometheus.Registerer, overrides Overrides, ...) *Router
  • func (m *Router) Send(ctx context.Context, req *distributormodel.PushRequest) error
  • func (m *Router) Service() services.Service
• type SegmentWriterClient
• type WritePath
  • func (m *WritePath) Set(text string) error
  • func (m *WritePath) String() string

Constants

const (
	// IngesterPath specifies old write path the requests are sent to ingester.
	IngesterPath WritePath = "ingester"
	// SegmentWriterPath specifies the new write path: distributor sends
	// the request to segment writers before profile split, using the new
	// distribution algorithm and the segment-writer ring.
	SegmentWriterPath = "segment-writer"
	// CombinedPath specifies that each request should be sent to both write
	// paths. For each request we decide on how a failure is handled:
	//  * If the request is sent to ingester (regardless of anything),
	//    the response is returned to the client immediately after the old
	//    write path returns. Failure of the new write path should be logged
	//    and counted in metrics but NOT returned to the client.
	//  * If the request is sent to segment-writer exclusively: the response
	//    returns to the client only when the new write path returns.
	//    Failure of the new write is returned to the client.
	//    Failure of the old write path is NOT returned to the client.
	CombinedPath = "combined"
)

Variables

var ErrInvalidWritePath = errors.New("invalid write path")

Types

type Config

type Config struct {
	WritePath           WritePath `yaml:"write_path" json:"write_path" doc:"hidden"`
	IngesterWeight      float64   `yaml:"write_path_ingester_weight" json:"write_path_ingester_weight" doc:"hidden"`
	SegmentWriterWeight float64   `yaml:"write_path_segment_writer_weight" json:"write_path_segment_writer_weight" doc:"hidden"`
}

func (*Config) RegisterFlags

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

type IngesterClient

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

type IngesterFunc

type IngesterFunc func(
	context.Context,
	*distributormodel.PushRequest,
) (*connect.Response[pushv1.PushResponse], error)

func (IngesterFunc) Push

func (f IngesterFunc) Push(
	ctx context.Context,
	req *distributormodel.PushRequest,
) (*connect.Response[pushv1.PushResponse], error)

type Overrides

type Overrides interface {
	WritePathOverrides(tenantID string) Config
}

type Router

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

func NewRouter

func NewRouter(
	logger log.Logger,
	registerer prometheus.Registerer,
	overrides Overrides,
	ingester IngesterClient,
	segwriter SegmentWriterClient,
) *Router

func (*Router) Send

func (m *Router) Send(ctx context.Context, req *distributormodel.PushRequest) error

func (*Router) Service

func (m *Router) Service() services.Service

type SegmentWriterClient

type SegmentWriterClient interface {
	Push(context.Context, *segmentwriterv1.PushRequest) (*segmentwriterv1.PushResponse, error)
}

type WritePath

type WritePath string

WritePath controls the write path.

func (*WritePath) Set

func (m *WritePath) Set(text string) error

func (*WritePath) String

func (m *WritePath) String() string