prometheus/storage/remote/client.go

203 lines
5.7 KiB
Go
Raw Normal View History

2017-05-10 02:44:13 -07:00
// Copyright 2016 The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package remote
import (
"bufio"
"bytes"
"context"
2017-05-10 02:44:13 -07:00
"fmt"
"io"
"io/ioutil"
"net/http"
"strings"
2017-05-10 02:44:13 -07:00
"time"
"github.com/gogo/protobuf/proto"
2017-05-10 02:44:13 -07:00
"github.com/golang/snappy"
"github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go/ext"
"github.com/pkg/errors"
config_util "github.com/prometheus/common/config"
"github.com/prometheus/common/model"
"github.com/prometheus/common/version"
2017-05-10 02:44:13 -07:00
2017-07-12 14:06:35 -07:00
"github.com/prometheus/prometheus/prompb"
2017-05-10 02:44:13 -07:00
)
const maxErrMsgLen = 256
var userAgent = fmt.Sprintf("Prometheus/%s", version.Version)
2017-05-10 02:44:13 -07:00
// Client allows reading and writing from/to a remote HTTP endpoint.
type Client struct {
remoteName string // Used to differentiate clients in metrics.
url *config_util.URL
client *http.Client
timeout time.Duration
2017-05-10 02:44:13 -07:00
}
// ClientConfig configures a Client.
type ClientConfig struct {
Refactor SD configuration to remove `config` dependency (#3629) * refactor: move targetGroup struct and CheckOverflow() to their own package * refactor: move auth and security related structs to a utility package, fix import error in utility package * refactor: Azure SD, remove SD struct from config * refactor: DNS SD, remove SD struct from config into dns package * refactor: ec2 SD, move SD struct from config into the ec2 package * refactor: file SD, move SD struct from config to file discovery package * refactor: gce, move SD struct from config to gce discovery package * refactor: move HTTPClientConfig and URL into util/config, fix import error in httputil * refactor: consul, move SD struct from config into consul discovery package * refactor: marathon, move SD struct from config into marathon discovery package * refactor: triton, move SD struct from config to triton discovery package, fix test * refactor: zookeeper, move SD structs from config to zookeeper discovery package * refactor: openstack, remove SD struct from config, move into openstack discovery package * refactor: kubernetes, move SD struct from config into kubernetes discovery package * refactor: notifier, use targetgroup package instead of config * refactor: tests for file, marathon, triton SD - use targetgroup package instead of config.TargetGroup * refactor: retrieval, use targetgroup package instead of config.TargetGroup * refactor: storage, use config util package * refactor: discovery manager, use targetgroup package instead of config.TargetGroup * refactor: use HTTPClient and TLS config from configUtil instead of config * refactor: tests, use targetgroup package instead of config.TargetGroup * refactor: fix tagetgroup.Group pointers that were removed by mistake * refactor: openstack, kubernetes: drop prefixes * refactor: remove import aliases forced due to vscode bug * refactor: move main SD struct out of config into discovery/config * refactor: rename configUtil to config_util * refactor: rename yamlUtil to yaml_config * refactor: kubernetes, remove prefixes * refactor: move the TargetGroup package to discovery/ * refactor: fix order of imports
2017-12-29 12:01:34 -08:00
URL *config_util.URL
Timeout model.Duration
Refactor SD configuration to remove `config` dependency (#3629) * refactor: move targetGroup struct and CheckOverflow() to their own package * refactor: move auth and security related structs to a utility package, fix import error in utility package * refactor: Azure SD, remove SD struct from config * refactor: DNS SD, remove SD struct from config into dns package * refactor: ec2 SD, move SD struct from config into the ec2 package * refactor: file SD, move SD struct from config to file discovery package * refactor: gce, move SD struct from config to gce discovery package * refactor: move HTTPClientConfig and URL into util/config, fix import error in httputil * refactor: consul, move SD struct from config into consul discovery package * refactor: marathon, move SD struct from config into marathon discovery package * refactor: triton, move SD struct from config to triton discovery package, fix test * refactor: zookeeper, move SD structs from config to zookeeper discovery package * refactor: openstack, remove SD struct from config, move into openstack discovery package * refactor: kubernetes, move SD struct from config into kubernetes discovery package * refactor: notifier, use targetgroup package instead of config * refactor: tests for file, marathon, triton SD - use targetgroup package instead of config.TargetGroup * refactor: retrieval, use targetgroup package instead of config.TargetGroup * refactor: storage, use config util package * refactor: discovery manager, use targetgroup package instead of config.TargetGroup * refactor: use HTTPClient and TLS config from configUtil instead of config * refactor: tests, use targetgroup package instead of config.TargetGroup * refactor: fix tagetgroup.Group pointers that were removed by mistake * refactor: openstack, kubernetes: drop prefixes * refactor: remove import aliases forced due to vscode bug * refactor: move main SD struct out of config into discovery/config * refactor: rename configUtil to config_util * refactor: rename yamlUtil to yaml_config * refactor: kubernetes, remove prefixes * refactor: move the TargetGroup package to discovery/ * refactor: fix order of imports
2017-12-29 12:01:34 -08:00
HTTPClientConfig config_util.HTTPClientConfig
2017-05-10 02:44:13 -07:00
}
// NewClient creates a new Client.
func NewClient(remoteName string, conf *ClientConfig) (*Client, error) {
httpClient, err := config_util.NewClientFromConfig(conf.HTTPClientConfig, "remote_storage", false)
2017-05-10 02:44:13 -07:00
if err != nil {
return nil, err
}
return &Client{
remoteName: remoteName,
url: conf.URL,
client: httpClient,
timeout: time.Duration(conf.Timeout),
2017-05-10 02:44:13 -07:00
}, nil
}
type recoverableError struct {
error
}
Tail the TSDB WAL for remote_write This change switches the remote_write API to use the TSDB WAL. This should reduce memory usage and prevent sample loss when the remote end point is down. We use the new LiveReader from TSDB to tail WAL segments. Logic for finding the tracking segment is included in this PR. The WAL is tailed once for each remote_write endpoint specified. Reading from the segment is based on a ticker rather than relying on fsnotify write events, which were found to be complicated and unreliable in early prototypes. Enqueuing a sample for sending via remote_write can now block, to provide back pressure. Queues are still required to acheive parallelism and batching. We have updated the queue config based on new defaults for queue capacity and pending samples values - much smaller values are now possible. The remote_write resharding code has been updated to prevent deadlocks, and extra tests have been added for these cases. As part of this change, we attempt to guarantee that samples are not lost; however this initial version doesn't guarantee this across Prometheus restarts or non-retryable errors from the remote end (eg 400s). This changes also includes the following optimisations: - only marshal the proto request once, not once per retry - maintain a single copy of the labels for given series to reduce GC pressure Other minor tweaks: - only reshard if we've also successfully sent recently - add pending samples, latest sent timestamp, WAL events processed metrics Co-authored-by: Chris Marchbanks <csmarchbanks.com> (initial prototype) Co-authored-by: Tom Wilkie <tom.wilkie@gmail.com> (sharding changes) Signed-off-by: Callum Styan <callumstyan@gmail.com>
2018-09-07 14:26:04 -07:00
// Store sends a batch of samples to the HTTP endpoint, the request is the proto marshalled
// and encoded bytes from codec.go.
func (c *Client) Store(ctx context.Context, req []byte) error {
httpReq, err := http.NewRequest("POST", c.url.String(), bytes.NewReader(req))
2017-05-10 02:44:13 -07:00
if err != nil {
// Errors from NewRequest are from unparsable URLs, so are not
2017-05-10 02:44:13 -07:00
// recoverable.
return err
}
httpReq.Header.Add("Content-Encoding", "snappy")
httpReq.Header.Set("Content-Type", "application/x-protobuf")
httpReq.Header.Set("User-Agent", userAgent)
2017-05-10 02:44:13 -07:00
httpReq.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
ctx, cancel := context.WithTimeout(ctx, c.timeout)
2017-05-10 02:44:13 -07:00
defer cancel()
httpResp, err := c.client.Do(httpReq.WithContext(ctx))
2017-05-10 02:44:13 -07:00
if err != nil {
// Errors from client.Do are from (for example) network errors, so are
// recoverable.
return recoverableError{err}
}
defer func() {
io.Copy(ioutil.Discard, httpResp.Body)
httpResp.Body.Close()
}()
2017-05-10 02:44:13 -07:00
if httpResp.StatusCode/100 != 2 {
scanner := bufio.NewScanner(io.LimitReader(httpResp.Body, maxErrMsgLen))
line := ""
if scanner.Scan() {
line = scanner.Text()
}
err = errors.Errorf("server returned HTTP status %s: %s", httpResp.Status, line)
2017-05-10 02:44:13 -07:00
}
if httpResp.StatusCode/100 == 5 {
return recoverableError{err}
}
return err
}
// Name uniquely identifies the client.
2017-05-10 02:44:13 -07:00
func (c Client) Name() string {
return c.remoteName
}
// Endpoint is the remote read or write endpoint.
func (c Client) Endpoint() string {
return c.url.String()
2017-05-10 02:44:13 -07:00
}
// Read reads from a remote endpoint.
2017-10-23 13:28:17 -07:00
func (c *Client) Read(ctx context.Context, query *prompb.Query) (*prompb.QueryResult, error) {
2017-10-23 06:44:57 -07:00
req := &prompb.ReadRequest{
// TODO: Support batching multiple queries into one read request,
// as the protobuf interface allows for it.
2017-10-23 06:44:57 -07:00
Queries: []*prompb.Query{
query,
},
}
2017-05-10 02:44:13 -07:00
data, err := proto.Marshal(req)
if err != nil {
return nil, errors.Wrapf(err, "unable to marshal read request")
2017-05-10 02:44:13 -07:00
}
compressed := snappy.Encode(nil, data)
httpReq, err := http.NewRequest("POST", c.url.String(), bytes.NewReader(compressed))
if err != nil {
return nil, errors.Wrap(err, "unable to create request")
2017-05-10 02:44:13 -07:00
}
httpReq.Header.Add("Content-Encoding", "snappy")
httpReq.Header.Add("Accept-Encoding", "snappy")
2017-05-10 02:44:13 -07:00
httpReq.Header.Set("Content-Type", "application/x-protobuf")
httpReq.Header.Set("User-Agent", userAgent)
2017-05-10 02:44:13 -07:00
httpReq.Header.Set("X-Prometheus-Remote-Read-Version", "0.1.0")
if span := opentracing.SpanFromContext(ctx); span != nil {
ext.SpanKindRPCClient.Set(span)
ext.HTTPUrl.Set(span, httpReq.URL.String())
ext.HTTPMethod.Set(span, httpReq.Method)
_ = span.Tracer().Inject(
span.Context(),
opentracing.HTTPHeaders,
opentracing.HTTPHeadersCarrier(httpReq.Header),
)
}
2017-05-10 02:44:13 -07:00
ctx, cancel := context.WithTimeout(ctx, c.timeout)
defer cancel()
httpResp, err := c.client.Do(httpReq.WithContext(ctx))
2017-05-10 02:44:13 -07:00
if err != nil {
return nil, errors.Wrap(err, "error sending request")
2017-05-10 02:44:13 -07:00
}
defer func() {
io.Copy(ioutil.Discard, httpResp.Body)
httpResp.Body.Close()
}()
2017-05-10 02:44:13 -07:00
compressed, err = ioutil.ReadAll(httpResp.Body)
if err != nil {
return nil, errors.Wrap(err, fmt.Sprintf("error reading response. HTTP status code: %s", httpResp.Status))
}
if httpResp.StatusCode/100 != 2 {
return nil, errors.Errorf("remote server %s returned HTTP status %s: %s", c.url.String(), httpResp.Status, strings.TrimSpace(string(compressed)))
2017-05-10 02:44:13 -07:00
}
uncompressed, err := snappy.Decode(nil, compressed)
if err != nil {
return nil, errors.Wrap(err, "error reading response")
2017-05-10 02:44:13 -07:00
}
2017-07-12 14:06:35 -07:00
var resp prompb.ReadResponse
2017-05-10 02:44:13 -07:00
err = proto.Unmarshal(uncompressed, &resp)
if err != nil {
return nil, errors.Wrap(err, "unable to unmarshal response body")
2017-05-10 02:44:13 -07:00
}
if len(resp.Results) != len(req.Queries) {
return nil, errors.Errorf("responses: want %d, got %d", len(req.Queries), len(resp.Results))
2017-05-10 02:44:13 -07:00
}
2017-10-23 13:28:17 -07:00
return resp.Results[0], nil
2017-05-10 02:44:13 -07:00
}