X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/6b77f5f193b9c7e931d2e0b0712c1b27cd547fc8..1f7aeab7038ed8e2aa0f824185b392c8c13e6623:/lib/service/cmd_test.go

diff --git a/lib/service/cmd_test.go b/lib/service/cmd_test.go
index 7db9109274..08b3a239dc 100644
--- 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"
 
@@ -37,15 +39,19 @@ const (
 	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 "/"
@@ -58,17 +64,17 @@ func (*Suite) TestGetListenAddress(c *check.C) {
 		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
@@ -83,16 +89,16 @@ func (*Suite) TestGetListenAddress(c *check.C) {
 		{
 			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",
 		},
@@ -192,7 +198,149 @@ func (*Suite) TestCommand(c *check.C) {
 	c.Check(stderr.String(), check.Matches, `(?ms).*"msg":"CheckHealth called".*`)
 }
 
+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()
+}
+
 func (*Suite) TestTLS(c *check.C) {
+	port := unusedPort(c)
 	cwd, err := os.Getwd()
 	c.Assert(err, check.IsNil)
 
@@ -202,8 +350,8 @@ Clusters:
   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
@@ -228,7 +376,7 @@ Clusters:
 		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