mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2024-12-15 16:30:55 +01:00
505d359b39
* app/vminsert: allows parsing tenant id from labels it should help mitigate issues with vmagent's multiTenant mode, which works incorrectly at heavy load and it cannot handle more then 100 different tenants. This functional hidden with flag and do not change vminsert default behaviour https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2970 * Update docs/Cluster-VictoriaMetrics.md Co-authored-by: Roman Khavronenko <roman@victoriametrics.com> * wip * app/vminsert/netstorage: clean remaining labels in order to free up GC * docs/Cluster-VictoriaMetrics.md: typo fix * wip * wip Co-authored-by: Roman Khavronenko <roman@victoriametrics.com> Co-authored-by: Aliaksandr Valialkin <valyala@victoriametrics.com>
369 lines
15 KiB
Go
369 lines
15 KiB
Go
package main
|
|
|
|
import (
|
|
"flag"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/clusternative"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/csvimport"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/datadog"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/graphite"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/influx"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/native"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/netstorage"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/opentsdb"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/opentsdbhttp"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/prometheusimport"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/promremotewrite"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/vmimport"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/influxutils"
|
|
clusternativeserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/clusternative"
|
|
graphiteserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/graphite"
|
|
influxserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/influx"
|
|
opentsdbserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdb"
|
|
opentsdbhttpserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdbhttp"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
|
|
"github.com/VictoriaMetrics/metrics"
|
|
)
|
|
|
|
var (
|
|
clusternativeListenAddr = flag.String("clusternativeListenAddr", "", "TCP address to listen for data from other vminsert nodes in multi-level cluster setup. "+
|
|
"See https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#multi-level-cluster-setup . Usually :8400 must be set. Doesn't work if empty")
|
|
graphiteListenAddr = flag.String("graphiteListenAddr", "", "TCP and UDP address to listen for Graphite plaintext data. Usually :2003 must be set. Doesn't work if empty")
|
|
influxListenAddr = flag.String("influxListenAddr", "", "TCP and UDP address to listen for InfluxDB line protocol data. Usually :8089 must be set. Doesn't work if empty. "+
|
|
"This flag isn't needed when ingesting data over HTTP - just send it to http://<victoriametrics>:8428/write")
|
|
opentsdbListenAddr = flag.String("opentsdbListenAddr", "", "TCP and UDP address to listen for OpentTSDB metrics. "+
|
|
"Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. "+
|
|
"Usually :4242 must be set. Doesn't work if empty")
|
|
opentsdbHTTPListenAddr = flag.String("opentsdbHTTPListenAddr", "", "TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty")
|
|
httpListenAddr = flag.String("httpListenAddr", ":8480", "Address to listen for http connections")
|
|
maxLabelsPerTimeseries = flag.Int("maxLabelsPerTimeseries", 30, "The maximum number of labels accepted per time series. Superfluous labels are dropped. In this case the vm_metrics_with_dropped_labels_total metric at /metrics page is incremented")
|
|
maxLabelValueLen = flag.Int("maxLabelValueLen", 16*1024, "The maximum length of label values in the accepted time series. Longer label values are truncated. In this case the vm_too_long_label_values_total metric at /metrics page is incremented")
|
|
storageNodes = flagutil.NewArray("storageNode", "Comma-separated addresses of vmstorage nodes; usage: -storageNode=vmstorage-host1,...,vmstorage-hostN")
|
|
)
|
|
|
|
var (
|
|
clusternativeServer *clusternativeserver.Server
|
|
graphiteServer *graphiteserver.Server
|
|
influxServer *influxserver.Server
|
|
opentsdbServer *opentsdbserver.Server
|
|
opentsdbhttpServer *opentsdbhttpserver.Server
|
|
)
|
|
|
|
func main() {
|
|
// Write flags and help message to stdout, since it is easier to grep or pipe.
|
|
flag.CommandLine.SetOutput(os.Stdout)
|
|
flag.Usage = usage
|
|
envflag.Parse()
|
|
buildinfo.Init()
|
|
logger.Init()
|
|
pushmetrics.Init()
|
|
|
|
logger.Infof("initializing netstorage for storageNodes %s...", *storageNodes)
|
|
startTime := time.Now()
|
|
if len(*storageNodes) == 0 {
|
|
logger.Fatalf("missing -storageNode arg")
|
|
}
|
|
if duplicatedAddr := checkDuplicates(*storageNodes); duplicatedAddr != "" {
|
|
logger.Fatalf("found equal addresses of storage nodes in the -storageNodes flag: %q", duplicatedAddr)
|
|
}
|
|
hashSeed := uint64(0)
|
|
if *clusternativeListenAddr != "" {
|
|
// Use different hash seed for the second level of vminsert nodes in multi-level cluster setup.
|
|
// This should fix uneven distribution of time series among storage nodes.
|
|
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1672
|
|
hashSeed = 0xabcdef0123456789
|
|
}
|
|
netstorage.InitStorageNodes(*storageNodes, hashSeed)
|
|
logger.Infof("successfully initialized netstorage in %.3f seconds", time.Since(startTime).Seconds())
|
|
|
|
relabel.Init()
|
|
storage.SetMaxLabelsPerTimeseries(*maxLabelsPerTimeseries)
|
|
storage.SetMaxLabelValueLen(*maxLabelValueLen)
|
|
common.StartUnmarshalWorkers()
|
|
writeconcurrencylimiter.Init()
|
|
if len(*clusternativeListenAddr) > 0 {
|
|
clusternativeServer = clusternativeserver.MustStart(*clusternativeListenAddr, func(c net.Conn) error {
|
|
return clusternative.InsertHandler(c)
|
|
})
|
|
}
|
|
if len(*graphiteListenAddr) > 0 {
|
|
graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, func(r io.Reader) error {
|
|
return graphite.InsertHandler(nil, r)
|
|
})
|
|
}
|
|
if len(*influxListenAddr) > 0 {
|
|
influxServer = influxserver.MustStart(*influxListenAddr, func(r io.Reader) error {
|
|
return influx.InsertHandlerForReader(nil, r)
|
|
})
|
|
}
|
|
if len(*opentsdbListenAddr) > 0 {
|
|
opentsdbServer = opentsdbserver.MustStart(*opentsdbListenAddr, func(r io.Reader) error {
|
|
return opentsdb.InsertHandler(nil, r)
|
|
}, opentsdbhttp.InsertHandler)
|
|
}
|
|
if len(*opentsdbHTTPListenAddr) > 0 {
|
|
opentsdbhttpServer = opentsdbhttpserver.MustStart(*opentsdbHTTPListenAddr, opentsdbhttp.InsertHandler)
|
|
}
|
|
|
|
go func() {
|
|
httpserver.Serve(*httpListenAddr, requestHandler)
|
|
}()
|
|
|
|
sig := procutil.WaitForSigterm()
|
|
logger.Infof("service received signal %s", sig)
|
|
|
|
logger.Infof("gracefully shutting down http service at %q", *httpListenAddr)
|
|
startTime = time.Now()
|
|
if err := httpserver.Stop(*httpListenAddr); err != nil {
|
|
logger.Fatalf("cannot stop http service: %s", err)
|
|
}
|
|
logger.Infof("successfully shut down http service in %.3f seconds", time.Since(startTime).Seconds())
|
|
|
|
if len(*clusternativeListenAddr) > 0 {
|
|
clusternativeServer.MustStop()
|
|
}
|
|
if len(*graphiteListenAddr) > 0 {
|
|
graphiteServer.MustStop()
|
|
}
|
|
if len(*influxListenAddr) > 0 {
|
|
influxServer.MustStop()
|
|
}
|
|
if len(*opentsdbListenAddr) > 0 {
|
|
opentsdbServer.MustStop()
|
|
}
|
|
if len(*opentsdbHTTPListenAddr) > 0 {
|
|
opentsdbhttpServer.MustStop()
|
|
}
|
|
common.StopUnmarshalWorkers()
|
|
|
|
logger.Infof("shutting down neststorage...")
|
|
startTime = time.Now()
|
|
netstorage.Stop()
|
|
logger.Infof("successfully stopped netstorage in %.3f seconds", time.Since(startTime).Seconds())
|
|
|
|
fs.MustStopDirRemover()
|
|
|
|
logger.Infof("the vminsert has been stopped")
|
|
}
|
|
|
|
func requestHandler(w http.ResponseWriter, r *http.Request) bool {
|
|
startTime := time.Now()
|
|
defer requestDuration.UpdateDuration(startTime)
|
|
|
|
if r.URL.Path == "/" {
|
|
if r.Method != "GET" {
|
|
return false
|
|
}
|
|
fmt.Fprintf(w, "vminsert - a component of VictoriaMetrics cluster. See docs at https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html")
|
|
return true
|
|
}
|
|
p, err := httpserver.ParsePath(r.URL.Path)
|
|
if err != nil {
|
|
httpserver.Errorf(w, r, "cannot parse path %q: %s", r.URL.Path, err)
|
|
return true
|
|
}
|
|
if p.Prefix != "insert" {
|
|
// This is not our link.
|
|
return false
|
|
}
|
|
at, err := auth.NewToken(p.AuthToken)
|
|
if err != nil {
|
|
httpserver.Errorf(w, r, "auth error: %s", err)
|
|
return true
|
|
}
|
|
|
|
if strings.HasPrefix(p.Suffix, "datadog/") {
|
|
// Trim suffix from paths starting from /datadog/ in order to support legacy DataDog agent.
|
|
// See https://github.com/VictoriaMetrics/VictoriaMetrics/pull/2670
|
|
p.Suffix = strings.TrimSuffix(p.Suffix, "/")
|
|
}
|
|
switch p.Suffix {
|
|
case "prometheus/", "prometheus", "prometheus/api/v1/write":
|
|
prometheusWriteRequests.Inc()
|
|
if err := promremotewrite.InsertHandler(at, r); err != nil {
|
|
prometheusWriteErrors.Inc()
|
|
httpserver.Errorf(w, r, "%s", err)
|
|
return true
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return true
|
|
case "prometheus/api/v1/import":
|
|
vmimportRequests.Inc()
|
|
if err := vmimport.InsertHandler(at, r); err != nil {
|
|
vmimportErrors.Inc()
|
|
httpserver.Errorf(w, r, "%s", err)
|
|
return true
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return true
|
|
case "prometheus/api/v1/import/csv":
|
|
csvimportRequests.Inc()
|
|
if err := csvimport.InsertHandler(at, r); err != nil {
|
|
csvimportErrors.Inc()
|
|
httpserver.Errorf(w, r, "%s", err)
|
|
return true
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return true
|
|
case "prometheus/api/v1/import/prometheus":
|
|
prometheusimportRequests.Inc()
|
|
if err := prometheusimport.InsertHandler(at, r); err != nil {
|
|
prometheusimportErrors.Inc()
|
|
httpserver.Errorf(w, r, "%s", err)
|
|
return true
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return true
|
|
case "prometheus/api/v1/import/native":
|
|
nativeimportRequests.Inc()
|
|
if err := native.InsertHandler(at, r); err != nil {
|
|
nativeimportErrors.Inc()
|
|
httpserver.Errorf(w, r, "%s", err)
|
|
return true
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return true
|
|
case "influx/write", "influx/api/v2/write":
|
|
influxWriteRequests.Inc()
|
|
addInfluxResponseHeaders(w)
|
|
if err := influx.InsertHandlerForHTTP(at, r); err != nil {
|
|
influxWriteErrors.Inc()
|
|
httpserver.Errorf(w, r, "%s", err)
|
|
return true
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return true
|
|
case "influx/query":
|
|
influxQueryRequests.Inc()
|
|
addInfluxResponseHeaders(w)
|
|
influxutils.WriteDatabaseNames(w)
|
|
return true
|
|
case "datadog/api/v1/series":
|
|
datadogWriteRequests.Inc()
|
|
if err := datadog.InsertHandlerForHTTP(at, r); err != nil {
|
|
datadogWriteErrors.Inc()
|
|
httpserver.Errorf(w, r, "%s", err)
|
|
return true
|
|
}
|
|
// See https://docs.datadoghq.com/api/latest/metrics/#submit-metrics
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(202)
|
|
fmt.Fprintf(w, `{"status":"ok"}`)
|
|
return true
|
|
case "datadog/api/v1/validate":
|
|
datadogValidateRequests.Inc()
|
|
// See https://docs.datadoghq.com/api/latest/authentication/#validate-api-key
|
|
w.Header().Set("Content-Type", "application/json")
|
|
fmt.Fprintf(w, `{"valid":true}`)
|
|
return true
|
|
case "datadog/api/v1/check_run":
|
|
datadogCheckRunRequests.Inc()
|
|
// See https://docs.datadoghq.com/api/latest/service-checks/#submit-a-service-check
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(202)
|
|
fmt.Fprintf(w, `{"status":"ok"}`)
|
|
return true
|
|
case "datadog/intake":
|
|
datadogIntakeRequests.Inc()
|
|
w.Header().Set("Content-Type", "application/json")
|
|
fmt.Fprintf(w, `{}`)
|
|
return true
|
|
case "datadog/api/v1/metadata":
|
|
datadogMetadataRequests.Inc()
|
|
w.Header().Set("Content-Type", "application/json")
|
|
fmt.Fprintf(w, `{}`)
|
|
return true
|
|
default:
|
|
// This is not our link
|
|
return false
|
|
}
|
|
}
|
|
|
|
func addInfluxResponseHeaders(w http.ResponseWriter) {
|
|
// This is needed for some clients, which expect InfluxDB version header.
|
|
// See, for example, https://github.com/ntop/ntopng/issues/5449#issuecomment-1005347597
|
|
w.Header().Set("X-Influxdb-Version", "1.8.0")
|
|
}
|
|
|
|
var (
|
|
requestDuration = metrics.NewHistogram(`vminsert_request_duration_seconds`)
|
|
|
|
prometheusWriteRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/prometheus/", protocol="promremotewrite"}`)
|
|
prometheusWriteErrors = metrics.NewCounter(`vm_http_request_errors_total{path="/insert/{}/prometheus/", protocol="promremotewrite"}`)
|
|
|
|
vmimportRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/prometheus/api/v1/import", protocol="vmimport"}`)
|
|
vmimportErrors = metrics.NewCounter(`vm_http_request_errors_total{path="/insert/{}/prometheus/api/v1/import", protocol="vmimport"}`)
|
|
|
|
csvimportRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/prometheus/api/v1/import/csv", protocol="csvimport"}`)
|
|
csvimportErrors = metrics.NewCounter(`vm_http_request_errors_total{path="/insert/{}/prometheus/api/v1/import/csv", protocol="csvimport"}`)
|
|
|
|
prometheusimportRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/prometheus/api/v1/import/prometheus", protocol="prometheusimport"}`)
|
|
prometheusimportErrors = metrics.NewCounter(`vm_http_request_errors_total{path="/insert/{}/prometheus/api/v1/import/prometheus", protocol="prometheusimport"}`)
|
|
|
|
nativeimportRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/prometheus/api/v1/import/native", protocol="nativeimport"}`)
|
|
nativeimportErrors = metrics.NewCounter(`vm_http_request_errors_total{path="/insert/{}/prometheus/api/v1/import/native", protocol="nativeimport"}`)
|
|
|
|
influxWriteRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/influx/write", protocol="influx"}`)
|
|
influxWriteErrors = metrics.NewCounter(`vm_http_request_errors_total{path="/insert/{}/influx/write", protocol="influx"}`)
|
|
|
|
influxQueryRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/influx/query", protocol="influx"}`)
|
|
|
|
datadogWriteRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/datadog/api/v1/series", protocol="datadog"}`)
|
|
datadogWriteErrors = metrics.NewCounter(`vm_http_request_errors_total{path="/insert/{}/datadog/api/v1/series", protocol="datadog"}`)
|
|
|
|
datadogValidateRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/datadog/api/v1/validate", protocol="datadog"}`)
|
|
datadogCheckRunRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/datadog/api/v1/check_run", protocol="datadog"}`)
|
|
datadogIntakeRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/datadog/intake", protocol="datadog"}`)
|
|
datadogMetadataRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/datadog/api/v1/metadata", protocol="datadog"}`)
|
|
|
|
_ = metrics.NewGauge(`vm_metrics_with_dropped_labels_total`, func() float64 {
|
|
return float64(atomic.LoadUint64(&storage.MetricsWithDroppedLabels))
|
|
})
|
|
_ = metrics.NewGauge(`vm_too_long_label_names_total`, func() float64 {
|
|
return float64(atomic.LoadUint64(&storage.TooLongLabelNames))
|
|
})
|
|
_ = metrics.NewGauge(`vm_too_long_label_values_total`, func() float64 {
|
|
return float64(atomic.LoadUint64(&storage.TooLongLabelValues))
|
|
})
|
|
)
|
|
|
|
func usage() {
|
|
const s = `
|
|
vminsert accepts data via popular data ingestion protocols and routes it to vmstorage nodes configured via -storageNode.
|
|
|
|
See the docs at https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html .
|
|
`
|
|
flagutil.Usage(s)
|
|
}
|
|
|
|
func checkDuplicates(arr []string) string {
|
|
visited := make(map[string]struct{})
|
|
for _, s := range arr {
|
|
if _, ok := visited[s]; ok {
|
|
return s
|
|
}
|
|
visited[s] = struct{}{}
|
|
}
|
|
return ""
|
|
}
|