mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2025-02-09 15:27:11 +00:00
71bb9fc0d0
Regression was introduced at 564e6ea024
after implementing:
https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6928
ctx.Labels array could be incorrectly updated and changes to it after
relabeling rules can be lost.
E.g. ctx.Labels passed to WriteDataPoint function as slice copy, but
results of relabeling only changed an actual slice at ctx.Labels.
This commit replaces implicit relabeling call with explicit
`TryPrepareLabels` function.
It also reduces code diffs with cluster version and adds integration tests
related issue:
https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7865
---------
Signed-off-by: f41gh7 <nik@victoriametrics.com>
Co-authored-by: Roman Khavronenko <roman@victoriametrics.com>
104 lines
2.8 KiB
Go
104 lines
2.8 KiB
Go
package vmimport
|
|
|
|
import (
|
|
"net/http"
|
|
"sync"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
|
|
parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
|
|
parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport/stream"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
|
|
"github.com/VictoriaMetrics/metrics"
|
|
)
|
|
|
|
var (
|
|
rowsInserted = metrics.NewCounter(`vm_rows_inserted_total{type="vmimport"}`)
|
|
rowsPerInsert = metrics.NewHistogram(`vm_rows_per_insert{type="vmimport"}`)
|
|
)
|
|
|
|
// InsertHandler processes `/api/v1/import` request.
|
|
//
|
|
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6
|
|
func InsertHandler(req *http.Request) error {
|
|
extraLabels, err := parserCommon.GetExtraLabels(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
isGzipped := req.Header.Get("Content-Encoding") == "gzip"
|
|
return stream.Parse(req.Body, isGzipped, func(rows []parser.Row) error {
|
|
return insertRows(rows, extraLabels)
|
|
})
|
|
}
|
|
|
|
func insertRows(rows []parser.Row, extraLabels []prompbmarshal.Label) error {
|
|
ctx := getPushCtx()
|
|
defer putPushCtx(ctx)
|
|
|
|
rowsLen := 0
|
|
for i := range rows {
|
|
rowsLen += len(rows[i].Values)
|
|
}
|
|
ic := &ctx.Common
|
|
ic.Reset(rowsLen)
|
|
rowsTotal := 0
|
|
hasRelabeling := relabel.HasRelabeling()
|
|
for i := range rows {
|
|
r := &rows[i]
|
|
rowsTotal += len(r.Values)
|
|
ic.Labels = ic.Labels[:0]
|
|
for j := range r.Tags {
|
|
tag := &r.Tags[j]
|
|
ic.AddLabelBytes(tag.Key, tag.Value)
|
|
}
|
|
for j := range extraLabels {
|
|
label := &extraLabels[j]
|
|
ic.AddLabel(label.Name, label.Value)
|
|
}
|
|
if !ic.TryPrepareLabels(hasRelabeling) {
|
|
continue
|
|
}
|
|
ctx.metricNameBuf = storage.MarshalMetricNameRaw(ctx.metricNameBuf[:0], ic.Labels)
|
|
values := r.Values
|
|
timestamps := r.Timestamps
|
|
if len(timestamps) != len(values) {
|
|
logger.Panicf("BUG: len(timestamps)=%d must match len(values)=%d", len(timestamps), len(values))
|
|
}
|
|
for j, value := range values {
|
|
timestamp := timestamps[j]
|
|
if err := ic.WriteDataPoint(ctx.metricNameBuf, nil, timestamp, value); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
rowsInserted.Add(rowsTotal)
|
|
rowsPerInsert.Update(float64(rowsTotal))
|
|
return ic.FlushBufs()
|
|
}
|
|
|
|
type pushCtx struct {
|
|
Common common.InsertCtx
|
|
metricNameBuf []byte
|
|
}
|
|
|
|
func (ctx *pushCtx) reset() {
|
|
ctx.Common.Reset(0)
|
|
ctx.metricNameBuf = ctx.metricNameBuf[:0]
|
|
}
|
|
|
|
func getPushCtx() *pushCtx {
|
|
if v := pushCtxPool.Get(); v != nil {
|
|
return v.(*pushCtx)
|
|
}
|
|
return &pushCtx{}
|
|
}
|
|
|
|
func putPushCtx(ctx *pushCtx) {
|
|
ctx.reset()
|
|
pushCtxPool.Put(ctx)
|
|
}
|
|
|
|
var pushCtxPool sync.Pool
|