cmd/arvados-client/container_gateway_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package main
   6
   7 import (
   8         "bytes"
   9         "context"
  10         "crypto/hmac"
  11         "crypto/sha256"
  12         "crypto/tls"
  13         "fmt"
  14         "io"
  15         "io/ioutil"
  16         "net"
  17         "net/http"
  18         "net/url"
  19         "os"
  20         "os/exec"
  21         "strings"
  22         "sync"
  23         "syscall"
  24         "time"
  25
  26         "git.arvados.org/arvados.git/lib/controller/rpc"
  27         "git.arvados.org/arvados.git/lib/crunchrun"
  28         "git.arvados.org/arvados.git/sdk/go/arvados"
  29         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  30         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  31         "git.arvados.org/arvados.git/sdk/go/auth"
  32         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  33         "git.arvados.org/arvados.git/sdk/go/httpserver"
  34         "git.arvados.org/arvados.git/sdk/go/keepclient"
  35         check "gopkg.in/check.v1"
  36 )
  37
  38 var _ = check.Suite(&shellSuite{})
  39
  40 type shellSuite struct {
  41         gobindir    string
  42         homedir     string
  43         runningUUID string
  44 }
  45
  46 func (s *shellSuite) SetUpSuite(c *check.C) {
  47         tmpdir := c.MkDir()
  48         s.gobindir = tmpdir + "/bin"
  49         c.Check(os.Mkdir(s.gobindir, 0777), check.IsNil)
  50         s.homedir = tmpdir + "/home"
  51         c.Check(os.Mkdir(s.homedir, 0777), check.IsNil)
  52
  53         // We explicitly build a client binary in our tempdir here,
  54         // instead of using "go run .", because (a) we're going to
  55         // invoke the same binary several times, and (b) we're going
  56         // to change $HOME to a temp dir in some of the tests, which
  57         // would force "go run ." to recompile the world instead of
  58         // using the cached object files in the real $HOME.
  59         c.Logf("building arvados-client binary in %s", s.gobindir)
  60         cmd := exec.Command("go", "install", ".")
  61         cmd.Env = append(os.Environ(), "GOBIN="+s.gobindir)
  62         cmd.Stdout = os.Stdout
  63         cmd.Stderr = os.Stderr
  64         c.Assert(cmd.Run(), check.IsNil)
  65
  66         s.runningUUID = arvadostest.RunningContainerUUID
  67         h := hmac.New(sha256.New, []byte(arvadostest.SystemRootToken))
  68         fmt.Fprint(h, s.runningUUID)
  69         authSecret := fmt.Sprintf("%x", h.Sum(nil))
  70         gw := crunchrun.Gateway{
  71                 ContainerUUID: s.runningUUID,
  72                 Address:       "0.0.0.0:0",
  73                 AuthSecret:    authSecret,
  74                 Log:           ctxlog.TestLogger(c),
  75                 // Just forward connections to localhost instead of a
  76                 // container, so we can test without running a
  77                 // container.
  78                 Target: crunchrun.GatewayTargetStub{},
  79         }
  80         err := gw.Start()
  81         c.Assert(err, check.IsNil)
  82
  83         rpcconn := rpc.NewConn("",
  84                 &url.URL{
  85                         Scheme: "https",
  86                         Host:   os.Getenv("ARVADOS_API_HOST"),
  87                 },
  88                 true,
  89                 func(context.Context) ([]string, error) {
  90                         return []string{arvadostest.SystemRootToken}, nil
  91                 })
  92         _, err = rpcconn.ContainerUpdate(context.TODO(), arvados.UpdateOptions{UUID: s.runningUUID, Attrs: map[string]interface{}{
  93                 "gateway_address": gw.Address,
  94         }})
  95         c.Assert(err, check.IsNil)
  96 }
  97
  98 func (s *shellSuite) TearDownSuite(c *check.C) {
  99         c.Check(arvados.NewClientFromEnv().RequestAndDecode(nil, "POST", "database/reset", nil, nil), check.IsNil)
 100 }
 101
 102 func (s *shellSuite) TestShellGatewayNotAvailable(c *check.C) {
 103         var stdout, stderr bytes.Buffer
 104         cmd := exec.Command(s.gobindir+"/arvados-client", "shell", arvadostest.QueuedContainerUUID, "-o", "controlpath=none", "echo", "ok")
 105         cmd.Env = append(cmd.Env, os.Environ()...)
 106         cmd.Env = append(cmd.Env, "ARVADOS_API_TOKEN="+arvadostest.ActiveTokenV2)
 107         cmd.Stdout = &stdout
 108         cmd.Stderr = &stderr
 109         c.Check(cmd.Run(), check.NotNil)
 110         c.Log(stderr.String())
 111         c.Check(stderr.String(), check.Matches, `(?ms).*container is not running yet \(state is "Queued"\).*`)
 112 }
 113
 114 func (s *shellSuite) TestShellGatewayUsingEnvVars(c *check.C) {
 115         s.testShellGateway(c, false)
 116 }
 117 func (s *shellSuite) TestShellGatewayUsingSettingsConf(c *check.C) {
 118         s.testShellGateway(c, true)
 119 }
 120 func (s *shellSuite) testShellGateway(c *check.C, useSettingsConf bool) {
 121         var stdout, stderr bytes.Buffer
 122         cmd := exec.Command(
 123                 s.gobindir+"/arvados-client", "shell", s.runningUUID,
 124                 "-o", "controlpath=none",
 125                 "-o", "userknownhostsfile="+s.homedir+"/known_hosts",
 126                 "echo", "ok")
 127         if useSettingsConf {
 128                 settings := "ARVADOS_API_HOST=" + os.Getenv("ARVADOS_API_HOST") + "\nARVADOS_API_TOKEN=" + arvadostest.ActiveTokenV2 + "\nARVADOS_API_HOST_INSECURE=true\n"
 129                 err := os.MkdirAll(s.homedir+"/.config/arvados", 0777)
 130                 c.Assert(err, check.IsNil)
 131                 err = os.WriteFile(s.homedir+"/.config/arvados/settings.conf", []byte(settings), 0777)
 132                 c.Assert(err, check.IsNil)
 133                 for _, kv := range os.Environ() {
 134                         if !strings.HasPrefix(kv, "ARVADOS_") && !strings.HasPrefix(kv, "HOME=") {
 135                                 cmd.Env = append(cmd.Env, kv)
 136                         }
 137                 }
 138                 cmd.Env = append(cmd.Env, "HOME="+s.homedir)
 139         } else {
 140                 err := os.Remove(s.homedir + "/.config/arvados/settings.conf")
 141                 if !os.IsNotExist(err) {
 142                         c.Assert(err, check.IsNil)
 143                 }
 144                 cmd.Env = append(cmd.Env, os.Environ()...)
 145                 cmd.Env = append(cmd.Env, "ARVADOS_API_TOKEN="+arvadostest.ActiveTokenV2)
 146         }
 147         cmd.Stdout = &stdout
 148         cmd.Stderr = &stderr
 149         stdin, err := cmd.StdinPipe()
 150         c.Assert(err, check.IsNil)
 151         go fmt.Fprintln(stdin, "data appears on stdin, but stdin does not close; cmd should exit anyway, not hang")
 152         timeout := time.AfterFunc(5*time.Second, func() {
 153                 c.Errorf("timed out -- remote end is probably hung waiting for us to close stdin")
 154                 stdin.Close()
 155         })
 156         c.Logf("cmd.Args: %s", cmd.Args)
 157         c.Check(cmd.Run(), check.IsNil)
 158         timeout.Stop()
 159         c.Check(stdout.String(), check.Equals, "ok\n")
 160 }
 161
 162 func stubHTTPTarget(c *check.C) *httpserver.Server {
 163         c.Log("setting up an http server")
 164         // Set up an http server, and try using "arvados-client shell"
 165         // to forward traffic to it.
 166         httpTarget := &httpserver.Server{}
 167         httpTarget.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 168                 c.Logf("httpTarget.Handler: incoming request: %s %s", r.Method, r.URL)
 169                 if r.URL.Path == "/foo" {
 170                         fmt.Fprintln(w, "bar baz")
 171                 } else {
 172                         w.WriteHeader(http.StatusNotFound)
 173                 }
 174         })
 175         err := httpTarget.Start()
 176         c.Assert(err, check.IsNil)
 177         return httpTarget
 178 }
 179
 180 func (s *shellSuite) TestShellGatewayPortForwarding(c *check.C) {
 181         httpTarget := stubHTTPTarget(c)
 182
 183         ln, err := net.Listen("tcp", ":0")
 184         c.Assert(err, check.IsNil)
 185         _, forwardedPort, _ := net.SplitHostPort(ln.Addr().String())
 186         ln.Close()
 187
 188         c.Log("connecting")
 189         var stdout, stderr bytes.Buffer
 190         ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(10*time.Second))
 191         defer cancel()
 192         cmd := exec.CommandContext(ctx,
 193                 s.gobindir+"/arvados-client", "shell", s.runningUUID,
 194                 "-L", forwardedPort+":"+httpTarget.Addr,
 195                 "-o", "controlpath=none",
 196                 "-o", "userknownhostsfile="+s.homedir+"/known_hosts",
 197                 "-N",
 198         )
 199         cmd.Env = append(cmd.Env, os.Environ()...)
 200         cmd.Env = append(cmd.Env, "ARVADOS_API_TOKEN="+arvadostest.ActiveTokenV2)
 201         cmd.Stdout = &stdout
 202         cmd.Stderr = &stderr
 203         c.Logf("cmd.Args: %s", cmd.Args)
 204         cmd.Start()
 205
 206         forwardedURL := fmt.Sprintf("http://localhost:%s/foo", forwardedPort)
 207
 208         for range time.NewTicker(time.Second / 20).C {
 209                 resp, err := http.Get(forwardedURL)
 210                 if err != nil {
 211                         if !strings.Contains(err.Error(), "connect") {
 212                                 c.Fatal(err)
 213                         } else if ctx.Err() != nil {
 214                                 if cmd.Process.Signal(syscall.Signal(0)) != nil {
 215                                         c.Error("OpenSSH exited")
 216                                 } else {
 217                                         c.Errorf("timed out trying to connect: %s", err)
 218                                 }
 219                                 c.Logf("OpenSSH stdout:\n%s", stdout.String())
 220                                 c.Logf("OpenSSH stderr:\n%s", stderr.String())
 221                                 c.FailNow()
 222                         }
 223                         // Retry until OpenSSH starts listening
 224                         continue
 225                 }
 226                 c.Check(resp.StatusCode, check.Equals, http.StatusOK)
 227                 body, err := ioutil.ReadAll(resp.Body)
 228                 c.Check(err, check.IsNil)
 229                 c.Check(string(body), check.Equals, "bar baz\n")
 230                 break
 231         }
 232
 233         var wg sync.WaitGroup
 234         for i := 0; i < 10; i++ {
 235                 wg.Add(1)
 236                 go func() {
 237                         defer wg.Done()
 238                         resp, err := http.Get(forwardedURL)
 239                         if !c.Check(err, check.IsNil) {
 240                                 return
 241                         }
 242                         body, err := ioutil.ReadAll(resp.Body)
 243                         c.Check(err, check.IsNil)
 244                         c.Check(string(body), check.Equals, "bar baz\n")
 245                 }()
 246         }
 247         wg.Wait()
 248 }
 249
 250 // This test is arguably misplaced: arvados-client does not (yet?)
 251 // have a "do http request against container X port Y" feature, so
 252 // we're not really testing arvados-client here.  However, (a) it
 253 // might have one someday, and (b) testing the same http server setup
 254 // via both access mechanisms might help troubleshoot if one of them
 255 // fails.
 256 func (s *shellSuite) TestGatewayHTTPProxy(c *check.C) {
 257         ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(10*time.Second))
 258         defer cancel()
 259         httpTarget := stubHTTPTarget(c)
 260         client := &http.Client{
 261                 Transport: &http.Transport{
 262                         TLSClientConfig: &tls.Config{
 263                                 InsecureSkipVerify: true,
 264                         },
 265                 },
 266         }
 267         _, port, _ := net.SplitHostPort(httpTarget.Addr)
 268         req, err := http.NewRequestWithContext(ctx, "GET", "https://"+os.Getenv("ARVADOS_API_HOST")+"/foo", nil)
 269         c.Assert(err, check.IsNil)
 270         req.AddCookie(&http.Cookie{Name: "arvados_api_token", Value: auth.EncodeTokenCookie([]byte(arvadostest.ActiveTokenV2))})
 271         req.Host = s.runningUUID + "-" + port + ".example.com"
 272         resp, err := client.Do(req)
 273         c.Assert(err, check.IsNil)
 274         c.Check(resp.StatusCode, check.Equals, http.StatusOK)
 275         body, err := ioutil.ReadAll(resp.Body)
 276         c.Check(err, check.IsNil)
 277         c.Check(string(body), check.Equals, "bar baz\n")
 278 }
 279
 280 var _ = check.Suite(&logsSuite{})
 281
 282 type logsSuite struct{}
 283
 284 func (s *logsSuite) TestContainerRequestLog(c *check.C) {
 285         arvadostest.StartKeep(2, true)
 286         ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(30*time.Second))
 287         defer cancel()
 288
 289         rpcconn := rpc.NewConn("",
 290                 &url.URL{
 291                         Scheme: "https",
 292                         Host:   os.Getenv("ARVADOS_API_HOST"),
 293                 },
 294                 true,
 295                 func(context.Context) ([]string, error) {
 296                         return []string{arvadostest.SystemRootToken}, nil
 297                 })
 298         imageColl, err := rpcconn.CollectionCreate(ctx, arvados.CreateOptions{Attrs: map[string]interface{}{
 299                 "manifest_text": ". d41d8cd98f00b204e9800998ecf8427e+0 0:0:sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855.tar\n",
 300         }})
 301         c.Assert(err, check.IsNil)
 302         c.Logf("imageColl %+v", imageColl)
 303         cr, err := rpcconn.ContainerRequestCreate(ctx, arvados.CreateOptions{Attrs: map[string]interface{}{
 304                 "state":           "Committed",
 305                 "command":         []string{"echo", fmt.Sprintf("%d", time.Now().Unix())},
 306                 "container_image": imageColl.PortableDataHash,
 307                 "cwd":             "/",
 308                 "output_path":     "/",
 309                 "priority":        1,
 310                 "runtime_constraints": arvados.RuntimeConstraints{
 311                         VCPUs: 1,
 312                         RAM:   1000000000,
 313                 },
 314                 "container_count_max": 1,
 315         }})
 316         c.Assert(err, check.IsNil)
 317         h := hmac.New(sha256.New, []byte(arvadostest.SystemRootToken))
 318         fmt.Fprint(h, cr.ContainerUUID)
 319         authSecret := fmt.Sprintf("%x", h.Sum(nil))
 320
 321         coll := arvados.Collection{}
 322         client := arvados.NewClientFromEnv()
 323         ac, err := arvadosclient.New(client)
 324         c.Assert(err, check.IsNil)
 325         kc, err := keepclient.MakeKeepClient(ac)
 326         c.Assert(err, check.IsNil)
 327         cfs, err := coll.FileSystem(client, kc)
 328         c.Assert(err, check.IsNil)
 329
 330         c.Log("running logs command on queued container")
 331         var stdout, stderr bytes.Buffer
 332         cmd := exec.CommandContext(ctx, "go", "run", ".", "logs", "-f", "-poll=250ms", cr.UUID)
 333         cmd.Env = append(cmd.Env, os.Environ()...)
 334         cmd.Env = append(cmd.Env, "ARVADOS_API_TOKEN="+arvadostest.SystemRootToken)
 335         cmd.Stdout = io.MultiWriter(&stdout, os.Stderr)
 336         cmd.Stderr = io.MultiWriter(&stderr, os.Stderr)
 337         err = cmd.Start()
 338         c.Assert(err, check.Equals, nil)
 339
 340         c.Log("changing container state to Locked")
 341         _, err = rpcconn.ContainerUpdate(ctx, arvados.UpdateOptions{UUID: cr.ContainerUUID, Attrs: map[string]interface{}{
 342                 "state": arvados.ContainerStateLocked,
 343         }})
 344         c.Assert(err, check.IsNil)
 345         c.Log("starting gateway")
 346         gw := crunchrun.Gateway{
 347                 ContainerUUID: cr.ContainerUUID,
 348                 Address:       "0.0.0.0:0",
 349                 AuthSecret:    authSecret,
 350                 Log:           ctxlog.TestLogger(c),
 351                 Target:        crunchrun.GatewayTargetStub{},
 352                 LogCollection: cfs,
 353         }
 354         err = gw.Start()
 355         c.Assert(err, check.IsNil)
 356         c.Log("updating container gateway address")
 357         _, err = rpcconn.ContainerUpdate(ctx, arvados.UpdateOptions{UUID: cr.ContainerUUID, Attrs: map[string]interface{}{
 358                 "gateway_address": gw.Address,
 359                 "state":           arvados.ContainerStateRunning,
 360         }})
 361         c.Assert(err, check.IsNil)
 362
 363         const rfc3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00"
 364         fCrunchrun, err := cfs.OpenFile("crunch-run.txt", os.O_CREATE|os.O_WRONLY, 0777)
 365         c.Assert(err, check.IsNil)
 366         _, err = fmt.Fprintf(fCrunchrun, "%s line 1 of crunch-run.txt\n", time.Now().UTC().Format(rfc3339NanoFixed))
 367         c.Assert(err, check.IsNil)
 368         fStderr, err := cfs.OpenFile("stderr.txt", os.O_CREATE|os.O_WRONLY, 0777)
 369         c.Assert(err, check.IsNil)
 370         _, err = fmt.Fprintf(fStderr, "%s line 1 of stderr\n", time.Now().UTC().Format(rfc3339NanoFixed))
 371         c.Assert(err, check.IsNil)
 372
 373         {
 374                 // Without "-f", just show the existing logs and
 375                 // exit. Timeout needs to be long enough for "go run".
 376                 ctxNoFollow, cancel := context.WithDeadline(ctx, time.Now().Add(time.Second*5))
 377                 defer cancel()
 378                 cmdNoFollow := exec.CommandContext(ctxNoFollow, "go", "run", ".", "logs", "-poll=250ms", cr.UUID)
 379                 buf, err := cmdNoFollow.CombinedOutput()
 380                 c.Check(err, check.IsNil)
 381                 c.Check(string(buf), check.Matches, `(?ms).*line 1 of stderr\n`)
 382         }
 383
 384         time.Sleep(time.Second * 2)
 385         _, err = fmt.Fprintf(fCrunchrun, "%s line 2 of crunch-run.txt", time.Now().UTC().Format(rfc3339NanoFixed))
 386         c.Assert(err, check.IsNil)
 387         _, err = fmt.Fprintf(fStderr, "%s --end--", time.Now().UTC().Format(rfc3339NanoFixed))
 388         c.Assert(err, check.IsNil)
 389
 390         for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && !strings.Contains(stdout.String(), "--end--"); time.Sleep(time.Second / 10) {
 391         }
 392         c.Check(stdout.String(), check.Matches, `(?ms).*stderr\.txt +20\S+Z --end--\n.*`)
 393
 394         mtxt, err := cfs.MarshalManifest(".")
 395         c.Assert(err, check.IsNil)
 396         savedLog, err := rpcconn.CollectionCreate(ctx, arvados.CreateOptions{Attrs: map[string]interface{}{
 397                 "manifest_text": mtxt,
 398         }})
 399         c.Assert(err, check.IsNil)
 400         _, err = rpcconn.ContainerUpdate(ctx, arvados.UpdateOptions{UUID: cr.ContainerUUID, Attrs: map[string]interface{}{
 401                 "state":     arvados.ContainerStateComplete,
 402                 "log":       savedLog.PortableDataHash,
 403                 "output":    "d41d8cd98f00b204e9800998ecf8427e+0",
 404                 "exit_code": 0,
 405         }})
 406         c.Assert(err, check.IsNil)
 407
 408         err = cmd.Wait()
 409         c.Check(err, check.IsNil)
 410         // Ensure controller doesn't cheat by fetching data from the
 411         // gateway after the container is complete.
 412         gw.LogCollection = nil
 413
 414         c.Logf("re-running logs command on completed container")
 415         {
 416                 ctx, cancel := context.WithDeadline(ctx, time.Now().Add(time.Second*5))
 417                 defer cancel()
 418                 cmd := exec.CommandContext(ctx, "go", "run", ".", "logs", "-f", cr.UUID)
 419                 cmd.Env = append(cmd.Env, os.Environ()...)
 420                 cmd.Env = append(cmd.Env, "ARVADOS_API_TOKEN="+arvadostest.SystemRootToken)
 421                 buf, err := cmd.CombinedOutput()
 422                 c.Check(err, check.Equals, nil)
 423                 c.Check(string(buf), check.Matches, `(?ms).*--end--\n`)
 424         }
 425 }