20475: Option to dump active requests when queue is >=90% full.

[arvados.git] / lib / service / cmd_test.go

diff --git a/lib/service/cmd_test.go b/lib/service/cmd_test.go
index 7db91092745e2e4886f0b1b35a3015da0f0387fc..22b8415137b76fafaa54319ec4c71f6df7e9a98c 100644 (file)
--- a/lib/service/cmd_test.go
+++ b/lib/service/cmd_test.go
@@ -9,12 +9,14 @@ import (
        "bytes"
        "context"
        "crypto/tls"
+       "encoding/json"
        "fmt"
        "io/ioutil"
        "net"
        "net/http"
        "net/url"
        "os"
+       "strings"
        "testing"
        "time"
 
@@ -192,6 +194,102 @@ func (*Suite) TestCommand(c *check.C) {
        c.Check(stderr.String(), check.Matches, `(?ms).*"msg":"CheckHealth called".*`)
 }
 
+func (*Suite) TestDumpRequests(c *check.C) {
+       defer func(orig time.Duration) { requestQueueDumpCheckInterval = orig }(requestQueueDumpCheckInterval)
+       requestQueueDumpCheckInterval = time.Second / 10
+
+       tmpdir := c.MkDir()
+       cf, err := ioutil.TempFile(tmpdir, "cmd_test.")
+       c.Assert(err, check.IsNil)
+       defer os.Remove(cf.Name())
+       defer cf.Close()
+
+       max := 1
+       fmt.Fprintf(cf, `
+Clusters:
+ zzzzz:
+  SystemRootToken: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
+  ManagementToken: bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
+  API: {MaxConcurrentRequests: %d}
+  SystemLogs: {RequestQueueDumpDirectory: %q}
+  Services:
+   Controller:
+    ExternalURL: "http://localhost:12345"
+    InternalURLs: {"http://localhost:12345": {}}
+`, max, tmpdir)
+
+       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(arvados.ServiceNameController, 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("arvados-controller", []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:12345/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")
+                       }
+                       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())
+                       panic("timed out")
+               }
+       }
+       for {
+               j, err := os.ReadFile(tmpdir + "/arvados-controller-requests.json")
+               if os.IsNotExist(err) && deadline.After(time.Now()) {
+                       time.Sleep(time.Second / 100)
+                       continue
+               }
+               c.Check(err, check.IsNil)
+               c.Logf("%s", stderr.String())
+               c.Logf("%s", string(j))
+
+               var loaded []struct{ URL string }
+               err = json.Unmarshal(j, &loaded)
+               c.Check(err, check.IsNil)
+               c.Check(loaded, check.HasLen, max)
+               c.Check(loaded[0].URL, check.Equals, "/testpath")
+               break
+       }
+       close(hold)
+       cancel()
+
+}
+
 func (*Suite) TestTLS(c *check.C) {
        cwd, err := os.Getwd()
        c.Assert(err, check.IsNil)