1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
5 // package service provides a cmd.Handler that brings up a system service.
25 "git.arvados.org/arvados.git/sdk/go/arvados"
26 "git.arvados.org/arvados.git/sdk/go/ctxlog"
27 "github.com/prometheus/client_golang/prometheus"
28 check "gopkg.in/check.v1"
31 func Test(t *testing.T) {
35 var _ = check.Suite(&Suite{})
44 func unusedPort(c *check.C) string {
45 // Find an available port on the testing host, so the test
46 // cases don't get confused by "already in use" errors.
47 listener, err := net.Listen("tcp", ":")
48 c.Assert(err, check.IsNil)
50 _, port, err := net.SplitHostPort(listener.Addr().String())
51 c.Assert(err, check.IsNil)
55 func (*Suite) TestGetListenAddress(c *check.C) {
57 defer os.Unsetenv("ARVADOS_SERVICE_INTERNAL_URL")
58 for idx, trial := range []struct {
59 // internalURL => listenURL, both with trailing "/"
60 // because config loader always adds it
61 internalURLs map[string]string
63 expectErrorMatch string
64 expectLogsMatch string
69 internalURLs: map[string]string{"http://localhost:" + port + "/": ""},
70 expectListen: "http://localhost:" + port + "/",
71 expectInternal: "http://localhost:" + port + "/",
73 { // implicit port 80 in InternalURLs
74 internalURLs: map[string]string{"http://localhost/": ""},
75 expectErrorMatch: `.*:80: bind: permission denied`,
77 { // implicit port 443 in InternalURLs
78 internalURLs: map[string]string{"https://host.example/": "http://localhost:" + port + "/"},
79 expectListen: "http://localhost:" + port + "/",
80 expectInternal: "https://host.example/",
82 { // implicit port 443 in ListenURL
83 internalURLs: map[string]string{"wss://host.example/": "wss://localhost/"},
84 expectErrorMatch: `.*:443: bind: permission denied`,
87 internalURLs: map[string]string{"https://hostname.example/": "http://localhost:8000/"},
88 expectListen: "http://localhost:8000/",
89 expectInternal: "https://hostname.example/",
92 internalURLs: map[string]string{
93 "https://hostname1.example/": "http://localhost:12435/",
94 "https://hostname2.example/": "http://localhost:" + port + "/",
96 envVar: "https://hostname2.example", // note this works despite missing trailing "/"
97 expectListen: "http://localhost:" + port + "/",
98 expectInternal: "https://hostname2.example/",
100 { // cannot listen on any of the ListenURLs
101 internalURLs: map[string]string{
102 "https://hostname1.example/": "http://1.2.3.4:" + port + "/",
103 "https://hostname2.example/": "http://1.2.3.4:" + port + "/",
105 expectErrorMatch: "configuration does not enable the \"arvados-controller\" service on this host",
107 { // cannot listen on any of the (implied) ListenURLs
108 internalURLs: map[string]string{
109 "https://1.2.3.4/": "",
110 "https://1.2.3.5/": "",
112 expectErrorMatch: "configuration does not enable the \"arvados-controller\" service on this host",
114 { // impossible port number
115 internalURLs: map[string]string{
116 "https://host.example/": "http://0.0.0.0:1234567",
118 expectErrorMatch: `.*:1234567: listen tcp: address 1234567: invalid port`,
121 // env var URL not mentioned in config = obey env var, with warning
122 internalURLs: map[string]string{"https://hostname1.example/": "http://localhost:8000/"},
123 envVar: "https://hostname2.example",
124 expectListen: "https://hostname2.example/",
125 expectInternal: "https://hostname2.example/",
126 expectLogsMatch: `.*\Qpossible configuration error: listening on https://hostname2.example/ (from $ARVADOS_SERVICE_INTERNAL_URL) even though configuration does not have a matching InternalURLs entry\E.*\n`,
129 // env var + empty config = obey env var, with warning
130 envVar: "https://hostname.example",
131 expectListen: "https://hostname.example/",
132 expectInternal: "https://hostname.example/",
133 expectLogsMatch: `.*\Qpossible configuration error: listening on https://hostname.example/ (from $ARVADOS_SERVICE_INTERNAL_URL) even though configuration does not have a matching InternalURLs entry\E.*\n`,
136 c.Logf("trial %d %+v", idx, trial)
137 os.Setenv("ARVADOS_SERVICE_INTERNAL_URL", trial.envVar)
138 var logbuf bytes.Buffer
139 log := ctxlog.New(&logbuf, "text", "info")
140 services := arvados.Services{Controller: arvados.Service{InternalURLs: map[arvados.URL]arvados.ServiceInstance{}}}
141 for k, v := range trial.internalURLs {
142 u, err := url.Parse(k)
143 c.Assert(err, check.IsNil)
144 si := arvados.ServiceInstance{}
146 u, err := url.Parse(v)
147 c.Assert(err, check.IsNil)
148 si.ListenURL = arvados.URL(*u)
150 services.Controller.InternalURLs[arvados.URL(*u)] = si
152 listenURL, internalURL, err := getListenAddr(services, "arvados-controller", log)
153 if trial.expectLogsMatch != "" {
154 c.Check(logbuf.String(), check.Matches, trial.expectLogsMatch)
156 if trial.expectErrorMatch != "" {
157 c.Check(err, check.ErrorMatches, trial.expectErrorMatch)
160 if !c.Check(err, check.IsNil) {
163 c.Check(listenURL.String(), check.Equals, trial.expectListen)
164 c.Check(internalURL.String(), check.Equals, trial.expectInternal)
168 func (*Suite) TestCommand(c *check.C) {
169 cf, err := ioutil.TempFile("", "cmd_test.")
170 c.Assert(err, check.IsNil)
171 defer os.Remove(cf.Name())
173 fmt.Fprintf(cf, "Clusters:\n zzzzz:\n SystemRootToken: abcde\n NodeProfiles: {\"*\": {\"arvados-controller\": {Listen: \":1234\"}}}")
175 healthCheck := make(chan bool, 1)
176 ctx, cancel := context.WithCancel(context.Background())
179 cmd := Command(arvados.ServiceNameController, func(ctx context.Context, _ *arvados.Cluster, token string, reg *prometheus.Registry) Handler {
180 c.Check(ctx.Value(contextKey), check.Equals, "bar")
181 c.Check(token, check.Equals, "abcde")
182 return &testHandler{ctx: ctx, healthCheck: healthCheck}
184 cmd.(*command).ctx = context.WithValue(ctx, contextKey, "bar")
186 done := make(chan bool)
187 var stdin, stdout, stderr bytes.Buffer
190 cmd.RunCommand("arvados-controller", []string{"-config", cf.Name()}, &stdin, &stdout, &stderr)
196 c.Error("command exited without health check")
199 c.Check(stdout.String(), check.Equals, "")
200 c.Check(stderr.String(), check.Matches, `(?ms).*"msg":"CheckHealth called".*`)
203 func (s *Suite) TestRequestLimitsAndDumpRequests_Keepweb(c *check.C) {
204 s.testRequestLimitAndDumpRequests(c, arvados.ServiceNameKeepweb, "MaxConcurrentRequests")
207 func (s *Suite) TestRequestLimitsAndDumpRequests_Controller(c *check.C) {
208 s.testRequestLimitAndDumpRequests(c, arvados.ServiceNameController, "MaxConcurrentRailsRequests")
211 func (*Suite) testRequestLimitAndDumpRequests(c *check.C, serviceName arvados.ServiceName, maxReqsConfigKey string) {
212 defer func(orig time.Duration) { requestQueueDumpCheckInterval = orig }(requestQueueDumpCheckInterval)
213 requestQueueDumpCheckInterval = time.Second / 10
215 port := unusedPort(c)
217 cf, err := ioutil.TempFile(tmpdir, "cmd_test.")
218 c.Assert(err, check.IsNil)
219 defer os.Remove(cf.Name())
227 SystemRootToken: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
228 ManagementToken: bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
230 `+maxReqsConfigKey+`: %d
232 MaxGatewayTunnels: %d
233 SystemLogs: {RequestQueueDumpDirectory: %q}
236 ExternalURL: "http://localhost:`+port+`"
237 InternalURLs: {"http://localhost:`+port+`": {}}
239 ExternalURL: "http://localhost:`+port+`"
240 InternalURLs: {"http://localhost:`+port+`": {}}
241 `, max, maxTunnels, tmpdir)
244 started := make(chan bool, max+1)
245 hold := make(chan bool)
246 handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
247 if strings.HasPrefix(r.URL.Path, "/arvados/v1/connect") {
254 healthCheck := make(chan bool, 1)
255 ctx, cancel := context.WithCancel(context.Background())
258 cmd := Command(serviceName, func(ctx context.Context, _ *arvados.Cluster, token string, reg *prometheus.Registry) Handler {
259 return &testHandler{ctx: ctx, handler: handler, healthCheck: healthCheck}
261 cmd.(*command).ctx = context.WithValue(ctx, contextKey, "bar")
263 exited := make(chan bool)
264 var stdin, stdout, stderr bytes.Buffer
267 cmd.RunCommand(string(serviceName), []string{"-config", cf.Name()}, &stdin, &stdout, &stderr)
273 c.Logf("%s", stderr.String())
274 c.Error("command exited without health check")
276 client := http.Client{}
277 deadline := time.Now().Add(time.Second * 2)
278 var activeReqs sync.WaitGroup
280 // Start some API reqs
281 var apiResp200, apiResp503 int64
282 for i := 0; i < max+1; i++ {
285 defer activeReqs.Done()
286 resp, err := client.Get("http://localhost:" + port + "/testpath")
287 for err != nil && strings.Contains(err.Error(), "dial tcp") && deadline.After(time.Now()) {
288 time.Sleep(time.Second / 100)
289 resp, err = client.Get("http://localhost:" + port + "/testpath")
291 if c.Check(err, check.IsNil) {
292 if resp.StatusCode == http.StatusOK {
293 atomic.AddInt64(&apiResp200, 1)
294 } else if resp.StatusCode == http.StatusServiceUnavailable {
295 atomic.AddInt64(&apiResp503, 1)
301 // Start some gateway tunnel reqs that don't count toward our
303 extraTunnelReqs := 20
304 var tunnelResp200, tunnelResp503 int64
305 for i := 0; i < maxTunnels+extraTunnelReqs; i++ {
308 defer activeReqs.Done()
309 resp, err := client.Get("http://localhost:" + port + "/arvados/v1/connect/...")
310 for err != nil && strings.Contains(err.Error(), "dial tcp") && deadline.After(time.Now()) {
311 time.Sleep(time.Second / 100)
312 resp, err = client.Get("http://localhost:" + port + "/arvados/v1/connect/...")
314 if c.Check(err, check.IsNil) {
315 if resp.StatusCode == http.StatusOK {
316 atomic.AddInt64(&tunnelResp200, 1)
317 } else if resp.StatusCode == http.StatusServiceUnavailable {
318 atomic.AddInt64(&tunnelResp503, 1)
320 c.Errorf("tunnel response code %d", resp.StatusCode)
325 for i := 0; i < max; i++ {
328 case <-time.After(time.Second):
329 c.Logf("%s", stderr.String())
333 for delay := time.Second / 100; ; delay = delay * 2 {
335 j, err := os.ReadFile(tmpdir + "/" + string(serviceName) + "-requests.json")
336 if os.IsNotExist(err) && deadline.After(time.Now()) {
339 c.Assert(err, check.IsNil)
340 c.Logf("stderr:\n%s", stderr.String())
341 c.Logf("json:\n%s", string(j))
343 var loaded []struct{ URL string }
344 err = json.Unmarshal(j, &loaded)
345 c.Check(err, check.IsNil)
347 for i := 0; i < len(loaded); i++ {
348 if strings.HasPrefix(loaded[i].URL, "/arvados/v1/connect/") {
349 // Filter out a gateway tunnel req
350 // that doesn't count toward our API
352 if i < len(loaded)-1 {
353 copy(loaded[i:], loaded[i+1:])
356 loaded = loaded[:len(loaded)-1]
360 if len(loaded) < max {
361 // Dumped when #requests was >90% but <100% of
362 // limit. If we stop now, we won't be able to
363 // confirm (below) that management endpoints
364 // are still accessible when normal requests
366 c.Logf("loaded dumped requests, but len %d < max %d -- still waiting", len(loaded), max)
369 c.Check(loaded, check.HasLen, max+1)
370 c.Check(loaded[0].URL, check.Equals, "/testpath")
374 for _, path := range []string{"/_inspect/requests", "/metrics"} {
375 req, err := http.NewRequest("GET", "http://localhost:"+port+""+path, nil)
376 c.Assert(err, check.IsNil)
377 req.Header.Set("Authorization", "Bearer bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
378 resp, err := client.Do(req)
379 if !c.Check(err, check.IsNil) {
382 c.Logf("got response for %s", path)
383 c.Check(resp.StatusCode, check.Equals, http.StatusOK)
384 buf, err := ioutil.ReadAll(resp.Body)
385 c.Check(err, check.IsNil)
388 c.Check(string(buf), check.Matches, `(?ms).*arvados_concurrent_requests{queue="api"} `+fmt.Sprintf("%d", max)+`\n.*`)
389 c.Check(string(buf), check.Matches, `(?ms).*arvados_queued_requests{priority="normal",queue="api"} 1\n.*`)
390 case "/_inspect/requests":
391 c.Check(string(buf), check.Matches, `(?ms).*"URL":"/testpath".*`)
393 c.Error("oops, testing bug")
398 c.Check(int(apiResp200), check.Equals, max+1)
399 c.Check(int(apiResp503), check.Equals, 0)
400 c.Check(int(tunnelResp200), check.Equals, maxTunnels)
401 c.Check(int(tunnelResp503), check.Equals, extraTunnelReqs)
405 func (*Suite) TestTLS(c *check.C) {
406 port := unusedPort(c)
407 cwd, err := os.Getwd()
408 c.Assert(err, check.IsNil)
410 stdin := bytes.NewBufferString(`
413 SystemRootToken: abcde
416 ExternalURL: "https://localhost:` + port + `"
417 InternalURLs: {"https://localhost:` + port + `": {}}
419 Key: file://` + cwd + `/../../services/api/tmp/self-signed.key
420 Certificate: file://` + cwd + `/../../services/api/tmp/self-signed.pem
423 called := make(chan bool)
424 cmd := Command(arvados.ServiceNameController, func(ctx context.Context, _ *arvados.Cluster, token string, reg *prometheus.Registry) Handler {
425 return &testHandler{handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
426 w.Write([]byte("ok"))
431 exited := make(chan bool)
432 var stdout, stderr bytes.Buffer
434 cmd.RunCommand("arvados-controller", []string{"-config", "-"}, stdin, &stdout, &stderr)
437 got := make(chan bool)
440 client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}}
441 for range time.NewTicker(time.Millisecond).C {
442 resp, err := client.Get("https://localhost:" + port)
447 body, err := ioutil.ReadAll(resp.Body)
448 c.Check(err, check.IsNil)
449 c.Logf("status %d, body %s", resp.StatusCode, string(body))
450 c.Check(resp.StatusCode, check.Equals, http.StatusOK)
457 c.Error("command exited without calling handler")
458 case <-time.After(time.Second):
464 c.Error("command exited before client received response")
465 case <-time.After(time.Second):
468 c.Log(stderr.String())
471 type testHandler struct {
474 healthCheck chan bool
477 func (th *testHandler) Done() <-chan struct{} { return nil }
478 func (th *testHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { th.handler.ServeHTTP(w, r) }
479 func (th *testHandler) CheckHealth() error {
480 ctxlog.FromContext(th.ctx).Info("CheckHealth called")
482 case th.healthCheck <- true: