2020-05-05 09:53:42 +02:00
package main
import (
2023-09-08 00:46:34 +02:00
"context"
"errors"
2020-05-05 09:53:42 +02:00
"flag"
2021-09-14 11:17:49 +02:00
"fmt"
2023-01-27 22:38:13 +01:00
"io"
"net"
2020-05-05 09:53:42 +02:00
"net/http"
2023-01-27 22:38:13 +01:00
"net/textproto"
2023-01-27 23:06:42 +01:00
"net/url"
2020-05-16 10:59:30 +02:00
"os"
2024-03-06 16:35:38 +01:00
"slices"
2022-03-18 17:31:58 +01:00
"strings"
2021-11-09 18:18:27 +01:00
"sync"
2020-05-05 09:53:42 +02:00
"time"
2023-11-03 12:04:17 +01:00
"github.com/VictoriaMetrics/metrics"
2020-05-05 09:53:42 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
2023-01-27 22:38:13 +01:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
2020-05-05 09:53:42 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
2020-12-03 20:40:30 +01:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
2020-05-05 09:53:42 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
2023-01-27 22:38:13 +01:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
2020-05-05 09:53:42 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
2024-04-17 16:46:27 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
2022-07-21 18:58:22 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
2020-05-05 09:53:42 +02:00
)
var (
2024-02-09 02:15:04 +01:00
httpListenAddrs = flagutil . NewArrayString ( "httpListenAddr" , "TCP address to listen for incoming http requests. See also -tls and -httpListenAddr.useProxyProtocol" )
useProxyProtocol = flagutil . NewArrayBool ( "httpListenAddr.useProxyProtocol" , "Whether to use proxy protocol for connections accepted at the corresponding -httpListenAddr . " +
2023-03-08 10:26:53 +01:00
"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt . " +
"With enabled proxy protocol http server cannot serve regular /metrics endpoint. Use -pushmetrics.url for metrics pushing" )
2023-01-27 23:06:42 +01:00
maxIdleConnsPerBackend = flag . Int ( "maxIdleConnsPerBackend" , 100 , "The maximum number of idle connections vmauth can open per each backend host. " +
"See also -maxConcurrentRequests" )
responseTimeout = flag . Duration ( "responseTimeout" , 5 * time . Minute , "The timeout for receiving a response from backend" )
maxConcurrentRequests = flag . Int ( "maxConcurrentRequests" , 1000 , "The maximum number of concurrent requests vmauth can process. Other requests are rejected with " +
2023-02-11 06:57:49 +01:00
"'429 Too Many Requests' http status code. See also -maxConcurrentPerUserRequests and -maxIdleConnsPerBackend command-line options" )
maxConcurrentPerUserRequests = flag . Int ( "maxConcurrentPerUserRequests" , 300 , "The maximum number of concurrent requests vmauth can process per each configured user. " +
"Other requests are rejected with '429 Too Many Requests' http status code. See also -maxConcurrentRequests command-line option and max_concurrent_requests option " +
"in per-user config" )
2024-01-21 20:58:26 +01:00
reloadAuthKey = flagutil . NewPassword ( "reloadAuthKey" , "Auth key for /-/reload http endpoint. It must be passed as authKey=..." )
2023-01-27 23:06:42 +01:00
logInvalidAuthTokens = flag . Bool ( "logInvalidAuthTokens" , false , "Whether to log requests with invalid auth tokens. " +
2021-10-19 14:29:07 +02:00
` Such requests are always counted at vmauth_http_request_errors_total { reason="invalid_auth_token"} metric, which is exposed at /metrics page ` )
2023-09-08 00:46:34 +02:00
failTimeout = flag . Duration ( "failTimeout" , 3 * time . Second , "Sets a delay period for load balancing to skip a malfunctioning backend" )
maxRequestBodySizeToRetry = flagutil . NewBytes ( "maxRequestBodySizeToRetry" , 16 * 1024 , "The maximum request body size, which can be cached and re-tried at other backends. " +
"Bigger values may require more memory" )
2023-11-13 08:23:35 +01:00
backendTLSInsecureSkipVerify = flag . Bool ( "backend.tlsInsecureSkipVerify" , false , "Whether to skip TLS verification when connecting to backends over HTTPS. " +
"See https://docs.victoriametrics.com/vmauth.html#backend-tls-setup" )
backendTLSCAFile = flag . String ( "backend.TLSCAFile" , "" , "Optional path to TLS root CA file, which is used for TLS verification when connecting to backends over HTTPS. " +
"See https://docs.victoriametrics.com/vmauth.html#backend-tls-setup" )
2024-04-17 17:12:17 +02:00
backendTLSCertFile = flag . String ( "backend.TLSCertFile" , "" , "Optional path to TLS client certificate file, which must be sent to HTTPS backend. " +
"See https://docs.victoriametrics.com/vmauth.html#backend-tls-setup" )
backendTLSKeyFile = flag . String ( "backend.TLSKeyFile" , "" , "Optional path to TLS client key file, which must be sent to HTTPS backend. " +
"See https://docs.victoriametrics.com/vmauth.html#backend-tls-setup" )
backendTLSServerName = flag . String ( "backend.TLSServerName" , "" , "Optional TLS ServerName, which must be sent to HTTPS backend. " +
"See https://docs.victoriametrics.com/vmauth.html#backend-tls-setup" )
2020-05-05 09:53:42 +02:00
)
func main ( ) {
2020-05-16 10:59:30 +02:00
// Write flags and help message to stdout, since it is easier to grep or pipe.
flag . CommandLine . SetOutput ( os . Stdout )
2020-06-05 09:39:46 +02:00
flag . Usage = usage
2020-05-05 09:53:42 +02:00
envflag . Parse ( )
buildinfo . Init ( )
logger . Init ( )
2022-07-22 12:35:58 +02:00
2024-02-09 02:15:04 +01:00
listenAddrs := * httpListenAddrs
if len ( listenAddrs ) == 0 {
listenAddrs = [ ] string { ":8427" }
}
logger . Infof ( "starting vmauth at %q..." , listenAddrs )
2020-05-05 09:53:42 +02:00
startTime := time . Now ( )
initAuthConfig ( )
2024-02-09 02:15:04 +01:00
go httpserver . Serve ( listenAddrs , useProxyProtocol , requestHandler )
2020-05-05 09:53:42 +02:00
logger . Infof ( "started vmauth in %.3f seconds" , time . Since ( startTime ) . Seconds ( ) )
2024-01-15 12:37:02 +01:00
pushmetrics . Init ( )
2020-05-05 09:53:42 +02:00
sig := procutil . WaitForSigterm ( )
logger . Infof ( "received signal %s" , sig )
2024-01-15 12:37:02 +01:00
pushmetrics . Stop ( )
2020-05-05 09:53:42 +02:00
startTime = time . Now ( )
2024-02-09 02:15:04 +01:00
logger . Infof ( "gracefully shutting down webservice at %q" , listenAddrs )
if err := httpserver . Stop ( listenAddrs ) ; err != nil {
2020-05-05 09:53:42 +02:00
logger . Fatalf ( "cannot stop the webservice: %s" , err )
}
logger . Infof ( "successfully shut down the webservice in %.3f seconds" , time . Since ( startTime ) . Seconds ( ) )
stopAuthConfig ( )
logger . Infof ( "successfully stopped vmauth in %.3f seconds" , time . Since ( startTime ) . Seconds ( ) )
}
func requestHandler ( w http . ResponseWriter , r * http . Request ) bool {
2021-05-18 01:23:53 +02:00
switch r . URL . Path {
case "/-/reload" :
2024-01-21 20:58:26 +01:00
if ! httpserver . CheckAuthFlag ( w , r , reloadAuthKey . Get ( ) , "reloadAuthKey" ) {
2021-05-20 17:46:12 +02:00
return true
}
2021-05-18 01:23:53 +02:00
configReloadRequests . Inc ( )
procutil . SelfSIGHUP ( )
w . WriteHeader ( http . StatusOK )
return true
}
2024-02-12 23:57:53 +01:00
ats := getAuthTokensFromRequest ( r )
if len ( ats ) == 0 {
2023-04-24 14:57:13 +02:00
// Process requests for unauthorized users
ui := authConfig . Load ( ) . UnauthorizedUser
if ui != nil {
processUserRequest ( w , r , ui )
return true
}
2020-08-09 08:38:41 +02:00
w . Header ( ) . Set ( "WWW-Authenticate" , ` Basic realm="Restricted" ` )
2021-04-02 21:14:53 +02:00
http . Error ( w , "missing `Authorization` request header" , http . StatusUnauthorized )
2020-05-05 09:53:42 +02:00
return true
}
2023-01-27 23:06:42 +01:00
2024-02-12 23:57:53 +01:00
ui := getUserInfoByAuthTokens ( ats )
2021-04-02 21:14:53 +02:00
if ui == nil {
2021-09-14 11:17:49 +02:00
invalidAuthTokenRequests . Inc ( )
if * logInvalidAuthTokens {
2024-02-12 23:57:53 +01:00
err := fmt . Errorf ( "cannot authorize request with auth tokens %q" , ats )
2023-01-27 22:38:13 +01:00
err = & httpserver . ErrorWithStatusCode {
Err : err ,
StatusCode : http . StatusUnauthorized ,
}
httpserver . Errorf ( w , r , "%s" , err )
2021-09-14 11:17:49 +02:00
} else {
2023-05-17 09:09:47 +02:00
http . Error ( w , "Unauthorized" , http . StatusUnauthorized )
2021-09-14 11:17:49 +02:00
}
2020-05-05 09:53:42 +02:00
return true
}
2023-04-24 14:57:13 +02:00
processUserRequest ( w , r , ui )
return true
}
2024-02-12 23:57:53 +01:00
func getUserInfoByAuthTokens ( ats [ ] string ) * UserInfo {
ac := * authUsers . Load ( )
for _ , at := range ats {
ui := ac [ at ]
if ui != nil {
return ui
}
}
return nil
}
2023-04-24 14:57:13 +02:00
func processUserRequest ( w http . ResponseWriter , r * http . Request , ui * UserInfo ) {
2023-06-27 20:15:17 +02:00
startTime := time . Now ( )
defer ui . requestsDuration . UpdateDuration ( startTime )
2021-02-11 11:40:59 +01:00
ui . requests . Inc ( )
2023-01-27 22:38:13 +01:00
2023-01-27 23:06:42 +01:00
// Limit the concurrency of requests to backends
concurrencyLimitOnce . Do ( concurrencyLimitInit )
select {
case concurrencyLimitCh <- struct { } { } :
2023-02-10 05:03:01 +01:00
if err := ui . beginConcurrencyLimit ( ) ; err != nil {
handleConcurrencyLimitError ( w , r , err )
<- concurrencyLimitCh
2023-04-24 14:57:13 +02:00
return
2023-01-27 23:06:42 +01:00
}
2023-02-10 05:03:01 +01:00
default :
concurrentRequestsLimitReached . Inc ( )
err := fmt . Errorf ( "cannot serve more than -maxConcurrentRequests=%d concurrent requests" , cap ( concurrencyLimitCh ) )
handleConcurrencyLimitError ( w , r , err )
2023-04-24 14:57:13 +02:00
return
2023-01-27 23:06:42 +01:00
}
2023-02-10 06:05:13 +01:00
processRequest ( w , r , ui )
2023-02-10 05:03:01 +01:00
ui . endConcurrencyLimit ( )
2023-01-27 23:06:42 +01:00
<- concurrencyLimitCh
}
2023-02-10 06:05:13 +01:00
func processRequest ( w http . ResponseWriter , r * http . Request , ui * UserInfo ) {
u := normalizeURL ( r . URL )
2024-03-06 20:56:32 +01:00
up , hc := ui . getURLPrefixAndHeaders ( u , r . Header )
2023-04-26 11:04:35 +02:00
isDefault := false
if up == nil {
if ui . DefaultURL == nil {
2023-11-01 20:59:46 +01:00
// Authorization should be requested for http requests without credentials
// to a route that is not in the configuration for unauthorized user.
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5236
if ui . BearerToken == "" && ui . Username == "" && len ( * authUsers . Load ( ) ) > 0 {
w . Header ( ) . Set ( "WWW-Authenticate" , ` Basic realm="Restricted" ` )
http . Error ( w , "missing `Authorization` request header" , http . StatusUnauthorized )
return
}
missingRouteRequests . Inc ( )
2023-04-26 11:04:35 +02:00
httpserver . Errorf ( w , r , "missing route for %q" , u . String ( ) )
return
}
2023-12-08 22:27:53 +01:00
up , hc = ui . DefaultURL , ui . HeadersConf
2023-04-26 11:04:35 +02:00
isDefault = true
2023-02-10 06:05:13 +01:00
}
maxAttempts := up . getBackendsCount ( )
2023-09-08 00:46:34 +02:00
if maxAttempts > 1 {
r . Body = & readTrackingBody {
r : r . Body ,
}
}
2023-02-10 06:05:13 +01:00
for i := 0 ; i < maxAttempts ; i ++ {
2023-12-08 22:27:53 +01:00
bu := up . getBackendURL ( )
2023-04-26 11:04:35 +02:00
targetURL := bu . url
// Don't change path and add request_path query param for default route.
if isDefault {
query := targetURL . Query ( )
2023-11-29 19:48:48 +01:00
query . Set ( "request_path" , u . String ( ) )
2023-04-26 11:04:35 +02:00
targetURL . RawQuery = query . Encode ( )
} else { // Update path for regular routes.
2023-12-14 00:04:46 +01:00
targetURL = mergeURLs ( targetURL , u , up . dropSrcPathPrefixParts )
2023-04-26 11:04:35 +02:00
}
2024-01-21 03:40:52 +01:00
ok := tryProcessingRequest ( w , r , targetURL , hc , up . retryStatusCodes , ui )
2023-02-11 09:27:40 +01:00
bu . put ( )
if ok {
2023-02-10 06:05:13 +01:00
return
}
2023-02-11 09:27:40 +01:00
bu . setBroken ( )
2023-02-10 06:05:13 +01:00
}
2023-04-26 11:04:35 +02:00
err := & httpserver . ErrorWithStatusCode {
2023-02-10 06:05:13 +01:00
Err : fmt . Errorf ( "all the backends for the user %q are unavailable" , ui . name ( ) ) ,
StatusCode : http . StatusServiceUnavailable ,
}
httpserver . Errorf ( w , r , "%s" , err )
2024-01-21 03:40:52 +01:00
ui . backendErrors . Inc ( )
2023-02-10 06:05:13 +01:00
}
2024-01-21 03:40:52 +01:00
func tryProcessingRequest ( w http . ResponseWriter , r * http . Request , targetURL * url . URL , hc HeadersConf , retryStatusCodes [ ] int , ui * UserInfo ) bool {
2023-01-27 22:38:13 +01:00
// This code has been copied from net/http/httputil/reverseproxy.go
req := sanitizeRequestHeaders ( r )
req . URL = targetURL
2024-03-07 00:02:13 +01:00
if req . URL . Scheme == "https" {
// Override req.Host only for https requests, since https server verifies hostnames during TLS handshake,
// so it expects the targetURL.Host in the request.
// There is no need in overriding the req.Host for http requests, since it is expected that backend server
// may properly process queries with the original req.Host.
req . Host = targetURL . Host
}
2023-09-08 22:39:17 +02:00
updateHeadersByConfig ( req . Header , hc . RequestHeaders )
2024-04-17 16:46:27 +02:00
res , err := ui . rt . RoundTrip ( req )
2023-09-08 00:46:34 +02:00
rtb , rtbOK := req . Body . ( * readTrackingBody )
2023-01-27 22:38:13 +01:00
if err != nil {
2023-09-08 00:46:34 +02:00
if errors . Is ( err , context . Canceled ) || errors . Is ( err , context . DeadlineExceeded ) {
// Do not retry canceled or timed out requests
remoteAddr := httpserver . GetQuotedRemoteAddr ( r )
requestURI := httpserver . GetRequestURI ( r )
logger . Warnf ( "remoteAddr: %s; requestURI: %s; error when proxying response body from %s: %s" , remoteAddr , requestURI , targetURL , err )
2024-01-21 03:40:52 +01:00
if errors . Is ( err , context . DeadlineExceeded ) {
// Timed out request must be counted as errors, since this usually means that the backend is slow.
ui . backendErrors . Inc ( )
}
2023-09-08 00:46:34 +02:00
return true
}
if ! rtbOK || ! rtb . canRetry ( ) {
// Request body cannot be re-sent to another backend. Return the error to the client then.
2023-02-10 06:05:13 +01:00
err = & httpserver . ErrorWithStatusCode {
2024-01-26 20:39:55 +01:00
Err : fmt . Errorf ( "cannot proxy the request to %s: %w" , targetURL , err ) ,
2023-02-10 06:05:13 +01:00
StatusCode : http . StatusServiceUnavailable ,
}
httpserver . Errorf ( w , r , "%s" , err )
2024-01-21 03:40:52 +01:00
ui . backendErrors . Inc ( )
2023-02-10 06:05:13 +01:00
return true
2023-01-27 22:38:13 +01:00
}
2023-05-17 09:37:03 +02:00
// Retry the request if its body wasn't read yet. This usually means that the backend isn't reachable.
remoteAddr := httpserver . GetQuotedRemoteAddr ( r )
// NOTE: do not use httpserver.GetRequestURI
2023-09-08 00:46:34 +02:00
// it explicitly reads request body, which may fail retries.
2023-09-08 22:39:17 +02:00
logger . Warnf ( "remoteAddr: %s; requestURI: %s; retrying the request to %s because of response error: %s" , remoteAddr , req . URL , targetURL , err )
2023-09-08 00:46:34 +02:00
return false
}
2024-03-06 16:35:38 +01:00
if slices . Contains ( retryStatusCodes , res . StatusCode ) {
2024-01-26 20:39:55 +01:00
_ = res . Body . Close ( )
if ! rtbOK || ! rtb . canRetry ( ) {
// If we get an error from the retry_status_codes list, but cannot execute retry,
// we consider such a request an error as well.
err := & httpserver . ErrorWithStatusCode {
Err : fmt . Errorf ( "got response status code=%d from %s, but cannot retry the request on another backend, because the request has been already consumed" ,
res . StatusCode , targetURL ) ,
StatusCode : http . StatusServiceUnavailable ,
}
httpserver . Errorf ( w , r , "%s" , err )
ui . backendErrors . Inc ( )
return true
2024-01-25 14:04:20 +01:00
}
2024-01-26 20:39:55 +01:00
// Retry requests at other backends if it matches retryStatusCodes.
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4893
remoteAddr := httpserver . GetQuotedRemoteAddr ( r )
// NOTE: do not use httpserver.GetRequestURI
// it explicitly reads request body, which may fail retries.
logger . Warnf ( "remoteAddr: %s; requestURI: %s; retrying the request to %s because response status code=%d belongs to retry_status_codes=%d" ,
remoteAddr , req . URL , targetURL , res . StatusCode , retryStatusCodes )
return false
2023-01-27 22:38:13 +01:00
}
removeHopHeaders ( res . Header )
copyHeader ( w . Header ( ) , res . Header )
2023-09-08 22:39:17 +02:00
updateHeadersByConfig ( w . Header ( ) , hc . ResponseHeaders )
2023-01-27 22:38:13 +01:00
w . WriteHeader ( res . StatusCode )
copyBuf := copyBufPool . Get ( )
copyBuf . B = bytesutil . ResizeNoCopyNoOverallocate ( copyBuf . B , 16 * 1024 )
_ , err = io . CopyBuffer ( w , res . Body , copyBuf . B )
copyBufPool . Put ( copyBuf )
2024-01-26 20:39:55 +01:00
_ = res . Body . Close ( )
2023-01-27 22:38:13 +01:00
if err != nil && ! netutil . IsTrivialNetworkError ( err ) {
remoteAddr := httpserver . GetQuotedRemoteAddr ( r )
requestURI := httpserver . GetRequestURI ( r )
logger . Warnf ( "remoteAddr: %s; requestURI: %s; error when proxying response body from %s: %s" , remoteAddr , requestURI , targetURL , err )
2023-02-10 06:05:13 +01:00
return true
2021-10-22 18:08:06 +02:00
}
2023-02-10 06:05:13 +01:00
return true
2020-05-05 09:53:42 +02:00
}
2023-01-27 22:38:13 +01:00
var copyBufPool bytesutil . ByteBufferPool
func copyHeader ( dst , src http . Header ) {
for k , vv := range src {
for _ , v := range vv {
dst . Add ( k , v )
2021-06-11 11:50:22 +02:00
}
2023-01-27 22:38:13 +01:00
}
}
2024-04-17 14:03:15 +02:00
func updateHeadersByConfig ( headers http . Header , config [ ] * Header ) {
2023-08-31 14:26:51 +02:00
for _ , h := range config {
if h . Value == "" {
headers . Del ( h . Name )
} else {
headers . Set ( h . Name , h . Value )
}
}
}
2023-01-27 22:38:13 +01:00
func sanitizeRequestHeaders ( r * http . Request ) * http . Request {
// This code has been copied from net/http/httputil/reverseproxy.go
req := r . Clone ( r . Context ( ) )
removeHopHeaders ( req . Header )
if clientIP , _ , err := net . SplitHostPort ( req . RemoteAddr ) ; err == nil {
// If we aren't the first proxy retain prior
// X-Forwarded-For information as a comma+space
// separated list and fold multiple headers into one.
prior := req . Header [ "X-Forwarded-For" ]
if len ( prior ) > 0 {
clientIP = strings . Join ( prior , ", " ) + ", " + clientIP
}
req . Header . Set ( "X-Forwarded-For" , clientIP )
}
return req
}
func removeHopHeaders ( h http . Header ) {
// remove hop-by-hop headers listed in the "Connection" header of h.
// See RFC 7230, section 6.1
for _ , f := range h [ "Connection" ] {
for _ , sf := range strings . Split ( f , "," ) {
if sf = textproto . TrimString ( sf ) ; sf != "" {
h . Del ( sf )
}
}
}
// Remove hop-by-hop headers to the backend. Especially
// important is "Connection" because we want a persistent
// connection, regardless of what the client sent to us.
for _ , key := range hopHeaders {
h . Del ( key )
}
}
// Hop-by-hop headers. These are removed when sent to the backend.
// As of RFC 7230, hop-by-hop headers are required to appear in the
// Connection header field. These are the headers defined by the
// obsoleted RFC 2616 (section 13.5.1) and are used for backward
// compatibility.
var hopHeaders = [ ] string {
"Connection" ,
"Proxy-Connection" , // non-standard but still sent by libcurl and rejected by e.g. google
"Keep-Alive" ,
"Proxy-Authenticate" ,
"Proxy-Authorization" ,
"Te" , // canonicalized version of "TE"
"Trailer" , // not Trailers per URL above; https://www.rfc-editor.org/errata_search.php?eid=4522
"Transfer-Encoding" ,
"Upgrade" ,
2021-06-11 11:50:22 +02:00
}
2021-09-14 11:17:49 +02:00
var (
2021-10-19 14:29:07 +02:00
configReloadRequests = metrics . NewCounter ( ` vmauth_http_requests_total { path="/-/reload"} ` )
invalidAuthTokenRequests = metrics . NewCounter ( ` vmauth_http_request_errors_total { reason="invalid_auth_token"} ` )
missingRouteRequests = metrics . NewCounter ( ` vmauth_http_request_errors_total { reason="missing_route"} ` )
2021-09-14 11:17:49 +02:00
)
2021-05-18 01:23:53 +02:00
2024-04-17 17:12:17 +02:00
func newRoundTripper ( caFileOpt , certFileOpt , keyFileOpt , serverNameOpt string , insecureSkipVerifyP * bool ) ( http . RoundTripper , error ) {
caFile := * backendTLSCAFile
if caFileOpt != "" {
caFile = caFileOpt
}
certFile := * backendTLSCertFile
if certFileOpt != "" {
certFile = certFileOpt
}
keyFile := * backendTLSKeyFile
if keyFileOpt != "" {
keyFile = keyFileOpt
}
serverName := * backendTLSServerName
if serverNameOpt != "" {
serverName = serverNameOpt
}
2024-04-17 16:46:27 +02:00
insecureSkipVerify := * backendTLSInsecureSkipVerify
if p := insecureSkipVerifyP ; p != nil {
insecureSkipVerify = * p
2023-11-13 08:23:35 +01:00
}
2024-04-17 16:46:27 +02:00
opts := & promauth . Options {
TLSConfig : & promauth . TLSConfig {
CAFile : caFile ,
2024-04-17 17:12:17 +02:00
CertFile : certFile ,
KeyFile : keyFile ,
ServerName : serverName ,
InsecureSkipVerify : insecureSkipVerify ,
2024-04-17 16:46:27 +02:00
} ,
}
cfg , err := opts . NewConfig ( )
if err != nil {
return nil , fmt . Errorf ( "cannot initialize promauth.Config: %w" , err )
2023-11-13 08:23:35 +01:00
}
2023-01-27 22:38:13 +01:00
tr := http . DefaultTransport . ( * http . Transport ) . Clone ( )
tr . ResponseHeaderTimeout = * responseTimeout
// Automatic compression must be disabled in order to fix https://github.com/VictoriaMetrics/VictoriaMetrics/issues/535
tr . DisableCompression = true
tr . MaxIdleConnsPerHost = * maxIdleConnsPerBackend
if tr . MaxIdleConns != 0 && tr . MaxIdleConns < tr . MaxIdleConnsPerHost {
tr . MaxIdleConns = tr . MaxIdleConnsPerHost
2021-11-09 18:18:27 +01:00
}
2024-04-17 20:45:48 +02:00
tr . DialContext = netutil . DialMaybeSRV
2024-04-17 16:46:27 +02:00
rt := cfg . NewRoundTripper ( tr )
return rt , nil
2020-05-05 09:53:42 +02:00
}
2020-06-05 09:39:46 +02:00
2023-01-27 23:06:42 +01:00
var (
concurrencyLimitCh chan struct { }
concurrencyLimitOnce sync . Once
)
func concurrencyLimitInit ( ) {
concurrencyLimitCh = make ( chan struct { } , * maxConcurrentRequests )
_ = metrics . NewGauge ( "vmauth_concurrent_requests_capacity" , func ( ) float64 {
return float64 ( * maxConcurrentRequests )
} )
_ = metrics . NewGauge ( "vmauth_concurrent_requests_current" , func ( ) float64 {
return float64 ( len ( concurrencyLimitCh ) )
} )
}
2023-02-10 05:03:01 +01:00
var concurrentRequestsLimitReached = metrics . NewCounter ( "vmauth_concurrent_requests_limit_reached_total" )
2023-01-27 23:06:42 +01:00
2020-06-05 09:39:46 +02:00
func usage ( ) {
const s = `
vmauth authenticates and authorizes incoming requests and proxies them to VictoriaMetrics .
2021-04-20 19:16:17 +02:00
See the docs at https : //docs.victoriametrics.com/vmauth.html .
2020-06-05 09:39:46 +02:00
`
2020-12-03 20:40:30 +01:00
flagutil . Usage ( s )
2020-06-05 09:39:46 +02:00
}
2023-02-10 05:03:01 +01:00
func handleConcurrencyLimitError ( w http . ResponseWriter , r * http . Request , err error ) {
w . Header ( ) . Add ( "Retry-After" , "10" )
err = & httpserver . ErrorWithStatusCode {
Err : err ,
StatusCode : http . StatusTooManyRequests ,
}
httpserver . Errorf ( w , r , "%s" , err )
}
2023-05-17 09:19:33 +02:00
type readTrackingBody struct {
2023-09-08 00:46:34 +02:00
// r contains reader for initial data reading
r io . ReadCloser
// buf is a buffer for data read from r. Buf size is limited by maxRequestBodySizeToRetry.
// If more than maxRequestBodySizeToRetry is read from r, then cannotRetry is set to true.
buf [ ] byte
// cannotRetry is set to true when more than maxRequestBodySizeToRetry are read from r.
// In this case the read data cannot fit buf, so it cannot be re-read from buf.
cannotRetry bool
// bufComplete is set to true when buf contains complete request body read from r.
bufComplete bool
// needReadBuf is set to true when Read() must be performed from buf instead of r.
needReadBuf bool
// offset is an offset at buf for the next data read if needReadBuf is set to true.
offset int
2023-05-17 09:19:33 +02:00
}
// Read implements io.Reader interface
// tracks body reading requests
func ( rtb * readTrackingBody ) Read ( p [ ] byte ) ( int , error ) {
2023-09-08 00:46:34 +02:00
if rtb . needReadBuf {
if rtb . offset >= len ( rtb . buf ) {
return 0 , io . EOF
}
n := copy ( p , rtb . buf [ rtb . offset : ] )
rtb . offset += n
return n , nil
2023-05-17 09:19:33 +02:00
}
2023-09-08 00:46:34 +02:00
if rtb . r == nil {
return 0 , fmt . Errorf ( "cannot read data after closing the reader" )
}
n , err := rtb . r . Read ( p )
if rtb . cannotRetry {
return n , err
}
if len ( rtb . buf ) + n > maxRequestBodySizeToRetry . IntN ( ) {
rtb . cannotRetry = true
return n , err
}
rtb . buf = append ( rtb . buf , p [ : n ] ... )
if err == io . EOF {
rtb . bufComplete = true
}
return n , err
}
func ( rtb * readTrackingBody ) canRetry ( ) bool {
if rtb . cannotRetry {
return false
}
if len ( rtb . buf ) > 0 && ! rtb . needReadBuf {
return false
}
return true
2023-05-17 09:19:33 +02:00
}
2023-05-17 09:37:03 +02:00
// Close implements io.Closer interface.
2023-05-17 09:19:33 +02:00
func ( rtb * readTrackingBody ) Close ( ) error {
2023-09-08 00:46:34 +02:00
rtb . offset = 0
if rtb . bufComplete {
rtb . needReadBuf = true
2023-05-17 09:19:33 +02:00
}
2023-09-08 00:46:34 +02:00
// Close rtb.r only if the request body is completely read or if it is too big.
// http.Roundtrip performs body.Close call even without any Read calls,
// so this hack allows us to reuse request body.
if rtb . bufComplete || rtb . cannotRetry {
if rtb . r == nil {
return nil
}
err := rtb . r . Close ( )
rtb . r = nil
return err
}
2023-05-17 09:19:33 +02:00
return nil
}