远程写入prometheus存储,


简介

prometheus一般都是采用pull方式获取数据,但是有一些情况下,不方便配置exporter,就希望能通过push的方式上传指标数据。

1、可以采用pushgateway的方式,推送到pushgateway,然后prometheus通过pushgateway拉取数据。

2、在新版本中增加了一个参数:--enable-feature=remote-write-receiver,允许远程通过接口/api/v1/write,直接写数据到prometheus里面。

pushgateway在高并发的情况下还是比较消耗资源的,特别是开启一致性检查,高并发写入的时候特别慢。

第二种方式少了一层转发,速度应该比较快。

远程写入prometheus存储

接口

可以通过prometheus的http接口/api/v1/write提交数据,这个接口的数据格式有有要求:

  • 使用POST方式提交
  • 需要经过protobuf编码,依赖github.com/gogo/protobuf/proto
  • 可以使用snappy进行压缩,依赖github.com/golang/snappy

步骤:

  1. package prome 
  2.  
  3. import ( 
  4.     "bufio" 
  5.     "bytes" 
  6.     "context" 
  7.     "io" 
  8.     "io/ioutil" 
  9.     "net/http" 
  10.     "net/url" 
  11.     "regexp" 
  12.     "time" 
  13.  
  14.     "github.com/gogo/protobuf/proto" 
  15.     "github.com/golang/snappy" 
  16.     "github.com/opentracing-contrib/go-stdlib/nethttp" 
  17.     opentracing "github.com/opentracing/opentracing-go" 
  18.     "github.com/pkg/errors" 
  19.     "github.com/prometheus/common/model" 
  20.     "github.com/prometheus/prometheus/pkg/labels" 
  21.     "github.com/prometheus/prometheus/prompb" 
  22.  
  23. type RecoverableError struct { 
  24.     error 
  25.  
  26. type HttpClient struct { 
  27.     url     *url.URL 
  28.     Client  *http.Client 
  29.     timeout time.Duration 
  30.  
  31. var MetricNameRE = regexp.MustCompile(`^[a-zA-Z_:][a-zA-Z0-9_:]*$`) 
  32.  
  33. type MetricPoint struct { 
  34.     Metric  string            `json:"metric"` // 指标名称 
  35.     TagsMap map[string]string `json:"tags"`   // 数据标签 
  36.     Time    int64             `json:"time"`   // 时间戳,单位是秒 
  37.     Value   float64           `json:"value"`  // 内部字段,最终转换之后的float64数值 
  38.  
  39. func (c *HttpClient) remoteWritePost(req []byte) error { 
  40.     httpReq, err := http.NewRequest("POST", c.url.String(), bytes.NewReader(req)) 
  41.     if err != nil { 
  42.         return err 
  43.     } 
  44.     httpReq.Header.Add("Content-Encoding", "snappy") 
  45.     httpReq.Header.Set("Content-Type", "application/x-protobuf") 
  46.     httpReq.Header.Set("User-Agent", "opcai") 
  47.     httpReq.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0") 
  48.     ctx, cancel := context.WithTimeout(context.Background(), c.timeout) 
  49.     defer cancel() 
  50.  
  51.     httpReq = httpReq.WithContext(ctx) 
  52.  
  53.     if parentSpan := opentracing.SpanFromContext(ctx); parentSpan != nil { 
  54.         var ht *nethttp.Tracer 
  55.         httpReq, ht = nethttp.TraceRequest( 
  56.             parentSpan.Tracer(), 
  57.             httpReq, 
  58.             nethttp.OperationName("Remote Store"), 
  59.             nethttp.ClientTrace(false), 
  60.         ) 
  61.         defer ht.Finish() 
  62.     } 
  63.  
  64.     httpResp, err := c.Client.Do(httpReq) 
  65.     if err != nil { 
  66.         // Errors from Client.Do are from (for example) network errors, so are 
  67.         // recoverable. 
  68.         return RecoverableError{err} 
  69.     } 
  70.     defer func() { 
  71.         io.Copy(ioutil.Discard, httpResp.Body) 
  72.         httpResp.Body.Close() 
  73.     }() 
  74.  
  75.     if httpResp.StatusCode/100 != 2 { 
  76.         scanner := bufio.NewScanner(io.LimitReader(httpResp.Body, 512)) 
  77.         line := "" 
  78.         if scanner.Scan() { 
  79.             line = scanner.Text() 
  80.         } 
  81.         err = errors.Errorf("server returned HTTP status %s: %s", httpResp.Status, line) 
  82.     } 
  83.     if httpResp.StatusCode/100 == 5 { 
  84.         return RecoverableError{err} 
  85.     } 
  86.     return err 
  87.  
  88. func buildWriteRequest(samples []*prompb.TimeSeries) ([]byte, error) { 
  89.  
  90.     req := &prompb.WriteRequest{ 
  91.         Timeseries: samples, 
  92.     } 
  93.     data, err := proto.Marshal(req) 
  94.     if err != nil { 
  95.         return nil, err 
  96.     } 
  97.     compressed := snappy.Encode(nil, data) 
  98.     return compressed, nil 
  99.  
  100. type sample struct { 
  101.     labels labels.Labels 
  102.     t      int64 
  103.     v      float64 
  104.  
  105. const ( 
  106.     LABEL_NAME = "__name__" 
  107.  
  108. func convertOne(item *MetricPoint) (*prompb.TimeSeries, error) { 
  109.     pt := prompb.TimeSeries{} 
  110.     pt.Samples = []prompb.Sample{{}} 
  111.     s := sample{} 
  112.     s.t = item.Time 
  113.     s.v = item.Value 
  114.     // name 
  115.     if !MetricNameRE.MatchString(item.Metric) { 
  116.         return &pt, errors.New("invalid metrics name") 
  117.     } 
  118.     nameLs := labels.Label{ 
  119.         Name:  LABEL_NAME, 
  120.         Value: item.Metric, 
  121.     } 
  122.     s.labels = append(s.labels, nameLs) 
  123.     for k, v := range item.TagsMap { 
  124.         if model.LabelNameRE.MatchString(k) { 
  125.             ls := labels.Label{ 
  126.                 Name:  k, 
  127.                 Value: v, 
  128.             } 
  129.             s.labels = append(s.labels, ls) 
  130.         } 
  131.     } 
  132.  
  133.     pt.Labels = labelsToLabelsProto(s.labels, pt.Labels) 
  134.     // 时间赋值问题,使用毫秒时间戳 
  135.     tsMs := time.Unix(s.t, 0).UnixNano() / 1e6 
  136.     pt.Samples[0].Timestamp = tsMs 
  137.     pt.Samples[0].Value = s.v 
  138.     return &pt, nil 
  139.  
  140. func labelsToLabelsProto(labels labels.Labels, buf []*prompb.Label) []*prompb.Label { 
  141.     result := buf[:0] 
  142.     if cap(buf) < len(labels) { 
  143.         result = make([]*prompb.Label, 0, len(labels)) 
  144.     } 
  145.     for _, l := range labels { 
  146.         result = append(result, &prompb.Label{ 
  147.             Name:  l.Name, 
  148.             Value: l.Value, 
  149.         }) 
  150.     } 
  151.     return result 
  152.  
  153. func (c *HttpClient) RemoteWrite(items []MetricPoint) (err error) { 
  154.     if len(items) == 0 { 
  155.         return 
  156.     } 
  157.     ts := make([]*prompb.TimeSeries, len(items)) 
  158.     for i := range items { 
  159.         ts[i], err = convertOne(&items[i]) 
  160.         if err != nil { 
  161.             return 
  162.         } 
  163.     } 
  164.     data, err := buildWriteRequest(ts) 
  165.     if err != nil { 
  166.         return 
  167.     } 
  168.     err = c.remoteWritePost(data) 
  169.     return 
  170.  
  171. func NewClient(ur string, timeout time.Duration) (c *HttpClient, err error) { 
  172.     u, err := url.Parse(ur) 
  173.     if err != nil { 
  174.         return 
  175.     } 
  176.     c = &HttpClient{ 
  177.         url:     u, 
  178.         Client:  &http.Client{}, 
  179.         timeout: timeout, 
  180.     } 
  181.     return 

测试

prometheus启动的时候记得加参数--enable-feature=remote-write-receiver

  1. package prome 
  2.  
  3. import ( 
  4.     "testing" 
  5.     "time" 
  6.  
  7. func TestRemoteWrite(t *testing.T) { 
  8.     c, err := NewClient("http://localhost:9090/api/v1/write", 10*time.Second) 
  9.     if err != nil { 
  10.         t.Fatal(err) 
  11.     } 
  12.     metrics := []MetricPoint{ 
  13.         {Metric: "opcai1", 
  14.             TagsMap: map[string]string{"env": "testing", "op": "opcai"}, 
  15.             Time:    time.Now().Add(-1 * time.Minute).Unix(), 
  16.             Value:   1}, 
  17.         {Metric: "opcai2", 
  18.             TagsMap: map[string]string{"env": "testing", "op": "opcai"}, 
  19.             Time:    time.Now().Add(-2 * time.Minute).Unix(), 
  20.             Value:   2}, 
  21.         {Metric: "opcai3", 
  22.             TagsMap: map[string]string{"env": "testing", "op": "opcai"}, 
  23.             Time:    time.Now().Unix(), 
  24.             Value:   3}, 
  25.         {Metric: "opcai4", 
  26.             TagsMap: map[string]string{"env": "testing", "op": "opcai"}, 
  27.             Time:    time.Now().Unix(), 
  28.             Value:   4}, 
  29.     } 
  30.     err = c.RemoteWrite(metrics) 
  31.     if err != nil { 
  32.         t.Fatal(err) 
  33.     } 
  34.     t.Log("end...") 

使用go test进行测试

  1. go test -v 

总结

这个方法也是在看夜莺v5的代码的时候发现的,刚好有需要统一收集redis的监控指标,刚好可以用上,之前用pushgateway写的实在是慢。

相关内容