远程写入prometheus存储

复制package prome  import (      "bufio"     "bytes"     "context"     "io"     "io/ioutil"     "net/http"     "net/url"     "regexp"     "time"     "github.com/gogo/protobuf/proto"     "github.com/golang/snappy"     "github.com/opentracing-contrib/go-stdlib/nethttp"     opentracing "github.com/opentracing/opentracing-go"     "github.com/pkg/errors"     "github.com/prometheus/common/model"     "github.com/prometheus/prometheus/pkg/labels"     "github.com/prometheus/prometheus/prompb" )  type RecoverableError struct {      error }  type HttpClient struct {      url     *url.URL      Client  *http.Client      timeout time.Duration }  var MetricNameRE = regexp.MustCompile(`^[a-zA-Z_:][a-zA-Z0-9_:]*$`)  type MetricPoint struct {      Metric  string            `json:"metric"` // 指标名称      TagsMap map[string]string `json:"tags"`   // 数据标签      Time    int64             `json:"time"`   // 时间戳，远程单位是云服务器写入秒      Value   float64           `json:"value"`  // 内部字段，企商汇最终转换之后的香港云服务器存储float64数值 }  func (c *HttpClient) remoteWritePost(req []byte) error {      httpReq, err := http.NewRequest("POST", c.url.String(), bytes.NewReader(req))      if err != nil {          return err      }      httpReq.Header.Add("Content-Encoding", "snappy")      httpReq.Header.Set("Content-Type", "application/x-protobuf")      httpReq.Header.Set("User-Agent", "opcai")      httpReq.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")      ctx, cancel := context.WithTimeout(context.Background(), c.timeout)      defer cancel()      httpReq = httpReq.WithContext(ctx)      if parentSpan := opentracing.SpanFromContext(ctx); parentSpan != nil {          var ht *nethttp.Tracer          httpReq, ht = nethttp.TraceRequest(              parentSpan.Tracer(),              httpReq,              nethttp.OperationName("Remote Store"),              nethttp.ClientTrace(false),          )          defer ht.Finish()      }      httpResp, err := c.Client.Do(httpReq)      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()      }()      if httpResp.StatusCode/100 != 2 {          scanner := bufio.NewScanner(io.LimitReader(httpResp.Body, 512))          line := ""         if scanner.Scan() {              line = scanner.Text()          }          err = errors.Errorf("server returned HTTP status %s: %s", httpResp.Status, line)      }      if httpResp.StatusCode/100 == 5 {          return RecoverableError{err}      }      return err  }  func buildWriteRequest(samples []*prompb.TimeSeries) ([]byte, error) {      req := &prompb.WriteRequest{          Timeseries: samples,      }      data, err := proto.Marshal(req)      if err != nil {          return nil, err      }      compressed := snappy.Encode(nil, data)      return compressed, nil  }  type sample struct {      labels labels.Labels      t      int64      v      float64 }  const (      LABEL_NAME = "__name__" )  func convertOne(item *MetricPoint) (*prompb.TimeSeries, error) {      pt := prompb.TimeSeries{}      pt.Samples = []prompb.Sample{{}}      s := sample{}      s.t = item.Time     s.v = item.Value      // name     if !MetricNameRE.MatchString(item.Metric) {          return &pt, errors.New("invalid metrics name")      }      nameLs := labels.Label{          Name:  LABEL_NAME,          Value: item.Metric,      }      s.labels = append(s.labels, nameLs)      for k, v := range item.TagsMap {          if model.LabelNameRE.MatchString(k) {              ls := labels.Label{                  Name:  k,                  Value: v,              }              s.labels = append(s.labels, ls)          }      }      pt.Labels = labelsToLabelsProto(s.labels, pt.Labels)      // 时间赋值问题,使用毫秒时间戳      tsMs := time.Unix(s.t, 0).UnixNano() / 1e6      pt.Samples[0].Timestamp = tsMs      pt.Samples[0].Value = s.v      return &pt, nil  }  func labelsToLabelsProto(labels labels.Labels, buf []*prompb.Label) []*prompb.Label {      result := buf[:0]      if cap(buf) < len(labels) {          result = make([]*prompb.Label, 0, len(labels))      }      for _, l := range labels {          result = append(result, &prompb.Label{              Name:  l.Name,              Value: l.Value,          })      }      return result  }  func (c *HttpClient) RemoteWrite(items []MetricPoint) (err error) {      if len(items) == 0 {          return     }      ts := make([]*prompb.TimeSeries, len(items))      for i := range items {          ts[i], err = convertOne(&items[i])          if err != nil {              return         }      }      data, err := buildWriteRequest(ts)      if err != nil {          return     }      err = c.remoteWritePost(data)      return }  func NewClient(ur string, timeout time.Duration) (c *HttpClient, err error) {      u, err := url.Parse(ur)      if err != nil {          return     }      c = &HttpClient{          url:     u,          Client:  &http.Client{},          timeout: timeout,      }      return }  1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.39.40.41.42.43.44.45.46.47.48.49.50.51.52.53.54.55.56.57.58.59.60.61.62.63.64.65.66.67.68.69.70.71.72.73.74.75.76.77.78.79.80.81.82.83.84.85.86.87.88.89.90.91.92.93.94.95.96.97.98.99.100.101.102.103.104.105.106.107.108.109.110.111.112.113.114.115.116.117.118.119.120.121.122.123.124.125.126.127.128.129.130.131.132.133.134.135.136.137.138.139.140.141.142.143.144.145.146.147.148.149.150.151.152.153.154.155.156.157.158.159.160.161.162.163.164.165.166.167.168.169.170.171.172.173.174.175.176.177.178.179.180.181.182.183.184.185.186.187.188.189.190.191.192.193.