UploadMetrics RPC uploads metric info.
(req *proto.UploadMetricsReq, resp *proto.UploadMetricsResp)
| 64 | |
| 65 | // UploadMetrics RPC uploads metric info. |
| 66 | func (cs *CollectServer) UploadMetrics(req *proto.UploadMetricsReq, resp *proto.UploadMetricsResp) (err error) { |
| 67 | reqNodeID := req.GetNodeID().ToNodeID() |
| 68 | if reqNodeID.IsEmpty() { |
| 69 | err = errors.New("empty node id") |
| 70 | log.Error(err) |
| 71 | return |
| 72 | } |
| 73 | if !route.IsPermitted(&req.Envelope, route.MetricUploadMetrics) { |
| 74 | err = fmt.Errorf("calling from node %s is not permitted", reqNodeID) |
| 75 | log.Error(err) |
| 76 | return |
| 77 | } |
| 78 | |
| 79 | mfm := make(SimpleMetricMap, len(req.MFBytes)) |
| 80 | log.Debugf("RPC received MFS len %d", len(req.MFBytes)) |
| 81 | for _, mf := range req.MFBytes[:] { |
| 82 | bufReader := bytes.NewReader(mf) |
| 83 | //mf := new(dto.MetricFamily) |
| 84 | //dec := expfmt.NewDecoder(bufReader, expfmt.FmtProtoCompact) |
| 85 | //err = dec.Decode(mf) |
| 86 | tp := expfmt.TextParser{} |
| 87 | mf, err := tp.TextToMetricFamilies(bufReader) |
| 88 | if err != nil { |
| 89 | log.Warnf("decode MetricFamily failed: %s", err) |
| 90 | continue |
| 91 | } |
| 92 | //log.Debugf("RPC received MF: %#v", mf) |
| 93 | for k, v := range mf { |
| 94 | mfm[k] = v |
| 95 | } |
| 96 | } |
| 97 | //log.Debugf("MetricFamily uploaded: %v, %v", reqNodeID, mfm) |
| 98 | if len(mfm) > 0 { |
| 99 | cs.NodeMetric.Store(reqNodeID, mfm) |
| 100 | } else { |
| 101 | err = errors.New("no valid metric received") |
| 102 | log.Error(err) |
| 103 | } |
| 104 | return |
| 105 | } |
| 106 | |
| 107 | // GatherMetricBytes gathers the registered metric info and encode it to [][]byte. |
| 108 | func (cc *CollectClient) GatherMetricBytes() (mfb [][]byte, err error) { |