_ "net/http/pprof"
"net/url"
"os"
+ "regexp"
"strings"
"time"
loader := config.NewLoader(stdin, log)
loader.SetupFlags(flags)
- // prog is [keepstore, keep-web, git-httpd, ...] but the
+ // prog is [keepstore, keep-web, ...] but the
// legacy config flags are [-legacy-keepstore-config,
- // -legacy-keepweb-config, -legacy-git-httpd-config, ...]
+ // -legacy-keepweb-config, ...]
legacyFlag := "-legacy-" + strings.Replace(prog, "keep-", "keep", 1) + "-config"
args = loader.MungeLegacyConfigArgs(log, args, legacyFlag)
})
ctx := ctxlog.Context(c.ctx, logger)
+ // Check whether the caller is attempting to use environment
+ // variables to override cluster configuration, and advise
+ // that won't work.
+ {
+ envhost := os.Getenv("ARVADOS_API_HOST")
+ if envhost != "" && envhost != cluster.Services.Controller.ExternalURL.Host {
+ logger.Warn("ARVADOS_API_HOST environment variable is present, but will not be used")
+ }
+ envins := os.Getenv("ARVADOS_API_HOST_INSECURE")
+ if envins != "" && (envins != "0") != cluster.TLS.Insecure {
+ logger.Warn("ARVADOS_API_HOST_INSECURE environment variable is present, but will not be used")
+ }
+ envtoken := os.Getenv("ARVADOS_API_TOKEN")
+ if envtoken != "" && envtoken != cluster.SystemRootToken {
+ logger.Warn("ARVADOS_API_TOKEN environment variable is present, but will not be used")
+ }
+ }
+
listenURL, internalURL, err := getListenAddr(cluster.Services, c.svcName, log)
if err != nil {
return 1
httpserver.Inspect(reg, cluster.ManagementToken,
httpserver.LogRequests(
interceptHealthReqs(cluster.ManagementToken, handler.CheckHealth,
- &httpserver.RequestLimiter{
- Handler: handler,
- MaxConcurrent: cluster.API.MaxConcurrentRequests,
- MaxQueue: cluster.API.MaxQueuedRequests,
- Priority: c.requestPriority,
- Registry: reg}))))))
+ c.requestLimiter(handler, cluster, reg)))))))
srv := &httpserver.Server{
Server: http.Server{
Handler: ifCollectionInHost(instrumented, instrumented.ServeAPI(cluster.ManagementToken, instrumented)),
}
// If SystemLogs.RequestQueueDumpDirectory is set, monitor the
-// server's incoming HTTP request queue size. When it exceeds 90% of
-// API.MaxConcurrentRequests, write the /_inspect/requests data to a
-// JSON file in the specified directory.
+// server's incoming HTTP request limiters. When the number of
+// concurrent requests in any queue ("api" or "tunnel") exceeds 90% of
+// its maximum slots, write the /_inspect/requests data to a JSON file
+// in the specified directory.
func (c *command) requestQueueDumpCheck(cluster *arvados.Cluster, prog string, reg *prometheus.Registry, srv *http.Server, logger logrus.FieldLogger) {
outdir := cluster.SystemLogs.RequestQueueDumpDirectory
- if outdir == "" || cluster.ManagementToken == "" || cluster.API.MaxConcurrentRequests < 1 {
+ if outdir == "" || cluster.ManagementToken == "" {
return
}
logger = logger.WithField("worker", "RequestQueueDump")
logger.WithError(err).Warn("error getting metrics")
continue
}
- dump := false
+ cur := map[string]int{} // queue label => current
+ max := map[string]int{} // queue label => max
for _, mf := range mfs {
- if mf.Name != nil && *mf.Name == "arvados_concurrent_requests" && len(mf.Metric) == 1 {
- n := int(mf.Metric[0].GetGauge().GetValue())
- if n > 0 && n >= cluster.API.MaxConcurrentRequests*9/10 {
- dump = true
- break
+ for _, m := range mf.GetMetric() {
+ for _, ml := range m.GetLabel() {
+ if ml.GetName() == "queue" {
+ n := int(m.GetGauge().GetValue())
+ if name := mf.GetName(); name == "arvados_concurrent_requests" {
+ cur[*ml.Value] = n
+ } else if name == "arvados_max_concurrent_requests" {
+ max[*ml.Value] = n
+ }
+ }
}
}
}
+ dump := false
+ for queue, n := range cur {
+ if n > 0 && max[queue] > 0 && n >= max[queue]*9/10 {
+ dump = true
+ break
+ }
+ }
if dump {
req, err := http.NewRequest("GET", "/_inspect/requests", nil)
if err != nil {
}
}
+// Set up a httpserver.RequestLimiter with separate queues/streams for
+// API requests (obeying MaxConcurrentRequests etc) and gateway tunnel
+// requests (obeying MaxGatewayTunnels).
+func (c *command) requestLimiter(handler http.Handler, cluster *arvados.Cluster, reg *prometheus.Registry) http.Handler {
+ maxReqs := cluster.API.MaxConcurrentRequests
+ if maxRails := cluster.API.MaxConcurrentRailsRequests; maxRails > 0 &&
+ (maxRails < maxReqs || maxReqs == 0) &&
+ c.svcName == arvados.ServiceNameController {
+ // Ideally, we would accept up to
+ // MaxConcurrentRequests, and apply the
+ // MaxConcurrentRailsRequests limit only for requests
+ // that require calling upstream to RailsAPI. But for
+ // now we make the simplifying assumption that every
+ // controller request causes an upstream RailsAPI
+ // request.
+ maxReqs = maxRails
+ }
+ rqAPI := &httpserver.RequestQueue{
+ Label: "api",
+ MaxConcurrent: maxReqs,
+ MaxQueue: cluster.API.MaxQueuedRequests,
+ MaxQueueTimeForMinPriority: cluster.API.MaxQueueTimeForLockRequests.Duration(),
+ }
+ rqTunnel := &httpserver.RequestQueue{
+ Label: "tunnel",
+ MaxConcurrent: cluster.API.MaxGatewayTunnels,
+ MaxQueue: 0,
+ }
+ return &httpserver.RequestLimiter{
+ Handler: handler,
+ Priority: c.requestPriority,
+ Registry: reg,
+ Queue: func(req *http.Request) *httpserver.RequestQueue {
+ if req.Method == http.MethodPost && reTunnelPath.MatchString(req.URL.Path) {
+ return rqTunnel
+ } else {
+ return rqAPI
+ }
+ },
+ }
+}
+
+// reTunnelPath matches paths of API endpoints that go in the "tunnel"
+// queue.
+var reTunnelPath = regexp.MustCompile(func() string {
+ rePathVar := regexp.MustCompile(`{.*?}`)
+ out := ""
+ for _, endpoint := range []arvados.APIEndpoint{
+ arvados.EndpointContainerGatewayTunnel,
+ arvados.EndpointContainerGatewayTunnelCompat,
+ arvados.EndpointContainerSSH,
+ arvados.EndpointContainerSSHCompat,
+ } {
+ if out != "" {
+ out += "|"
+ }
+ out += `\Q/` + rePathVar.ReplaceAllString(endpoint.Path, `\E[^/]*\Q`) + `\E`
+ }
+ return "^(" + out + ")$"
+}())
+
func (c *command) requestPriority(req *http.Request, queued time.Time) int64 {
switch {
case req.Method == http.MethodPost && strings.HasPrefix(req.URL.Path, "/arvados/v1/containers/") && strings.HasSuffix(req.URL.Path, "/lock"):
// Return 503 immediately instead of queueing. We want
// to send feedback to dispatchcloud ASAP to stop
// bringing up new containers.
- return httpserver.IneligibleForQueuePriority
- case req.Method == http.MethodPost && strings.HasPrefix(req.URL.Path, "/arvados/v1/logs"):
- // "Create log entry" is the most harmless kind of
- // request to drop.
- return 0
+ return httpserver.MinPriority
case req.Header.Get("Origin") != "":
- // Handle interactive requests first.
- return 2
- default:
+ // Handle interactive requests first. Positive
+ // priority is called "high" in aggregate metrics.
return 1
+ default:
+ // Zero priority is called "normal" in aggregate
+ // metrics.
+ return 0
}
}
projects / arvados.git / blobdiff