19973: Test throttling after receiving 503 error.
[arvados.git] / lib / dispatchcloud / dispatcher_test.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package dispatchcloud
6
7 import (
8         "context"
9         "crypto/tls"
10         "encoding/json"
11         "io/ioutil"
12         "math/rand"
13         "net/http"
14         "net/http/httptest"
15         "net/url"
16         "os"
17         "sync"
18         "time"
19
20         "git.arvados.org/arvados.git/lib/config"
21         "git.arvados.org/arvados.git/lib/dispatchcloud/test"
22         "git.arvados.org/arvados.git/sdk/go/arvados"
23         "git.arvados.org/arvados.git/sdk/go/arvadostest"
24         "git.arvados.org/arvados.git/sdk/go/ctxlog"
25         "github.com/prometheus/client_golang/prometheus"
26         "golang.org/x/crypto/ssh"
27         check "gopkg.in/check.v1"
28 )
29
30 var _ = check.Suite(&DispatcherSuite{})
31
32 type DispatcherSuite struct {
33         ctx            context.Context
34         cancel         context.CancelFunc
35         cluster        *arvados.Cluster
36         stubDriver     *test.StubDriver
37         disp           *dispatcher
38         error503Server *httptest.Server
39 }
40
41 func (s *DispatcherSuite) SetUpTest(c *check.C) {
42         s.ctx, s.cancel = context.WithCancel(context.Background())
43         s.ctx = ctxlog.Context(s.ctx, ctxlog.TestLogger(c))
44         dispatchpub, _ := test.LoadTestKey(c, "test/sshkey_dispatch")
45         dispatchprivraw, err := ioutil.ReadFile("test/sshkey_dispatch")
46         c.Assert(err, check.IsNil)
47
48         _, hostpriv := test.LoadTestKey(c, "test/sshkey_vm")
49         s.stubDriver = &test.StubDriver{
50                 HostKey:                   hostpriv,
51                 AuthorizedKeys:            []ssh.PublicKey{dispatchpub},
52                 ErrorRateDestroy:          0.1,
53                 MinTimeBetweenCreateCalls: time.Millisecond,
54         }
55
56         // We need the postgresql connection info from the integration
57         // test config.
58         cfg, err := config.NewLoader(nil, ctxlog.FromContext(s.ctx)).Load()
59         c.Assert(err, check.IsNil)
60         testcluster, err := cfg.GetCluster("")
61         c.Assert(err, check.IsNil)
62
63         s.cluster = &arvados.Cluster{
64                 ManagementToken: "test-management-token",
65                 PostgreSQL:      testcluster.PostgreSQL,
66                 Containers: arvados.ContainersConfig{
67                         CrunchRunCommand:       "crunch-run",
68                         CrunchRunArgumentsList: []string{"--foo", "--extra='args'"},
69                         DispatchPrivateKey:     string(dispatchprivraw),
70                         StaleLockTimeout:       arvados.Duration(5 * time.Millisecond),
71                         RuntimeEngine:          "stub",
72                         CloudVMs: arvados.CloudVMsConfig{
73                                 Driver:               "test",
74                                 SyncInterval:         arvados.Duration(10 * time.Millisecond),
75                                 TimeoutIdle:          arvados.Duration(150 * time.Millisecond),
76                                 TimeoutBooting:       arvados.Duration(150 * time.Millisecond),
77                                 TimeoutProbe:         arvados.Duration(15 * time.Millisecond),
78                                 TimeoutShutdown:      arvados.Duration(5 * time.Millisecond),
79                                 MaxCloudOpsPerSecond: 500,
80                                 PollInterval:         arvados.Duration(5 * time.Millisecond),
81                                 ProbeInterval:        arvados.Duration(5 * time.Millisecond),
82                                 MaxProbesPerSecond:   1000,
83                                 TimeoutSignal:        arvados.Duration(3 * time.Millisecond),
84                                 TimeoutStaleRunLock:  arvados.Duration(3 * time.Millisecond),
85                                 TimeoutTERM:          arvados.Duration(20 * time.Millisecond),
86                                 ResourceTags:         map[string]string{"testtag": "test value"},
87                                 TagKeyPrefix:         "test:",
88                         },
89                 },
90                 InstanceTypes: arvados.InstanceTypeMap{
91                         test.InstanceType(1).Name:  test.InstanceType(1),
92                         test.InstanceType(2).Name:  test.InstanceType(2),
93                         test.InstanceType(3).Name:  test.InstanceType(3),
94                         test.InstanceType(4).Name:  test.InstanceType(4),
95                         test.InstanceType(6).Name:  test.InstanceType(6),
96                         test.InstanceType(8).Name:  test.InstanceType(8),
97                         test.InstanceType(16).Name: test.InstanceType(16),
98                 },
99         }
100         arvadostest.SetServiceURL(&s.cluster.Services.DispatchCloud, "http://localhost:/")
101         arvadostest.SetServiceURL(&s.cluster.Services.Controller, "https://"+os.Getenv("ARVADOS_API_HOST")+"/")
102
103         arvClient, err := arvados.NewClientFromConfig(s.cluster)
104         c.Check(err, check.IsNil)
105
106         s.error503Server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusServiceUnavailable) }))
107         arvClient.Client = &http.Client{
108                 Transport: &http.Transport{
109                         Proxy: s.arvClientProxy(c),
110                         TLSClientConfig: &tls.Config{
111                                 InsecureSkipVerify: true}}}
112
113         s.disp = &dispatcher{
114                 Cluster:   s.cluster,
115                 Context:   s.ctx,
116                 ArvClient: arvClient,
117                 AuthToken: arvadostest.AdminToken,
118                 Registry:  prometheus.NewRegistry(),
119         }
120         // Test cases can modify s.cluster before calling
121         // initialize(), and then modify private state before calling
122         // go run().
123 }
124
125 func (s *DispatcherSuite) TearDownTest(c *check.C) {
126         s.cancel()
127         s.disp.Close()
128         s.error503Server.Close()
129 }
130
131 // Intercept outgoing API requests for "/503" and respond HTTP
132 // 503. This lets us force (*arvados.Client)Last503() to return
133 // something.
134 func (s *DispatcherSuite) arvClientProxy(c *check.C) func(*http.Request) (*url.URL, error) {
135         return func(req *http.Request) (*url.URL, error) {
136                 if req.URL.Path == "/503" {
137                         return url.Parse(s.error503Server.URL)
138                 } else {
139                         return nil, nil
140                 }
141         }
142 }
143
144 // DispatchToStubDriver checks that the dispatcher wires everything
145 // together effectively. It uses a real scheduler and worker pool with
146 // a fake queue and cloud driver. The fake cloud driver injects
147 // artificial errors in order to exercise a variety of code paths.
148 func (s *DispatcherSuite) TestDispatchToStubDriver(c *check.C) {
149         Drivers["test"] = s.stubDriver
150         s.disp.setupOnce.Do(s.disp.initialize)
151         queue := &test.Queue{
152                 ChooseType: func(ctr *arvados.Container) (arvados.InstanceType, error) {
153                         return ChooseInstanceType(s.cluster, ctr)
154                 },
155                 Logger: ctxlog.TestLogger(c),
156         }
157         for i := 0; i < 200; i++ {
158                 queue.Containers = append(queue.Containers, arvados.Container{
159                         UUID:     test.ContainerUUID(i + 1),
160                         State:    arvados.ContainerStateQueued,
161                         Priority: int64(i%20 + 1),
162                         RuntimeConstraints: arvados.RuntimeConstraints{
163                                 RAM:   int64(i%3+1) << 30,
164                                 VCPUs: i%8 + 1,
165                         },
166                 })
167         }
168         s.disp.queue = queue
169
170         var mtx sync.Mutex
171         done := make(chan struct{})
172         waiting := map[string]struct{}{}
173         for _, ctr := range queue.Containers {
174                 waiting[ctr.UUID] = struct{}{}
175         }
176         finishContainer := func(ctr arvados.Container) {
177                 mtx.Lock()
178                 defer mtx.Unlock()
179                 if _, ok := waiting[ctr.UUID]; !ok {
180                         c.Errorf("container completed twice: %s", ctr.UUID)
181                         return
182                 }
183                 delete(waiting, ctr.UUID)
184                 if len(waiting) == 100 {
185                         // trigger scheduler maxConcurrency limit
186                         s.disp.ArvClient.RequestAndDecode(nil, "GET", "503", nil, nil)
187                 }
188                 if len(waiting) == 0 {
189                         close(done)
190                 }
191         }
192         executeContainer := func(ctr arvados.Container) int {
193                 finishContainer(ctr)
194                 return int(rand.Uint32() & 0x3)
195         }
196         n := 0
197         s.stubDriver.Queue = queue
198         s.stubDriver.SetupVM = func(stubvm *test.StubVM) {
199                 n++
200                 stubvm.Boot = time.Now().Add(time.Duration(rand.Int63n(int64(5 * time.Millisecond))))
201                 stubvm.CrunchRunDetachDelay = time.Duration(rand.Int63n(int64(10 * time.Millisecond)))
202                 stubvm.ExecuteContainer = executeContainer
203                 stubvm.CrashRunningContainer = finishContainer
204                 stubvm.ExtraCrunchRunArgs = "'--runtime-engine=stub' '--foo' '--extra='\\''args'\\'''"
205                 switch n % 7 {
206                 case 0:
207                         stubvm.Broken = time.Now().Add(time.Duration(rand.Int63n(90)) * time.Millisecond)
208                 case 1:
209                         stubvm.CrunchRunMissing = true
210                 case 2:
211                         stubvm.ReportBroken = time.Now().Add(time.Duration(rand.Int63n(200)) * time.Millisecond)
212                 default:
213                         stubvm.CrunchRunCrashRate = 0.1
214                         stubvm.ArvMountDeadlockRate = 0.1
215                 }
216         }
217         s.stubDriver.Bugf = c.Errorf
218
219         start := time.Now()
220         go s.disp.run()
221         err := s.disp.CheckHealth()
222         c.Check(err, check.IsNil)
223
224         for len(waiting) > 0 {
225                 waswaiting := len(waiting)
226                 select {
227                 case <-done:
228                         // loop will end because len(waiting)==0
229                 case <-time.After(3 * time.Second):
230                         if len(waiting) >= waswaiting {
231                                 c.Fatalf("timed out; no progress in 3s while waiting for %d containers: %q", len(waiting), waiting)
232                         }
233                 }
234         }
235         c.Logf("containers finished (%s), waiting for instances to shutdown and queue to clear", time.Since(start))
236
237         deadline := time.Now().Add(5 * time.Second)
238         for range time.NewTicker(10 * time.Millisecond).C {
239                 insts, err := s.stubDriver.InstanceSets()[0].Instances(nil)
240                 c.Check(err, check.IsNil)
241                 queue.Update()
242                 ents, _ := queue.Entries()
243                 if len(ents) == 0 && len(insts) == 0 {
244                         break
245                 }
246                 if time.Now().After(deadline) {
247                         c.Fatalf("timed out with %d containers (%v), %d instances (%+v)", len(ents), ents, len(insts), insts)
248                 }
249         }
250
251         req := httptest.NewRequest("GET", "/metrics", nil)
252         req.Header.Set("Authorization", "Bearer "+s.cluster.ManagementToken)
253         resp := httptest.NewRecorder()
254         s.disp.ServeHTTP(resp, req)
255         c.Check(resp.Code, check.Equals, http.StatusOK)
256         c.Check(resp.Body.String(), check.Matches, `(?ms).*driver_operations{error="0",operation="Create"} [^0].*`)
257         c.Check(resp.Body.String(), check.Matches, `(?ms).*driver_operations{error="0",operation="List"} [^0].*`)
258         c.Check(resp.Body.String(), check.Matches, `(?ms).*driver_operations{error="0",operation="Destroy"} [^0].*`)
259         c.Check(resp.Body.String(), check.Matches, `(?ms).*driver_operations{error="1",operation="Create"} [^0].*`)
260         c.Check(resp.Body.String(), check.Matches, `(?ms).*driver_operations{error="1",operation="List"} 0\n.*`)
261         c.Check(resp.Body.String(), check.Matches, `(?ms).*boot_outcomes{outcome="aborted"} [0-9]+\n.*`)
262         c.Check(resp.Body.String(), check.Matches, `(?ms).*boot_outcomes{outcome="disappeared"} [^0].*`)
263         c.Check(resp.Body.String(), check.Matches, `(?ms).*boot_outcomes{outcome="failure"} [^0].*`)
264         c.Check(resp.Body.String(), check.Matches, `(?ms).*boot_outcomes{outcome="success"} [^0].*`)
265         c.Check(resp.Body.String(), check.Matches, `(?ms).*instances_disappeared{state="shutdown"} [^0].*`)
266         c.Check(resp.Body.String(), check.Matches, `(?ms).*instances_disappeared{state="unknown"} 0\n.*`)
267         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_to_ssh_seconds{quantile="0.95"} [0-9.]*`)
268         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_to_ssh_seconds_count [0-9]*`)
269         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_to_ssh_seconds_sum [0-9.]*`)
270         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_to_ready_for_container_seconds{quantile="0.95"} [0-9.]*`)
271         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_to_ready_for_container_seconds_count [0-9]*`)
272         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_to_ready_for_container_seconds_sum [0-9.]*`)
273         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_from_shutdown_request_to_disappearance_seconds_count [0-9]*`)
274         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_from_shutdown_request_to_disappearance_seconds_sum [0-9.]*`)
275         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_from_queue_to_crunch_run_seconds_count [0-9]*`)
276         c.Check(resp.Body.String(), check.Matches, `(?ms).*time_from_queue_to_crunch_run_seconds_sum [0-9e+.]*`)
277         c.Check(resp.Body.String(), check.Matches, `(?ms).*run_probe_duration_seconds_count{outcome="success"} [0-9]*`)
278         c.Check(resp.Body.String(), check.Matches, `(?ms).*run_probe_duration_seconds_sum{outcome="success"} [0-9e+.]*`)
279         c.Check(resp.Body.String(), check.Matches, `(?ms).*run_probe_duration_seconds_count{outcome="fail"} [0-9]*`)
280         c.Check(resp.Body.String(), check.Matches, `(?ms).*run_probe_duration_seconds_sum{outcome="fail"} [0-9e+.]*`)
281         c.Check(resp.Body.String(), check.Matches, `(?ms).*last_503_time [1-9][0-9e+.]*`)
282         c.Check(resp.Body.String(), check.Matches, `(?ms).*max_concurrent_containers [1-9][0-9e+.]*`)
283 }
284
285 func (s *DispatcherSuite) TestAPIPermissions(c *check.C) {
286         s.cluster.ManagementToken = "abcdefgh"
287         Drivers["test"] = s.stubDriver
288         s.disp.setupOnce.Do(s.disp.initialize)
289         s.disp.queue = &test.Queue{}
290         go s.disp.run()
291
292         for _, token := range []string{"abc", ""} {
293                 req := httptest.NewRequest("GET", "/arvados/v1/dispatch/instances", nil)
294                 if token != "" {
295                         req.Header.Set("Authorization", "Bearer "+token)
296                 }
297                 resp := httptest.NewRecorder()
298                 s.disp.ServeHTTP(resp, req)
299                 if token == "" {
300                         c.Check(resp.Code, check.Equals, http.StatusUnauthorized)
301                 } else {
302                         c.Check(resp.Code, check.Equals, http.StatusForbidden)
303                 }
304         }
305 }
306
307 func (s *DispatcherSuite) TestAPIDisabled(c *check.C) {
308         s.cluster.ManagementToken = ""
309         Drivers["test"] = s.stubDriver
310         s.disp.setupOnce.Do(s.disp.initialize)
311         s.disp.queue = &test.Queue{}
312         go s.disp.run()
313
314         for _, token := range []string{"abc", ""} {
315                 req := httptest.NewRequest("GET", "/arvados/v1/dispatch/instances", nil)
316                 if token != "" {
317                         req.Header.Set("Authorization", "Bearer "+token)
318                 }
319                 resp := httptest.NewRecorder()
320                 s.disp.ServeHTTP(resp, req)
321                 c.Check(resp.Code, check.Equals, http.StatusForbidden)
322         }
323 }
324
325 func (s *DispatcherSuite) TestInstancesAPI(c *check.C) {
326         s.cluster.ManagementToken = "abcdefgh"
327         s.cluster.Containers.CloudVMs.TimeoutBooting = arvados.Duration(time.Second)
328         Drivers["test"] = s.stubDriver
329         s.disp.setupOnce.Do(s.disp.initialize)
330         s.disp.queue = &test.Queue{}
331         go s.disp.run()
332
333         type instance struct {
334                 Instance             string
335                 WorkerState          string `json:"worker_state"`
336                 Price                float64
337                 LastContainerUUID    string `json:"last_container_uuid"`
338                 ArvadosInstanceType  string `json:"arvados_instance_type"`
339                 ProviderInstanceType string `json:"provider_instance_type"`
340         }
341         type instancesResponse struct {
342                 Items []instance
343         }
344         getInstances := func() instancesResponse {
345                 req := httptest.NewRequest("GET", "/arvados/v1/dispatch/instances", nil)
346                 req.Header.Set("Authorization", "Bearer abcdefgh")
347                 resp := httptest.NewRecorder()
348                 s.disp.ServeHTTP(resp, req)
349                 var sr instancesResponse
350                 c.Check(resp.Code, check.Equals, http.StatusOK)
351                 err := json.Unmarshal(resp.Body.Bytes(), &sr)
352                 c.Check(err, check.IsNil)
353                 return sr
354         }
355
356         sr := getInstances()
357         c.Check(len(sr.Items), check.Equals, 0)
358
359         ch := s.disp.pool.Subscribe()
360         defer s.disp.pool.Unsubscribe(ch)
361         ok := s.disp.pool.Create(test.InstanceType(1))
362         c.Check(ok, check.Equals, true)
363         <-ch
364
365         for deadline := time.Now().Add(time.Second); time.Now().Before(deadline); {
366                 sr = getInstances()
367                 if len(sr.Items) > 0 {
368                         break
369                 }
370                 time.Sleep(time.Millisecond)
371         }
372         c.Assert(len(sr.Items), check.Equals, 1)
373         c.Check(sr.Items[0].Instance, check.Matches, "inst.*")
374         c.Check(sr.Items[0].WorkerState, check.Equals, "booting")
375         c.Check(sr.Items[0].Price, check.Equals, 0.123)
376         c.Check(sr.Items[0].LastContainerUUID, check.Equals, "")
377         c.Check(sr.Items[0].ProviderInstanceType, check.Equals, test.InstanceType(1).ProviderType)
378         c.Check(sr.Items[0].ArvadosInstanceType, check.Equals, test.InstanceType(1).Name)
379 }