2019-05-22 23:16:55 +02:00
|
|
|
package graphite
|
|
|
|
|
|
|
|
import (
|
|
|
|
"fmt"
|
|
|
|
"io"
|
|
|
|
"net"
|
|
|
|
"runtime"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/concurrencylimiter"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
|
|
|
|
"github.com/VictoriaMetrics/metrics"
|
|
|
|
)
|
|
|
|
|
2019-07-27 12:20:47 +02:00
|
|
|
var (
|
|
|
|
rowsInserted = metrics.NewCounter(`vm_rows_inserted_total{type="graphite"}`)
|
|
|
|
rowsPerInsert = metrics.NewSummary(`vm_rows_per_insert{type="graphite"}`)
|
|
|
|
)
|
2019-05-22 23:16:55 +02:00
|
|
|
|
|
|
|
// insertHandler processes remote write for graphite plaintext protocol.
|
|
|
|
//
|
|
|
|
// See https://graphite.readthedocs.io/en/latest/feeding-carbon.html#the-plaintext-protocol
|
|
|
|
func insertHandler(r io.Reader) error {
|
|
|
|
return concurrencylimiter.Do(func() error {
|
|
|
|
return insertHandlerInternal(r)
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
func insertHandlerInternal(r io.Reader) error {
|
|
|
|
ctx := getPushCtx()
|
|
|
|
defer putPushCtx(ctx)
|
|
|
|
for ctx.Read(r) {
|
|
|
|
if err := ctx.InsertRows(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return ctx.Error()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (ctx *pushCtx) InsertRows() error {
|
|
|
|
rows := ctx.Rows.Rows
|
|
|
|
ic := &ctx.Common
|
|
|
|
ic.Reset(len(rows))
|
|
|
|
for i := range rows {
|
|
|
|
r := &rows[i]
|
|
|
|
ic.Labels = ic.Labels[:0]
|
|
|
|
ic.AddLabel("", r.Metric)
|
|
|
|
for j := range r.Tags {
|
|
|
|
tag := &r.Tags[j]
|
|
|
|
ic.AddLabel(tag.Key, tag.Value)
|
|
|
|
}
|
|
|
|
ic.WriteDataPoint(nil, ic.Labels, r.Timestamp, r.Value)
|
|
|
|
}
|
|
|
|
rowsInserted.Add(len(rows))
|
2019-07-27 12:20:47 +02:00
|
|
|
rowsPerInsert.Update(float64(len(rows)))
|
2019-05-22 23:16:55 +02:00
|
|
|
return ic.FlushBufs()
|
|
|
|
}
|
|
|
|
|
|
|
|
const flushTimeout = 3 * time.Second
|
|
|
|
|
|
|
|
func (ctx *pushCtx) Read(r io.Reader) bool {
|
|
|
|
graphiteReadCalls.Inc()
|
|
|
|
if ctx.err != nil {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
if c, ok := r.(net.Conn); ok {
|
|
|
|
if err := c.SetReadDeadline(time.Now().Add(flushTimeout)); err != nil {
|
|
|
|
graphiteReadErrors.Inc()
|
|
|
|
ctx.err = fmt.Errorf("cannot set read deadline: %s", err)
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
}
|
2019-05-28 16:31:35 +02:00
|
|
|
ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(r, ctx.reqBuf, ctx.tailBuf)
|
|
|
|
if ctx.err != nil {
|
|
|
|
if ne, ok := ctx.err.(net.Error); ok && ne.Timeout() {
|
2019-05-22 23:16:55 +02:00
|
|
|
// Flush the read data on timeout and try reading again.
|
2019-05-28 16:31:35 +02:00
|
|
|
ctx.err = nil
|
2019-05-22 23:16:55 +02:00
|
|
|
} else {
|
2019-05-28 16:31:35 +02:00
|
|
|
if ctx.err != io.EOF {
|
|
|
|
graphiteReadErrors.Inc()
|
|
|
|
ctx.err = fmt.Errorf("cannot read graphite plaintext protocol data: %s", ctx.err)
|
|
|
|
}
|
2019-05-22 23:16:55 +02:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
}
|
2019-05-28 16:31:35 +02:00
|
|
|
if err := ctx.Rows.Unmarshal(bytesutil.ToUnsafeString(ctx.reqBuf)); err != nil {
|
2019-05-22 23:16:55 +02:00
|
|
|
graphiteUnmarshalErrors.Inc()
|
2019-05-28 16:31:35 +02:00
|
|
|
ctx.err = fmt.Errorf("cannot unmarshal graphite plaintext protocol data with size %d: %s", len(ctx.reqBuf), err)
|
2019-05-22 23:16:55 +02:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2019-06-18 18:04:02 +02:00
|
|
|
// Fill missing timestamps with the current timestamp rounded to seconds.
|
|
|
|
currentTimestamp := time.Now().Unix()
|
|
|
|
rows := ctx.Rows.Rows
|
|
|
|
for i := range rows {
|
|
|
|
r := &rows[i]
|
|
|
|
if r.Timestamp == 0 {
|
|
|
|
r.Timestamp = currentTimestamp
|
|
|
|
}
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2019-06-18 18:04:02 +02:00
|
|
|
|
|
|
|
// Convert timestamps from seconds to milliseconds.
|
|
|
|
for i := range rows {
|
|
|
|
rows[i].Timestamp *= 1e3
|
|
|
|
}
|
|
|
|
|
2019-05-22 23:16:55 +02:00
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
type pushCtx struct {
|
|
|
|
Rows Rows
|
|
|
|
Common common.InsertCtx
|
|
|
|
|
2019-05-28 16:31:35 +02:00
|
|
|
reqBuf []byte
|
2019-05-22 23:16:55 +02:00
|
|
|
tailBuf []byte
|
|
|
|
|
|
|
|
err error
|
|
|
|
}
|
|
|
|
|
|
|
|
func (ctx *pushCtx) Error() error {
|
|
|
|
if ctx.err == io.EOF {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
return ctx.err
|
|
|
|
}
|
|
|
|
|
|
|
|
func (ctx *pushCtx) reset() {
|
|
|
|
ctx.Rows.Reset()
|
|
|
|
ctx.Common.Reset(0)
|
2019-05-28 16:31:35 +02:00
|
|
|
ctx.reqBuf = ctx.reqBuf[:0]
|
2019-05-22 23:16:55 +02:00
|
|
|
ctx.tailBuf = ctx.tailBuf[:0]
|
|
|
|
|
|
|
|
ctx.err = nil
|
|
|
|
}
|
|
|
|
|
|
|
|
var (
|
|
|
|
graphiteReadCalls = metrics.NewCounter(`vm_read_calls_total{name="graphite"}`)
|
|
|
|
graphiteReadErrors = metrics.NewCounter(`vm_read_errors_total{name="graphite"}`)
|
|
|
|
graphiteUnmarshalErrors = metrics.NewCounter(`vm_unmarshal_errors_total{name="graphite"}`)
|
|
|
|
)
|
|
|
|
|
|
|
|
func getPushCtx() *pushCtx {
|
|
|
|
select {
|
|
|
|
case ctx := <-pushCtxPoolCh:
|
|
|
|
return ctx
|
|
|
|
default:
|
|
|
|
if v := pushCtxPool.Get(); v != nil {
|
|
|
|
return v.(*pushCtx)
|
|
|
|
}
|
|
|
|
return &pushCtx{}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func putPushCtx(ctx *pushCtx) {
|
|
|
|
ctx.reset()
|
|
|
|
select {
|
|
|
|
case pushCtxPoolCh <- ctx:
|
|
|
|
default:
|
|
|
|
pushCtxPool.Put(ctx)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
var pushCtxPool sync.Pool
|
|
|
|
var pushCtxPoolCh = make(chan *pushCtx, runtime.GOMAXPROCS(-1))
|