package main import ( "flag" "fmt" "io" "net/http" "time" "github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/csvimport" "github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/graphite" "github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/influx" "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/promremotewrite" "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" 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/storage" "github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter" "github.com/VictoriaMetrics/metrics" ) var ( 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 Influx line protocol data. Usually :8189 must be set. Doesn't work if empty") 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. Superflouos labels are dropped") storageNodes = flagutil.NewArray("storageNode", "Address of vmstorage nodes; usage: -storageNode=vmstorage-host1:8400 -storageNode=vmstorage-host2:8400") ) var ( influxServer *influxserver.Server graphiteServer *graphiteserver.Server opentsdbServer *opentsdbserver.Server opentsdbhttpServer *opentsdbhttpserver.Server ) func main() { envflag.Parse() buildinfo.Init() logger.Init() logger.Infof("initializing netstorage for storageNodes %s...", *storageNodes) startTime := time.Now() if len(*storageNodes) == 0 { logger.Fatalf("missing -storageNode arg") } netstorage.InitStorageNodes(*storageNodes) logger.Infof("successfully initialized netstorage in %.3f seconds", time.Since(startTime).Seconds()) storage.SetMaxLabelsPerTimeseries(*maxLabelsPerTimeseries) writeconcurrencylimiter.Init() if len(*influxListenAddr) > 0 { influxServer = influxserver.MustStart(*influxListenAddr, func(r io.Reader) error { var at auth.Token // TODO: properly initialize auth token return influx.InsertHandlerForReader(&at, r) }) } if len(*graphiteListenAddr) > 0 { graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, func(r io.Reader) error { var at auth.Token // TODO: properly initialize auth token return graphite.InsertHandler(&at, r) }) } if len(*opentsdbListenAddr) > 0 { opentsdbServer = opentsdbserver.MustStart(*opentsdbListenAddr, func(r io.Reader) error { var at auth.Token // TODO: properly initialize auth token return opentsdb.InsertHandler(&at, 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 the service at %q", *httpListenAddr) startTime = time.Now() if err := httpserver.Stop(*httpListenAddr); err != nil { logger.Fatalf("cannot stop the service: %s", err) } logger.Infof("successfully shut down the service in %.3f seconds", time.Since(startTime).Seconds()) if len(*influxListenAddr) > 0 { influxServer.MustStop() } if len(*graphiteListenAddr) > 0 { graphiteServer.MustStop() } if len(*opentsdbListenAddr) > 0 { opentsdbServer.MustStop() } if len(*opentsdbHTTPListenAddr) > 0 { opentsdbhttpServer.MustStop() } 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 { p, err := httpserver.ParsePath(r.URL.Path) if err != nil { httpserver.Errorf(w, "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, "auth error: %s", err) return true } 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, "error in %q: %s", r.URL.Path, 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, "error in %q: %s", r.URL.Path, 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, "error in %q: %s", r.URL.Path, err) return true } w.WriteHeader(http.StatusNoContent) return true case "influx/write", "influx/api/v2/write": influxWriteRequests.Inc() if err := influx.InsertHandlerForHTTP(at, r); err != nil { influxWriteErrors.Inc() httpserver.Errorf(w, "error in %q: %s", r.URL.Path, err) return true } w.WriteHeader(http.StatusNoContent) return true case "influx/query": // Emulate fake response for influx query. // This is required for TSBS benchmark. influxQueryRequests.Inc() fmt.Fprintf(w, `{"results":[{"series":[{"values":[]}]}]}`) return true default: // This is not our link return false } } var ( 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"}`) influxWriteRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/influx/", protocol="influx"}`) influxWriteErrors = metrics.NewCounter(`vm_http_request_errors_total{path="/insert/{}/influx/", protocol="influx"}`) influxQueryRequests = metrics.NewCounter(`vm_http_requests_total{path="/insert/{}/influx/query", protocol="influx"}`) )