2019-05-22 23:23:23 +02:00
package main
2019-05-22 23:16:55 +02:00
import (
"flag"
2019-08-23 08:46:45 +02:00
"fmt"
2019-05-22 23:16:55 +02:00
"net/http"
"runtime"
"strings"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/prometheus"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
2019-05-22 23:23:23 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
2019-05-22 23:16:55 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
2019-05-22 23:23:23 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
2019-05-28 16:17:19 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
2019-05-22 23:16:55 +02:00
"github.com/VictoriaMetrics/metrics"
)
var (
2019-05-22 23:23:23 +02:00
httpListenAddr = flag . String ( "httpListenAddr" , ":8481" , "Address to listen for http connections" )
cacheDataPath = flag . String ( "cacheDataPath" , "" , "Path to directory for cache files. Cache isn't saved if empty" )
2020-01-17 14:43:47 +01:00
maxConcurrentRequests = flag . Int ( "search.maxConcurrentRequests" , getDefaultMaxConcurrentRequests ( ) , "The maximum number of concurrent search requests. " +
"It shouldn't be high, since a single request can saturate all the CPU cores. See also `-search.maxQueueDuration`" )
maxQueueDuration = flag . Duration ( "search.maxQueueDuration" , 10 * time . Second , "The maximum time the request waits for execution when -search.maxConcurrentRequests limit is reached" )
storageNodes = flagutil . NewArray ( "storageNode" , "Addresses of vmstorage nodes; usage: -storageNode=vmstorage-host1:8401 -storageNode=vmstorage-host2:8401" )
2019-05-22 23:16:55 +02:00
)
2020-01-17 14:43:47 +01:00
func getDefaultMaxConcurrentRequests ( ) int {
n := runtime . GOMAXPROCS ( - 1 )
if n <= 4 {
n *= 2
}
if n > 16 {
// A single request can saturate all the CPU cores, so there is no sense
// in allowing higher number of concurrent requests - they will just contend
// for unavailable CPU time.
n = 16
}
return n
}
2019-05-22 23:23:23 +02:00
func main ( ) {
flag . Parse ( )
buildinfo . Init ( )
logger . Init ( )
2019-07-20 09:21:59 +02:00
logger . Infof ( "starting netstorage at storageNodes %s" , * storageNodes )
2019-05-22 23:23:23 +02:00
startTime := time . Now ( )
2019-06-18 09:26:44 +02:00
if len ( * storageNodes ) == 0 {
2019-07-20 09:21:59 +02:00
logger . Fatalf ( "missing -storageNode arg" )
2019-05-22 23:23:23 +02:00
}
2019-06-18 09:26:44 +02:00
netstorage . InitStorageNodes ( * storageNodes )
2019-05-22 23:23:23 +02:00
logger . Infof ( "started netstorage in %s" , time . Since ( startTime ) )
if len ( * cacheDataPath ) > 0 {
tmpDataPath := * cacheDataPath + "/tmp"
fs . RemoveDirContents ( tmpDataPath )
netstorage . InitTmpBlocksDir ( tmpDataPath )
promql . InitRollupResultCache ( * cacheDataPath + "/rollupResult" )
} else {
netstorage . InitTmpBlocksDir ( "" )
promql . InitRollupResultCache ( "" )
}
2019-05-22 23:16:55 +02:00
concurrencyCh = make ( chan struct { } , * maxConcurrentRequests )
2019-05-22 23:23:23 +02:00
go func ( ) {
httpserver . Serve ( * httpListenAddr , requestHandler )
} ( )
2019-05-22 23:16:55 +02:00
2019-05-22 23:23:23 +02:00
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 %s" , time . Since ( startTime ) )
logger . Infof ( "shutting down neststorage..." )
startTime = time . Now ( )
netstorage . Stop ( )
if len ( * cacheDataPath ) > 0 {
promql . StopRollupResultCache ( )
}
logger . Infof ( "successfully stopped netstorage in %s" , time . Since ( startTime ) )
2019-11-12 15:29:43 +01:00
fs . MustStopDirRemover ( )
2019-05-22 23:23:23 +02:00
logger . Infof ( "the vmselect has been stopped" )
2019-05-22 23:16:55 +02:00
}
2019-05-22 23:23:23 +02:00
var concurrencyCh chan struct { }
2019-08-05 17:27:50 +02:00
var (
concurrencyLimitReached = metrics . NewCounter ( ` vm_concurrent_select_limit_reached_total ` )
concurrencyLimitTimeout = metrics . NewCounter ( ` vm_concurrent_select_limit_timeout_total ` )
_ = metrics . NewGauge ( ` vm_concurrent_select_capacity ` , func ( ) float64 {
return float64 ( cap ( concurrencyCh ) )
} )
_ = metrics . NewGauge ( ` vm_concurrent_select_current ` , func ( ) float64 {
return float64 ( len ( concurrencyCh ) )
} )
)
2019-05-22 23:23:23 +02:00
func requestHandler ( w http . ResponseWriter , r * http . Request ) bool {
2019-05-22 23:16:55 +02:00
// Limit the number of concurrent queries.
select {
case concurrencyCh <- struct { } { } :
defer func ( ) { <- concurrencyCh } ( )
2019-08-05 17:27:50 +02:00
default :
// Sleep for a while until giving up. This should resolve short bursts in requests.
concurrencyLimitReached . Inc ( )
t := timerpool . Get ( * maxQueueDuration )
select {
case concurrencyCh <- struct { } { } :
timerpool . Put ( t )
defer func ( ) { <- concurrencyCh } ( )
case <- t . C :
timerpool . Put ( t )
concurrencyLimitTimeout . Inc ( )
2019-08-23 08:46:45 +02:00
err := & httpserver . ErrorWithStatusCode {
2020-01-17 12:24:37 +01:00
Err : fmt . Errorf ( "cannot handle more than %d concurrent search requests during %s; possible solutions: " +
"increase `-search.maxQueueDuration`, increase `-search.maxConcurrentRequests`, increase server capacity" ,
* maxConcurrentRequests , * maxQueueDuration ) ,
2019-08-23 08:46:45 +02:00
StatusCode : http . StatusServiceUnavailable ,
}
httpserver . Errorf ( w , "%s" , err )
2019-08-05 17:27:50 +02:00
return true
}
2019-05-22 23:16:55 +02:00
}
2019-05-22 23:23:23 +02:00
path := r . URL . Path
if path == "/internal/resetRollupResultCache" {
promql . ResetRollupResultCache ( )
return true
}
p , err := httpserver . ParsePath ( path )
if err != nil {
httpserver . Errorf ( w , "cannot parse path %q: %s" , path , err )
return true
}
at , err := auth . NewToken ( p . AuthToken )
if err != nil {
httpserver . Errorf ( w , "auth error: %s" , err )
return true
}
switch p . Prefix {
case "select" :
return selectHandler ( w , r , p , at )
case "delete" :
return deleteHandler ( w , r , p , at )
default :
// This is not our link
return false
}
}
func selectHandler ( w http . ResponseWriter , r * http . Request , p * httpserver . Path , at * auth . Token ) bool {
if strings . HasPrefix ( p . Suffix , "prometheus/api/v1/label/" ) {
s := p . Suffix [ len ( "prometheus/api/v1/label/" ) : ]
2019-05-22 23:16:55 +02:00
if strings . HasSuffix ( s , "/values" ) {
labelValuesRequests . Inc ( )
labelName := s [ : len ( s ) - len ( "/values" ) ]
httpserver . EnableCORS ( w , r )
2019-05-22 23:23:23 +02:00
if err := prometheus . LabelValuesHandler ( at , labelName , w , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
labelValuesErrors . Inc ( )
sendPrometheusError ( w , r , err )
return true
}
return true
}
}
2019-05-22 23:23:23 +02:00
switch p . Suffix {
case "prometheus/api/v1/query" :
2019-05-22 23:16:55 +02:00
queryRequests . Inc ( )
httpserver . EnableCORS ( w , r )
2019-05-22 23:23:23 +02:00
if err := prometheus . QueryHandler ( at , w , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
queryErrors . Inc ( )
sendPrometheusError ( w , r , err )
return true
}
return true
2019-05-22 23:23:23 +02:00
case "prometheus/api/v1/query_range" :
2019-05-22 23:16:55 +02:00
queryRangeRequests . Inc ( )
httpserver . EnableCORS ( w , r )
2019-05-22 23:23:23 +02:00
if err := prometheus . QueryRangeHandler ( at , w , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
queryRangeErrors . Inc ( )
sendPrometheusError ( w , r , err )
return true
}
return true
2019-05-22 23:23:23 +02:00
case "prometheus/api/v1/series" :
2019-05-22 23:16:55 +02:00
seriesRequests . Inc ( )
httpserver . EnableCORS ( w , r )
2019-05-22 23:23:23 +02:00
if err := prometheus . SeriesHandler ( at , w , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
seriesErrors . Inc ( )
sendPrometheusError ( w , r , err )
return true
}
return true
2019-05-22 23:23:23 +02:00
case "prometheus/api/v1/series/count" :
2019-05-22 23:16:55 +02:00
seriesCountRequests . Inc ( )
httpserver . EnableCORS ( w , r )
2019-05-22 23:23:23 +02:00
if err := prometheus . SeriesCountHandler ( at , w , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
seriesCountErrors . Inc ( )
sendPrometheusError ( w , r , err )
return true
}
return true
2019-05-22 23:23:23 +02:00
case "prometheus/api/v1/labels" :
2019-05-22 23:16:55 +02:00
labelsRequests . Inc ( )
httpserver . EnableCORS ( w , r )
2019-05-22 23:23:23 +02:00
if err := prometheus . LabelsHandler ( at , w , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
labelsErrors . Inc ( )
sendPrometheusError ( w , r , err )
return true
}
return true
2019-06-10 17:55:20 +02:00
case "prometheus/api/v1/labels/count" :
labelsCountRequests . Inc ( )
httpserver . EnableCORS ( w , r )
if err := prometheus . LabelsCountHandler ( at , w , r ) ; err != nil {
labelsCountErrors . Inc ( )
sendPrometheusError ( w , r , err )
return true
}
return true
2019-05-22 23:23:23 +02:00
case "prometheus/api/v1/export" :
2019-05-22 23:16:55 +02:00
exportRequests . Inc ( )
2019-05-22 23:23:23 +02:00
if err := prometheus . ExportHandler ( at , w , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
exportErrors . Inc ( )
httpserver . Errorf ( w , "error in %q: %s" , r . URL . Path , err )
return true
}
return true
2019-05-22 23:23:23 +02:00
case "prometheus/federate" :
2019-05-22 23:16:55 +02:00
federateRequests . Inc ( )
2019-05-22 23:23:23 +02:00
if err := prometheus . FederateHandler ( at , w , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
federateErrors . Inc ( )
2019-05-22 23:23:23 +02:00
httpserver . Errorf ( w , "error in %q: %s" , r . URL . Path , err )
2019-05-22 23:16:55 +02:00
return true
}
return true
2019-12-03 18:32:57 +01:00
case "prometheus/api/v1/rules" :
// Return dumb placeholder
rulesRequests . Inc ( )
w . Header ( ) . Set ( "Content-Type" , "application/json" )
fmt . Fprintf ( w , "%s" , ` { "status":"success","data": { "groups":[]}} ` )
return true
case "prometheus/api/v1/alerts" :
// Return dumb placehloder
alertsRequests . Inc ( )
w . Header ( ) . Set ( "Content-Type" , "application/json" )
fmt . Fprintf ( w , "%s" , ` { "status":"success","data": { "alerts":[]}} ` )
return true
2019-05-22 23:23:23 +02:00
default :
return false
}
}
func deleteHandler ( w http . ResponseWriter , r * http . Request , p * httpserver . Path , at * auth . Token ) bool {
switch p . Suffix {
case "prometheus/api/v1/admin/tsdb/delete_series" :
2019-05-22 23:16:55 +02:00
deleteRequests . Inc ( )
2019-05-22 23:23:23 +02:00
if err := prometheus . DeleteHandler ( at , r ) ; err != nil {
2019-05-22 23:16:55 +02:00
deleteErrors . Inc ( )
httpserver . Errorf ( w , "error in %q: %s" , r . URL . Path , err )
return true
}
w . WriteHeader ( http . StatusNoContent )
return true
default :
return false
}
}
func sendPrometheusError ( w http . ResponseWriter , r * http . Request , err error ) {
logger . Errorf ( "error in %q: %s" , r . URL . Path , err )
w . Header ( ) . Set ( "Content-Type" , "application/json" )
2019-08-23 08:46:45 +02:00
statusCode := http . StatusUnprocessableEntity
if esc , ok := err . ( * httpserver . ErrorWithStatusCode ) ; ok {
statusCode = esc . StatusCode
}
2019-05-22 23:16:55 +02:00
w . WriteHeader ( statusCode )
prometheus . WriteErrorResponse ( w , statusCode , err )
}
var (
2019-05-22 23:23:23 +02:00
labelValuesRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/label/ { }/values"} ` )
labelValuesErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="select/ { }/prometheus/api/v1/label/ { }/values"} ` )
2019-05-22 23:16:55 +02:00
2019-05-22 23:23:23 +02:00
queryRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/query"} ` )
queryErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/select/ { }/prometheus/api/v1/query"} ` )
2019-05-22 23:16:55 +02:00
2019-05-22 23:23:23 +02:00
queryRangeRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/prometheus/api/v1/query_range"} ` )
queryRangeErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/select/ { }/prometheus/api/v1/query_range"} ` )
2019-05-22 23:16:55 +02:00
2019-05-22 23:23:23 +02:00
seriesRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/series"} ` )
seriesErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/select/ { }/prometheus/api/v1/series"} ` )
2019-05-22 23:16:55 +02:00
2019-05-22 23:23:23 +02:00
seriesCountRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/series/count"} ` )
seriesCountErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/select/ { }/prometheus/api/v1/series/count"} ` )
2019-05-22 23:16:55 +02:00
2019-05-22 23:23:23 +02:00
labelsRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/labels"} ` )
labelsErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/select/ { }/prometheus/api/v1/labels"} ` )
2019-05-22 23:16:55 +02:00
2019-06-10 17:55:20 +02:00
labelsCountRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/labels/count"} ` )
labelsCountErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/select/ { }/prometheus/api/v1/labels/count"} ` )
2019-05-22 23:23:23 +02:00
deleteRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/delete/ { }/prometheus/api/v1/admin/tsdb/delete_series"} ` )
deleteErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/delete/ { }/prometheus/api/v1/admin/tsdb/delete_series"} ` )
2019-05-22 23:16:55 +02:00
2019-05-22 23:23:23 +02:00
exportRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/export"} ` )
exportErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/select/ { }/prometheus/api/v1/export"} ` )
2019-05-22 23:16:55 +02:00
2019-05-22 23:23:23 +02:00
federateRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/federate"} ` )
federateErrors = metrics . NewCounter ( ` vm_http_request_errors_total { path="/select/ { }/prometheus/federate"} ` )
2019-12-03 18:32:57 +01:00
rulesRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/rules"} ` )
alertsRequests = metrics . NewCounter ( ` vm_http_requests_total { path="/select/ { }/prometheus/api/v1/alerts"} ` )
2019-05-22 23:16:55 +02:00
)