remote

package
v0.49.5 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 12, 2024 License: Apache-2.0 Imports: 57 Imported by: 0

Documentation

Index

Constants

View Source 
const DefaultChunkedReadLimit = 5e+7

DefaultChunkedReadLimit is the default value for the maximum size of the protobuf frame client allows. 50MB is the default. This is equivalent to ~100k full XOR chunks and average labelset.

Variables

View Source 
var UserAgent = fmt.Sprintf("Prometheus/%s", version.Version)

Functions

func DecodeOTLPWriteRequest 

func DecodeOTLPWriteRequest(r *http.Request) (pmetricotlp.ExportRequest, error)

func DecodeReadRequest 

func DecodeReadRequest(r *http.Request) (*prompb.ReadRequest, error)

DecodeReadRequest reads a remote.Request from a http.Request.

func DecodeWriteRequest 

func DecodeWriteRequest(r io.Reader) (*prompb.WriteRequest, error)

DecodeWriteRequest from an io.Reader into a prompb.WriteRequest, handling snappy decompression.

func EncodeReadResponse 

func EncodeReadResponse(resp *prompb.ReadResponse, w http.ResponseWriter) error

EncodeReadResponse writes a remote.Response to a http.ResponseWriter.

func FloatHistogramProtoToFloatHistogram 

func FloatHistogramProtoToFloatHistogram(hp prompb.Histogram) *histogram.FloatHistogram

FloatHistogramProtoToFloatHistogram extracts a float Histogram from the provided proto message to a Float Histogram. The caller has to make sure that the proto message represents a float histogram and not an integer histogram, or it panics.

func FloatHistogramToHistogramProto 

func FloatHistogramToHistogramProto(timestamp int64, fh *histogram.FloatHistogram) prompb.Histogram

func FromLabelMatchers 

func FromLabelMatchers(matchers []*prompb.LabelMatcher) ([]*labels.Matcher, error)

FromLabelMatchers parses protobuf label matchers to Prometheus label matchers.

func FromQueryResult 

func FromQueryResult(sortSeries bool, res *prompb.QueryResult) storage.SeriesSet

FromQueryResult unpacks and sorts a QueryResult proto.

func HistogramProtoToFloatHistogram 

func HistogramProtoToFloatHistogram(hp prompb.Histogram) *histogram.FloatHistogram

HistogramProtoToFloatHistogram extracts and converts a (normal integer) histogram from the provided proto message to a float histogram. The caller has to make sure that the proto message represents an integer histogram and not a float histogram, or it panics.

func HistogramProtoToHistogram 

func HistogramProtoToHistogram(hp prompb.Histogram) *histogram.Histogram

HistogramProtoToHistogram extracts a (normal integer) Histogram from the provided proto message. The caller has to make sure that the proto message represents an integer histogram and not a float histogram, or it panics.

func HistogramToHistogramProto 

func HistogramToHistogramProto(timestamp int64, h *histogram.Histogram) prompb.Histogram

func LabelProtosToMetric 

func LabelProtosToMetric(labelPairs []*prompb.Label) model.Metric

LabelProtosToMetric unpack a []*prompb.Label to a model.Metric.

func MergeLabels 

func MergeLabels(primary, secondary []prompb.Label) []prompb.Label

MergeLabels merges two sets of sorted proto labels, preferring those in primary to those in secondary when there is an overlap.

func NegotiateResponseType 

func NegotiateResponseType(accepted []prompb.ReadRequest_ResponseType) (prompb.ReadRequest_ResponseType, error)

NegotiateResponseType returns first accepted response type that this server supports. On the empty accepted list we assume that the SAMPLES response type was requested. This is to maintain backward compatibility.

func NewOTLPWriteHandler 

func NewOTLPWriteHandler(logger log.Logger, appendable storage.Appendable) http.Handler

NewOTLPWriteHandler creates a http.Handler that accepts OTLP write requests and writes them to the provided appendable.

func NewReadHandler 

func NewReadHandler(logger log.Logger, r prometheus.Registerer, queryable storage.SampleAndChunkQueryable, config func() config.Config, remoteReadSampleLimit, remoteReadConcurrencyLimit, remoteReadMaxBytesInFrame int) http.Handler

NewReadHandler creates a http.Handler that accepts remote read requests and writes them to the provided queryable.

func NewSampleAndChunkQueryableClient 

func NewSampleAndChunkQueryableClient(
	c ReadClient,
	externalLabels labels.Labels,
	requiredMatchers []*labels.Matcher,
	readRecent bool,
	callback startTimeCallback,
) storage.SampleAndChunkQueryable

NewSampleAndChunkQueryableClient returns a storage.SampleAndChunkQueryable which queries the given client to select series sets.

func NewWriteHandler 

func NewWriteHandler(logger log.Logger, reg prometheus.Registerer, appendable storage.Appendable) http.Handler

NewWriteHandler creates a http.Handler that accepts remote write requests and writes them to the provided appendable.

func StreamChunkedReadResponses 

func StreamChunkedReadResponses(
	stream io.Writer,
	queryIndex int64,
	ss storage.ChunkSeriesSet,
	sortedExternalLabels []prompb.Label,
	maxBytesInFrame int,
	marshalPool *sync.Pool,
) (annotations.Annotations, error)

StreamChunkedReadResponses iterates over series, builds chunks and streams those to the caller. It expects Series set with populated chunks.

func ToQuery 

func ToQuery(from, to int64, matchers []*labels.Matcher, hints *storage.SelectHints) (*prompb.Query, error)

ToQuery builds a Query proto.

func ToQueryResult 

func ToQueryResult(ss storage.SeriesSet, sampleLimit int) (*prompb.QueryResult, annotations.Annotations, error)

ToQueryResult builds a QueryResult proto.

Types

type ChunkedReader 

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

ChunkedReader is a buffered reader that expects uvarint delimiter and checksum before each message. It will allocate as much as the biggest frame defined by delimiter (on top of bufio.Reader allocations).

func NewChunkedReader 

func NewChunkedReader(r io.Reader, sizeLimit uint64, data []byte) *ChunkedReader

NewChunkedReader constructs a ChunkedReader. It allows passing data slice for byte slice reuse, which will be increased to needed size if smaller.

func (*ChunkedReader) Next 

func (r *ChunkedReader) Next() ([]byte, error)

Next returns the next length-delimited record from the input, or io.EOF if there are no more records available. Returns io.ErrUnexpectedEOF if a short record is found, with a length of n but fewer than n bytes of data. Next also verifies the given checksum with Castagnoli polynomial CRC-32 checksum.

NOTE: The slice returned is valid only until a subsequent call to Next. It's a caller's responsibility to copy the returned slice if needed.

func (*ChunkedReader) NextProto 

func (r *ChunkedReader) NextProto(pb proto.Message) error

NextProto consumes the next available record by calling r.Next, and decodes it into the protobuf with proto.Unmarshal.

type ChunkedWriter 

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

ChunkedWriter is an io.Writer wrapper that allows streaming by adding uvarint delimiter before each write in a form of length of the corresponded byte array.

func NewChunkedWriter 

func NewChunkedWriter(w io.Writer, f http.Flusher) *ChunkedWriter

NewChunkedWriter constructs a ChunkedWriter.

func (*ChunkedWriter) Write 

func (w *ChunkedWriter) Write(b []byte) (int, error)

Write writes given bytes to the stream and flushes it. Each frame includes:

1. uvarint for the size of the data frame. 2. big-endian uint32 for the Castagnoli polynomial CRC-32 checksum of the data frame. 3. the bytes of the given data.

Write returns number of sent bytes for a given buffer. The number does not include delimiter and checksum bytes.

type Client 

type Client struct {
	Client *http.Client
	// contains filtered or unexported fields
}

Client allows reading and writing from/to a remote HTTP endpoint.

func (Client) Endpoint 

func (c Client) Endpoint() string

Endpoint is the remote read or write endpoint.

func (Client) Name 

func (c Client) Name() string

Name uniquely identifies the client.

func (*Client) Read 

func (c *Client) Read(ctx context.Context, query *prompb.Query) (*prompb.QueryResult, error)

Read reads from a remote endpoint.

func (*Client) Store 

func (c *Client) Store(ctx context.Context, req []byte, attempt int) error

Store sends a batch of samples to the HTTP endpoint, the request is the proto marshalled and encoded bytes from codec.go.

type ClientConfig 

type ClientConfig struct {
	URL              *config_util.URL
	Timeout          model.Duration
	HTTPClientConfig config_util.HTTPClientConfig
	SigV4Config      *sigv4.SigV4Config
	AzureADConfig    *azuread.AzureADConfig
	Headers          map[string]string
	RetryOnRateLimit bool
}

ClientConfig configures a client.

type HTTPError 

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

func (HTTPError) Error 

func (e HTTPError) Error() string

func (HTTPError) Status 

func (e HTTPError) Status() int

type MetadataAppender 

type MetadataAppender interface {
	AppendMetadata(context.Context, []scrape.MetricMetadata)
}

MetadataAppender is an interface used by the Metadata Watcher to send metadata, It is read from the scrape manager, on to somewhere else.

type MetadataWatcher 

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

MetadataWatcher watches the Scrape Manager for a given WriteMetadataTo.

func NewMetadataWatcher 

func NewMetadataWatcher(l log.Logger, mg ReadyScrapeManager, name string, w MetadataAppender, interval model.Duration, deadline time.Duration) *MetadataWatcher

NewMetadataWatcher builds a new MetadataWatcher.

func (*MetadataWatcher) Start 

func (mw *MetadataWatcher) Start()

Start the MetadataWatcher.

func (*MetadataWatcher) Stop 

func (mw *MetadataWatcher) Stop()

Stop the MetadataWatcher.

type QueueManager 

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

QueueManager manages a queue of samples to be sent to the Storage indicated by the provided WriteClient. Implements writeTo interface used by WAL Watcher.

func NewQueueManager 

func NewQueueManager(
	metrics *queueManagerMetrics,
	watcherMetrics *wlog.WatcherMetrics,
	readerMetrics *wlog.LiveReaderMetrics,
	logger log.Logger,
	dir string,
	samplesIn *ewmaRate,
	cfg config.QueueConfig,
	mCfg config.MetadataConfig,
	externalLabels labels.Labels,
	relabelConfigs []*relabel.Config,
	client WriteClient,
	flushDeadline time.Duration,
	interner *pool,
	highestRecvTimestamp *maxTimestamp,
	sm ReadyScrapeManager,
	enableExemplarRemoteWrite bool,
	enableNativeHistogramRemoteWrite bool,
) *QueueManager

NewQueueManager builds a new QueueManager and starts a new WAL watcher with queue manager as the WriteTo destination. The WAL watcher takes the dir parameter as the base directory for where the WAL shall be located. Note that the full path to the WAL directory will be constructed as <dir>/wal.

func (*QueueManager) Append 

func (t *QueueManager) Append(samples []record.RefSample) bool

Append queues a sample to be sent to the remote storage. Blocks until all samples are enqueued on their shards or a shutdown signal is received.

func (*QueueManager) AppendExemplars 

func (t *QueueManager) AppendExemplars(exemplars []record.RefExemplar) bool

func (*QueueManager) AppendFloatHistograms 

func (t *QueueManager) AppendFloatHistograms(floatHistograms []record.RefFloatHistogramSample) bool

func (*QueueManager) AppendHistograms 

func (t *QueueManager) AppendHistograms(histograms []record.RefHistogramSample) bool

func (*QueueManager) AppendMetadata 

func (t *QueueManager) AppendMetadata(ctx context.Context, metadata []scrape.MetricMetadata)

AppendMetadata sends metadata to the remote storage. Metadata is sent in batches, but is not parallelized.

func (*QueueManager) SeriesReset 

func (t *QueueManager) SeriesReset(index int)

SeriesReset is used when reading a checkpoint. WAL Watcher should have stored series records with the checkpoints index number, so we can now delete any ref ID's lower than that # from the two maps.

func (*QueueManager) SetClient 

func (t *QueueManager) SetClient(c WriteClient)

SetClient updates the client used by a queue. Used when only client specific fields are updated to avoid restarting the queue.

func (*QueueManager) Start 

func (t *QueueManager) Start()

Start the queue manager sending samples to the remote storage. Does not block.

func (*QueueManager) Stop 

func (t *QueueManager) Stop()

Stop stops sending samples to the remote storage and waits for pending sends to complete.

func (*QueueManager) StoreSeries 

func (t *QueueManager) StoreSeries(series []record.RefSeries, index int)

StoreSeries keeps track of which series we know about for lookups when sending samples to remote.

func (*QueueManager) UpdateSeriesSegment 

func (t *QueueManager) UpdateSeriesSegment(series []record.RefSeries, index int)

UpdateSeriesSegment updates the segment number held against the series, so we can trim older ones in SeriesReset.

type ReadClient 

type ReadClient interface {
	Read(ctx context.Context, query *prompb.Query) (*prompb.QueryResult, error)
}

ReadClient uses the SAMPLES method of remote read to read series samples from remote server. TODO(bwplotka): Add streamed chunked remote read method as well (https://github.com/go-follow/prometheus/issues/5926).

func NewReadClient 

func NewReadClient(name string, conf *ClientConfig) (ReadClient, error)

NewReadClient creates a new client for remote read.

type ReadyScrapeManager 

type ReadyScrapeManager interface {
	Get() (*scrape.Manager, error)
}

type RecoverableError 

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

type Storage 

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

Storage represents all the remote read and write endpoints. It implements storage.Storage.

func NewStorage 

func NewStorage(l log.Logger, reg prometheus.Registerer, stCallback startTimeCallback, walDir string, flushDeadline time.Duration, sm ReadyScrapeManager) *Storage

NewStorage returns a remote.Storage.

func (*Storage) Appender 

func (s *Storage) Appender(ctx context.Context) storage.Appender

Appender implements storage.Storage.

func (*Storage) ApplyConfig 

func (s *Storage) ApplyConfig(conf *config.Config) error

ApplyConfig updates the state as the new config requires.

func (*Storage) ChunkQuerier 

func (s *Storage) ChunkQuerier(mint, maxt int64) (storage.ChunkQuerier, error)

ChunkQuerier returns a storage.MergeQuerier combining the remote client queriers of each configured remote read endpoint.

func (*Storage) Close 

func (s *Storage) Close() error

Close the background processing of the storage queues.

func (*Storage) LowestSentTimestamp 

func (s *Storage) LowestSentTimestamp() int64

LowestSentTimestamp returns the lowest sent timestamp across all queues.

func (*Storage) Notify 

func (s *Storage) Notify()

func (*Storage) Querier 

func (s *Storage) Querier(mint, maxt int64) (storage.Querier, error)

Querier returns a storage.MergeQuerier combining the remote client queriers of each configured remote read endpoint. Returned querier will never return error as all queryables are assumed best effort. Additionally all returned queriers ensure that its Select's SeriesSets have ready data after first `Next` invoke. This is because Prometheus (fanout and secondary queries) can't handle the stream failing half way through by design.

func (*Storage) StartTime 

func (s *Storage) StartTime() (int64, error)

StartTime implements the Storage interface.

type Watchable 

type Watchable interface {
	TargetsActive() map[string][]*scrape.Target
}

Watchable represents from where we fetch active targets for metadata.

type WriteClient 

type WriteClient interface {
	// Store stores the given samples in the remote storage.
	Store(context.Context, []byte, int) error
	// Name uniquely identifies the remote storage.
	Name() string
	// Endpoint is the remote read or write endpoint for the storage client.
	Endpoint() string
}

WriteClient defines an interface for sending a batch of samples to an external timeseries database.

func NewWriteClient 

func NewWriteClient(name string, conf *ClientConfig) (WriteClient, error)

NewWriteClient creates a new client for remote write.

type WriteStorage 

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

WriteStorage represents all the remote write storage.

func NewWriteStorage 

func NewWriteStorage(logger log.Logger, reg prometheus.Registerer, dir string, flushDeadline time.Duration, sm ReadyScrapeManager) *WriteStorage

NewWriteStorage creates and runs a WriteStorage.

func (*WriteStorage) Appender 

func (rws *WriteStorage) Appender(_ context.Context) storage.Appender

Appender implements storage.Storage.

func (*WriteStorage) ApplyConfig 

func (rws *WriteStorage) ApplyConfig(conf *config.Config) error

ApplyConfig updates the state as the new config requires. Only stop & create queues which have changes.

func (*WriteStorage) Close 

func (rws *WriteStorage) Close() error

Close closes the WriteStorage.

func (*WriteStorage) LowestSentTimestamp 

func (rws *WriteStorage) LowestSentTimestamp() int64

LowestSentTimestamp returns the lowest sent timestamp across all queues.

func (*WriteStorage) Notify 

func (rws *WriteStorage) Notify()

Source Files

View all Source files

Directories

Path Synopsis
otlptranslator
prometheus
prometheusremotewrite

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.