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