2019-05-22 21:16:55 +00:00
package influx
import (
2019-06-14 06:57:13 +00:00
"flag"
2020-02-25 17:09:46 +00:00
"io"
2019-05-22 21:16:55 +00:00
"net/http"
"sync"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
2020-07-02 16:42:12 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
2019-05-22 21:16:55 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
2020-12-08 18:49:32 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
2020-07-23 09:50:41 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
2021-01-12 22:52:50 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
2020-02-23 11:35:47 +00:00
parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
2019-05-22 21:16:55 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
2020-02-23 11:35:47 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
2019-05-22 21:16:55 +00:00
"github.com/VictoriaMetrics/metrics"
)
2019-06-14 06:57:13 +00:00
var (
2020-02-04 13:46:13 +00:00
measurementFieldSeparator = flag . String ( "influxMeasurementFieldSeparator" , "_" , "Separator for '{measurement}{separator}{field_name}' metric name when inserted via Influx line protocol" )
skipSingleField = flag . Bool ( "influxSkipSingleField" , false , "Uses '{measurement}' instead of '{measurement}{separator}{field_name}' for metic name if Influx line contains only a single field" )
2020-07-14 11:17:22 +00:00
skipMeasurement = flag . Bool ( "influxSkipMeasurement" , false , "Uses '{field_name}' as a metric name while ignoring '{measurement}' and '-influxMeasurementFieldSeparator'" )
2019-06-14 06:57:13 +00:00
)
2019-07-27 10:20:47 +00:00
var (
rowsInserted = metrics . NewCounter ( ` vm_rows_inserted_total { type="influx"} ` )
2020-02-23 11:35:47 +00:00
rowsPerInsert = metrics . NewHistogram ( ` vm_rows_per_insert { type="influx"} ` )
2019-07-27 10:20:47 +00:00
)
2019-05-22 21:16:55 +00:00
2020-02-25 17:09:46 +00:00
// InsertHandlerForReader processes remote write for influx line protocol.
//
// See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
func InsertHandlerForReader ( r io . Reader ) error {
return writeconcurrencylimiter . Do ( func ( ) error {
2021-01-12 22:52:50 +00:00
return parser . ParseStream ( r , false , "" , "" , func ( db string , rows [ ] parser . Row ) error {
return insertRows ( db , rows , nil )
} )
2020-02-25 17:09:46 +00:00
} )
}
// InsertHandlerForHTTP processes remote write for influx line protocol.
2019-05-22 21:16:55 +00:00
//
// See https://github.com/influxdata/influxdb/blob/4cbdc197b8117fee648d62e2e5be75c6575352f0/tsdb/README.md
2020-02-25 17:09:46 +00:00
func InsertHandlerForHTTP ( req * http . Request ) error {
2021-01-12 22:52:50 +00:00
extraLabels , err := parserCommon . GetExtraLabels ( req )
if err != nil {
return err
}
2020-02-23 11:35:47 +00:00
return writeconcurrencylimiter . Do ( func ( ) error {
2020-02-25 17:09:46 +00:00
isGzipped := req . Header . Get ( "Content-Encoding" ) == "gzip"
q := req . URL . Query ( )
precision := q . Get ( "precision" )
// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
db := q . Get ( "db" )
2021-01-12 22:52:50 +00:00
return parser . ParseStream ( req . Body , isGzipped , precision , db , func ( db string , rows [ ] parser . Row ) error {
return insertRows ( db , rows , extraLabels )
} )
2019-05-22 21:16:55 +00:00
} )
}
2021-01-12 22:52:50 +00:00
func insertRows ( db string , rows [ ] parser . Row , extraLabels [ ] prompbmarshal . Label ) error {
2019-05-22 21:16:55 +00:00
ctx := getPushCtx ( )
defer putPushCtx ( ctx )
rowsLen := 0
for i := range rows {
2019-12-09 18:58:19 +00:00
rowsLen += len ( rows [ i ] . Fields )
2019-05-22 21:16:55 +00:00
}
ic := & ctx . Common
ic . Reset ( rowsLen )
2019-07-27 10:20:47 +00:00
rowsTotal := 0
2020-07-02 16:42:12 +00:00
hasRelabeling := relabel . HasRelabeling ( )
2019-05-22 21:16:55 +00:00
for i := range rows {
r := & rows [ i ]
2020-10-09 10:29:27 +00:00
rowsTotal += len ( r . Fields )
2019-05-22 21:16:55 +00:00
ic . Labels = ic . Labels [ : 0 ]
2020-07-28 18:23:01 +00:00
hasDBKey := false
2019-05-22 21:16:55 +00:00
for j := range r . Tags {
tag := & r . Tags [ j ]
2019-08-24 10:51:51 +00:00
if tag . Key == "db" {
2020-07-28 18:23:01 +00:00
hasDBKey = true
2019-08-24 10:51:51 +00:00
}
2019-05-22 21:16:55 +00:00
ic . AddLabel ( tag . Key , tag . Value )
}
2020-07-28 18:23:01 +00:00
if ! hasDBKey {
ic . AddLabel ( "db" , db )
}
2021-01-12 22:52:50 +00:00
for j := range extraLabels {
label := & extraLabels [ j ]
ic . AddLabel ( label . Name , label . Value )
}
2020-07-14 11:17:22 +00:00
ctx . metricGroupBuf = ctx . metricGroupBuf [ : 0 ]
if ! * skipMeasurement {
ctx . metricGroupBuf = append ( ctx . metricGroupBuf , r . Measurement ... )
}
2019-06-14 07:51:57 +00:00
skipFieldKey := len ( r . Fields ) == 1 && * skipSingleField
2019-11-30 19:54:34 +00:00
if len ( ctx . metricGroupBuf ) > 0 && ! skipFieldKey {
2019-06-14 07:51:57 +00:00
ctx . metricGroupBuf = append ( ctx . metricGroupBuf , * measurementFieldSeparator ... )
}
2019-05-22 21:16:55 +00:00
metricGroupPrefixLen := len ( ctx . metricGroupBuf )
2020-07-23 09:50:41 +00:00
if hasRelabeling {
ctx . originLabels = append ( ctx . originLabels [ : 0 ] , ic . Labels ... )
for j := range r . Fields {
f := & r . Fields [ j ]
if ! skipFieldKey {
ctx . metricGroupBuf = append ( ctx . metricGroupBuf [ : metricGroupPrefixLen ] , f . Key ... )
}
metricGroup := bytesutil . ToUnsafeString ( ctx . metricGroupBuf )
ic . Labels = append ( ic . Labels [ : 0 ] , ctx . originLabels ... )
ic . AddLabel ( "" , metricGroup )
ic . ApplyRelabeling ( )
if len ( ic . Labels ) == 0 {
// Skip metric without labels.
continue
}
2020-07-24 20:19:49 +00:00
if err := ic . WriteDataPoint ( nil , ic . Labels , r . Timestamp , f . Value ) ; err != nil {
return err
}
2020-07-02 20:13:13 +00:00
}
2020-07-23 09:50:41 +00:00
} else {
ctx . metricNameBuf = storage . MarshalMetricNameRaw ( ctx . metricNameBuf [ : 0 ] , ic . Labels )
labelsLen := len ( ic . Labels )
for j := range r . Fields {
f := & r . Fields [ j ]
if ! skipFieldKey {
ctx . metricGroupBuf = append ( ctx . metricGroupBuf [ : metricGroupPrefixLen ] , f . Key ... )
}
metricGroup := bytesutil . ToUnsafeString ( ctx . metricGroupBuf )
ic . Labels = ic . Labels [ : labelsLen ]
ic . AddLabel ( "" , metricGroup )
if len ( ic . Labels ) == 0 {
// Skip metric without labels.
continue
}
2020-07-24 20:19:49 +00:00
if err := ic . WriteDataPoint ( ctx . metricNameBuf , ic . Labels [ len ( ic . Labels ) - 1 : ] , r . Timestamp , f . Value ) ; err != nil {
return err
}
2020-07-02 16:42:12 +00:00
}
2019-05-22 21:16:55 +00:00
}
}
2019-07-27 10:20:47 +00:00
rowsInserted . Add ( rowsTotal )
rowsPerInsert . Update ( float64 ( rowsTotal ) )
2019-05-22 21:16:55 +00:00
return ic . FlushBufs ( )
}
type pushCtx struct {
2020-02-23 11:35:47 +00:00
Common common . InsertCtx
2019-05-22 21:16:55 +00:00
metricNameBuf [ ] byte
metricGroupBuf [ ] byte
2020-07-23 09:50:41 +00:00
originLabels [ ] prompb . Label
2019-05-22 21:16:55 +00:00
}
func ( ctx * pushCtx ) reset ( ) {
ctx . Common . Reset ( 0 )
ctx . metricNameBuf = ctx . metricNameBuf [ : 0 ]
ctx . metricGroupBuf = ctx . metricGroupBuf [ : 0 ]
2020-07-23 09:50:41 +00:00
originLabels := ctx . originLabels
for i := range originLabels {
label := & originLabels [ i ]
label . Name = nil
label . Value = nil
}
ctx . originLabels = ctx . originLabels [ : 0 ]
2019-05-22 21:16:55 +00:00
}
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
2020-12-08 18:49:32 +00:00
var pushCtxPoolCh = make ( chan * pushCtx , cgroup . AvailableCPUs ( ) )