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