return 1
}
+ maxReqs := cluster.API.MaxConcurrentRequests
+ if maxRails := cluster.API.MaxConcurrentRailsRequests; maxRails > 0 &&
+ (maxRails < maxReqs || maxReqs == 0) &&
+ strings.HasSuffix(prog, "controller") {
+ // 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
+ }
instrumented := httpserver.Instrument(reg, log,
httpserver.HandlerWithDeadline(cluster.API.RequestTimeout.Duration(),
httpserver.AddRequestIDs(
interceptHealthReqs(cluster.ManagementToken, handler.CheckHealth,
&httpserver.RequestLimiter{
Handler: handler,
- MaxConcurrent: cluster.API.MaxConcurrentRequests,
+ MaxConcurrent: maxReqs,
MaxQueue: cluster.API.MaxQueuedRequests,
MaxQueueTimeForMinPriority: cluster.API.MaxQueueTimeForLockRequests.Duration(),
Priority: c.requestPriority,
<-handler.Done()
srv.Close()
}()
- go c.requestQueueDumpCheck(cluster, prog, reg, &srv.Server, logger)
+ go c.requestQueueDumpCheck(cluster, maxReqs, prog, reg, &srv.Server, logger)
err = srv.Wait()
if err != nil {
return 1
// 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.
-func (c *command) requestQueueDumpCheck(cluster *arvados.Cluster, prog string, reg *prometheus.Registry, srv *http.Server, logger logrus.FieldLogger) {
+func (c *command) requestQueueDumpCheck(cluster *arvados.Cluster, maxReqs int, 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 == "" || maxReqs < 1 {
return
}
logger = logger.WithField("worker", "RequestQueueDump")
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 {
+ if n > 0 && n >= maxReqs*9/10 {
dump = true
break
}
contextKey key = iota
)
-func (*Suite) TestGetListenAddress(c *check.C) {
+func unusedPort(c *check.C) string {
// Find an available port on the testing host, so the test
// cases don't get confused by "already in use" errors.
listener, err := net.Listen("tcp", ":")
c.Assert(err, check.IsNil)
- _, unusedPort, err := net.SplitHostPort(listener.Addr().String())
- c.Assert(err, check.IsNil)
listener.Close()
+ _, port, err := net.SplitHostPort(listener.Addr().String())
+ c.Assert(err, check.IsNil)
+ return port
+}
+func (*Suite) TestGetListenAddress(c *check.C) {
+ port := unusedPort(c)
defer os.Unsetenv("ARVADOS_SERVICE_INTERNAL_URL")
for idx, trial := range []struct {
// internalURL => listenURL, both with trailing "/"
expectInternal string
}{
{
- internalURLs: map[string]string{"http://localhost:" + unusedPort + "/": ""},
- expectListen: "http://localhost:" + unusedPort + "/",
- expectInternal: "http://localhost:" + unusedPort + "/",
+ internalURLs: map[string]string{"http://localhost:" + port + "/": ""},
+ expectListen: "http://localhost:" + port + "/",
+ expectInternal: "http://localhost:" + port + "/",
},
{ // implicit port 80 in InternalURLs
internalURLs: map[string]string{"http://localhost/": ""},
expectErrorMatch: `.*:80: bind: permission denied`,
},
{ // implicit port 443 in InternalURLs
- internalURLs: map[string]string{"https://host.example/": "http://localhost:" + unusedPort + "/"},
- expectListen: "http://localhost:" + unusedPort + "/",
+ internalURLs: map[string]string{"https://host.example/": "http://localhost:" + port + "/"},
+ expectListen: "http://localhost:" + port + "/",
expectInternal: "https://host.example/",
},
{ // implicit port 443 in ListenURL
{
internalURLs: map[string]string{
"https://hostname1.example/": "http://localhost:12435/",
- "https://hostname2.example/": "http://localhost:" + unusedPort + "/",
+ "https://hostname2.example/": "http://localhost:" + port + "/",
},
envVar: "https://hostname2.example", // note this works despite missing trailing "/"
- expectListen: "http://localhost:" + unusedPort + "/",
+ expectListen: "http://localhost:" + port + "/",
expectInternal: "https://hostname2.example/",
},
{ // cannot listen on any of the ListenURLs
internalURLs: map[string]string{
- "https://hostname1.example/": "http://1.2.3.4:" + unusedPort + "/",
- "https://hostname2.example/": "http://1.2.3.4:" + unusedPort + "/",
+ "https://hostname1.example/": "http://1.2.3.4:" + port + "/",
+ "https://hostname2.example/": "http://1.2.3.4:" + port + "/",
},
expectErrorMatch: "configuration does not enable the \"arvados-controller\" service on this host",
},
c.Check(stderr.String(), check.Matches, `(?ms).*"msg":"CheckHealth called".*`)
}
-func (*Suite) TestDumpRequests(c *check.C) {
+func (s *Suite) TestDumpRequestsKeepweb(c *check.C) {
+ s.testDumpRequests(c, arvados.ServiceNameKeepweb, "MaxConcurrentRequests")
+}
+
+func (s *Suite) TestDumpRequestsController(c *check.C) {
+ s.testDumpRequests(c, arvados.ServiceNameController, "MaxConcurrentRailsRequests")
+}
+
+func (*Suite) testDumpRequests(c *check.C, serviceName arvados.ServiceName, maxReqsConfigKey string) {
defer func(orig time.Duration) { requestQueueDumpCheckInterval = orig }(requestQueueDumpCheckInterval)
requestQueueDumpCheckInterval = time.Second / 10
+ port := unusedPort(c)
tmpdir := c.MkDir()
cf, err := ioutil.TempFile(tmpdir, "cmd_test.")
c.Assert(err, check.IsNil)
SystemRootToken: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
ManagementToken: bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
API:
- MaxConcurrentRequests: %d
+ `+maxReqsConfigKey+`: %d
MaxQueuedRequests: 0
SystemLogs: {RequestQueueDumpDirectory: %q}
Services:
Controller:
- ExternalURL: "http://localhost:12345"
- InternalURLs: {"http://localhost:12345": {}}
+ ExternalURL: "http://localhost:`+port+`"
+ InternalURLs: {"http://localhost:`+port+`": {}}
+ WebDAV:
+ ExternalURL: "http://localhost:`+port+`"
+ InternalURLs: {"http://localhost:`+port+`": {}}
`, max, tmpdir)
+ cf.Close()
started := make(chan bool, max+1)
hold := make(chan bool)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
- cmd := Command(arvados.ServiceNameController, func(ctx context.Context, _ *arvados.Cluster, token string, reg *prometheus.Registry) Handler {
+ cmd := Command(serviceName, func(ctx context.Context, _ *arvados.Cluster, token string, reg *prometheus.Registry) Handler {
return &testHandler{ctx: ctx, handler: handler, healthCheck: healthCheck}
})
cmd.(*command).ctx = context.WithValue(ctx, contextKey, "bar")
var stdin, stdout, stderr bytes.Buffer
go func() {
- cmd.RunCommand("arvados-controller", []string{"-config", cf.Name()}, &stdin, &stdout, &stderr)
+ cmd.RunCommand(string(serviceName), []string{"-config", cf.Name()}, &stdin, &stdout, &stderr)
close(exited)
}()
select {
deadline := time.Now().Add(time.Second * 2)
for i := 0; i < max+1; i++ {
go func() {
- resp, err := client.Get("http://localhost:12345/testpath")
+ resp, err := client.Get("http://localhost:" + port + "/testpath")
for err != nil && strings.Contains(err.Error(), "dial tcp") && deadline.After(time.Now()) {
time.Sleep(time.Second / 100)
- resp, err = client.Get("http://localhost:12345/testpath")
+ resp, err = client.Get("http://localhost:" + port + "/testpath")
}
if c.Check(err, check.IsNil) {
c.Logf("resp StatusCode %d", resp.StatusCode)
case <-started:
case <-time.After(time.Second):
c.Logf("%s", stderr.String())
- panic("timed out")
+ c.Fatal("timed out")
}
}
for delay := time.Second / 100; ; delay = delay * 2 {
time.Sleep(delay)
- j, err := os.ReadFile(tmpdir + "/arvados-controller-requests.json")
+ j, err := os.ReadFile(tmpdir + "/" + string(serviceName) + "-requests.json")
if os.IsNotExist(err) && deadline.After(time.Now()) {
continue
}
}
for _, path := range []string{"/_inspect/requests", "/metrics"} {
- req, err := http.NewRequest("GET", "http://localhost:12345"+path, nil)
+ req, err := http.NewRequest("GET", "http://localhost:"+port+""+path, nil)
c.Assert(err, check.IsNil)
req.Header.Set("Authorization", "Bearer bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
resp, err := client.Do(req)
}
close(hold)
cancel()
-
}
func (*Suite) TestTLS(c *check.C) {
+ port := unusedPort(c)
cwd, err := os.Getwd()
c.Assert(err, check.IsNil)
SystemRootToken: abcde
Services:
Controller:
- ExternalURL: "https://localhost:12345"
- InternalURLs: {"https://localhost:12345": {}}
+ ExternalURL: "https://localhost:` + port + `"
+ InternalURLs: {"https://localhost:` + port + `": {}}
TLS:
Key: file://` + cwd + `/../../services/api/tmp/self-signed.key
Certificate: file://` + cwd + `/../../services/api/tmp/self-signed.pem
defer close(got)
client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}}
for range time.NewTicker(time.Millisecond).C {
- resp, err := client.Get("https://localhost:12345")
+ resp, err := client.Get("https://localhost:" + port)
if err != nil {
c.Log(err)
continue
projects / arvados.git / commitdiff