+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)
+ defer os.Remove(cf.Name())
+ defer cf.Close()
+
+ max := 24
+ fmt.Fprintf(cf, `
+Clusters:
+ zzzzz:
+ SystemRootToken: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
+ ManagementToken: bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
+ API:
+ `+maxReqsConfigKey+`: %d
+ MaxQueuedRequests: 0
+ SystemLogs: {RequestQueueDumpDirectory: %q}
+ Services:
+ Controller:
+ 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)
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ started <- true
+ <-hold
+ })
+ healthCheck := make(chan bool, 1)
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ 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")
+
+ exited := make(chan bool)
+ var stdin, stdout, stderr bytes.Buffer
+
+ go func() {
+ cmd.RunCommand(string(serviceName), []string{"-config", cf.Name()}, &stdin, &stdout, &stderr)
+ close(exited)
+ }()
+ select {
+ case <-healthCheck:
+ case <-exited:
+ c.Logf("%s", stderr.String())
+ c.Error("command exited without health check")
+ }
+ client := http.Client{}
+ deadline := time.Now().Add(time.Second * 2)
+ for i := 0; i < max+1; i++ {
+ go func() {
+ 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:" + port + "/testpath")
+ }
+ if c.Check(err, check.IsNil) {
+ c.Logf("resp StatusCode %d", resp.StatusCode)
+ }
+ }()
+ }
+ for i := 0; i < max; i++ {
+ select {
+ case <-started:
+ case <-time.After(time.Second):
+ c.Logf("%s", stderr.String())
+ c.Fatal("timed out")
+ }
+ }
+ for delay := time.Second / 100; ; delay = delay * 2 {
+ time.Sleep(delay)
+ j, err := os.ReadFile(tmpdir + "/" + string(serviceName) + "-requests.json")
+ if os.IsNotExist(err) && deadline.After(time.Now()) {
+ continue
+ }
+ c.Assert(err, check.IsNil)
+ c.Logf("stderr:\n%s", stderr.String())
+ c.Logf("json:\n%s", string(j))
+
+ var loaded []struct{ URL string }
+ err = json.Unmarshal(j, &loaded)
+ c.Check(err, check.IsNil)
+ if len(loaded) < max {
+ // Dumped when #requests was >90% but <100% of
+ // limit. If we stop now, we won't be able to
+ // confirm (below) that management endpoints
+ // are still accessible when normal requests
+ // are at 100%.
+ c.Logf("loaded dumped requests, but len %d < max %d -- still waiting", len(loaded), max)
+ continue
+ }
+ c.Check(loaded, check.HasLen, max)
+ c.Check(loaded[0].URL, check.Equals, "/testpath")
+ break
+ }
+
+ for _, path := range []string{"/_inspect/requests", "/metrics"} {
+ 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)
+ if !c.Check(err, check.IsNil) {
+ break
+ }
+ c.Logf("got response for %s", path)
+ c.Check(resp.StatusCode, check.Equals, http.StatusOK)
+ buf, err := ioutil.ReadAll(resp.Body)
+ c.Check(err, check.IsNil)
+ switch path {
+ case "/metrics":
+ c.Check(string(buf), check.Matches, `(?ms).*arvados_concurrent_requests `+fmt.Sprintf("%d", max)+`\n.*`)
+ case "/_inspect/requests":
+ c.Check(string(buf), check.Matches, `(?ms).*"URL":"/testpath".*`)
+ default:
+ c.Error("oops, testing bug")
+ }
+ }
+ close(hold)
+ cancel()
+}
+