"bytes"
"context"
"crypto/tls"
+ "encoding/json"
"fmt"
"io/ioutil"
"net"
"net/http"
"net/url"
"os"
+ "strings"
"testing"
"time"
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 (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)
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 / blobdiff