lib/crunchrun/crunchrun_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package crunchrun
   6
   7 import (
   8         "bytes"
   9         "context"
  10         "crypto/md5"
  11         "encoding/json"
  12         "errors"
  13         "fmt"
  14         "io"
  15         "io/fs"
  16         "io/ioutil"
  17         "math/rand"
  18         "net/http"
  19         "net/http/httptest"
  20         "net/http/httputil"
  21         "net/url"
  22         "os"
  23         "os/exec"
  24         "path"
  25         "regexp"
  26         "runtime/pprof"
  27         "strings"
  28         "sync"
  29         "sync/atomic"
  30         "syscall"
  31         "testing"
  32         "time"
  33
  34         "git.arvados.org/arvados.git/lib/cloud"
  35         "git.arvados.org/arvados.git/lib/cmd"
  36         "git.arvados.org/arvados.git/sdk/go/arvados"
  37         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  38         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  39
  40         . "gopkg.in/check.v1"
  41 )
  42
  43 // Gocheck boilerplate
  44 func TestCrunchExec(t *testing.T) {
  45         TestingT(t)
  46 }
  47
  48 const logLineStart = `(?m)(.*\n)*\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d+Z `
  49
  50 var _ = Suite(&TestSuite{})
  51
  52 type TestSuite struct {
  53         client                   *arvados.Client
  54         api                      *ArvTestClient
  55         runner                   *ContainerRunner
  56         executor                 *stubExecutor
  57         keepmount                string
  58         keepmountTmp             []string
  59         testDispatcherKeepClient KeepTestClient
  60         testContainerKeepClient  KeepTestClient
  61         debian12MemoryCurrent    int64
  62         debian12SwapCurrent      int64
  63 }
  64
  65 func (s *TestSuite) SetUpSuite(c *C) {
  66         buf, err := os.ReadFile("../crunchstat/testdata/debian12/sys/fs/cgroup/user.slice/user-1000.slice/session-4.scope/memory.current")
  67         c.Assert(err, IsNil)
  68         _, err = fmt.Sscanf(string(buf), "%d", &s.debian12MemoryCurrent)
  69         c.Assert(err, IsNil)
  70
  71         buf, err = os.ReadFile("../crunchstat/testdata/debian12/sys/fs/cgroup/user.slice/user-1000.slice/session-4.scope/memory.swap.current")
  72         c.Assert(err, IsNil)
  73         _, err = fmt.Sscanf(string(buf), "%d", &s.debian12SwapCurrent)
  74         c.Assert(err, IsNil)
  75 }
  76
  77 func (s *TestSuite) SetUpTest(c *C) {
  78         s.client = arvados.NewClientFromEnv()
  79         s.executor = &stubExecutor{}
  80         var err error
  81         s.api = &ArvTestClient{}
  82         s.runner, err = NewContainerRunner(s.client, s.api, &s.testDispatcherKeepClient, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
  83         c.Assert(err, IsNil)
  84         s.runner.executor = s.executor
  85         s.runner.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) {
  86                 return s.api, &s.testContainerKeepClient, s.client, nil
  87         }
  88         s.runner.RunArvMount = func(cmd []string, tok string) (*exec.Cmd, error) {
  89                 s.runner.ArvMountPoint = s.keepmount
  90                 for i, opt := range cmd {
  91                         if opt == "--mount-tmp" {
  92                                 err := os.Mkdir(s.keepmount+"/"+cmd[i+1], 0700)
  93                                 if err != nil {
  94                                         return nil, err
  95                                 }
  96                                 s.keepmountTmp = append(s.keepmountTmp, cmd[i+1])
  97                         }
  98                 }
  99                 return nil, nil
 100         }
 101         s.keepmount = c.MkDir()
 102         err = os.Mkdir(s.keepmount+"/by_id", 0755)
 103         s.keepmountTmp = nil
 104         c.Assert(err, IsNil)
 105         err = os.Mkdir(s.keepmount+"/by_id/"+arvadostest.DockerImage112PDH, 0755)
 106         c.Assert(err, IsNil)
 107         err = ioutil.WriteFile(s.keepmount+"/by_id/"+arvadostest.DockerImage112PDH+"/"+arvadostest.DockerImage112Filename, []byte("#notarealtarball"), 0644)
 108         err = os.Mkdir(s.keepmount+"/by_id/"+fakeInputCollectionPDH, 0755)
 109         c.Assert(err, IsNil)
 110         err = ioutil.WriteFile(s.keepmount+"/by_id/"+fakeInputCollectionPDH+"/input.json", []byte(`{"input":true}`), 0644)
 111         c.Assert(err, IsNil)
 112         s.runner.ArvMountPoint = s.keepmount
 113         os.Setenv("InstanceType", `{"ProviderType":"a1.2xlarge","Price":1.2}`)
 114 }
 115
 116 type ArvTestClient struct {
 117         Total   int64
 118         Calls   int
 119         Content []arvadosclient.Dict
 120         arvados.Container
 121         secretMounts []byte
 122         sync.Mutex
 123         WasSetRunning bool
 124         callraw       bool
 125 }
 126
 127 type KeepTestClient struct {
 128         Called         bool
 129         StorageClasses []string
 130         blocks         sync.Map
 131 }
 132
 133 type stubExecutor struct {
 134         imageLoaded bool
 135         loaded      string
 136         loadErr     error
 137         exitCode    int
 138         createErr   error
 139         created     containerSpec
 140         startErr    error
 141         waitSleep   time.Duration
 142         waitErr     error
 143         stopErr     error
 144         stopped     bool
 145         closed      bool
 146         runFunc     func() int
 147         exit        chan int
 148 }
 149
 150 func (e *stubExecutor) LoadImage(imageId string, tarball string, container arvados.Container, keepMount string,
 151         containerClient *arvados.Client) error {
 152         e.loaded = tarball
 153         return e.loadErr
 154 }
 155 func (e *stubExecutor) Runtime() string                 { return "stub" }
 156 func (e *stubExecutor) Version() string                 { return "stub " + cmd.Version.String() }
 157 func (e *stubExecutor) Create(spec containerSpec) error { e.created = spec; return e.createErr }
 158 func (e *stubExecutor) Start() error {
 159         e.exit = make(chan int, 1)
 160         go func() { e.exit <- e.runFunc() }()
 161         return e.startErr
 162 }
 163 func (e *stubExecutor) Pid() int    { return 1115883 } // matches pid in ../crunchstat/testdata/debian12/proc/
 164 func (e *stubExecutor) Stop() error { e.stopped = true; go func() { e.exit <- -1 }(); return e.stopErr }
 165 func (e *stubExecutor) Close()      { e.closed = true }
 166 func (e *stubExecutor) Wait(context.Context) (int, error) {
 167         return <-e.exit, e.waitErr
 168 }
 169 func (e *stubExecutor) InjectCommand(ctx context.Context, _, _ string, _ bool, _ []string) (*exec.Cmd, error) {
 170         return nil, errors.New("unimplemented")
 171 }
 172 func (e *stubExecutor) IPAddress() (string, error) { return "", errors.New("unimplemented") }
 173
 174 const fakeInputCollectionPDH = "ffffffffaaaaaaaa88888888eeeeeeee+1234"
 175
 176 var hwManifest = ". 82ab40c24fc8df01798e57ba66795bb1+841216+Aa124ac75e5168396c73c0a18eda641a4f41791c0@569fa8c3 0:841216:9c31ee32b3d15268a0754e8edc74d4f815ee014b693bc5109058e431dd5caea7.tar\n"
 177 var hwPDH = "a45557269dcb65a6b78f9ac061c0850b+120"
 178 var hwImageID = "9c31ee32b3d15268a0754e8edc74d4f815ee014b693bc5109058e431dd5caea7"
 179
 180 var otherManifest = ". 68a84f561b1d1708c6baff5e019a9ab3+46+Ae5d0af96944a3690becb1decdf60cc1c937f556d@5693216f 0:46:md5sum.txt\n"
 181 var otherPDH = "a3e8f74c6f101eae01fa08bfb4e49b3a+54"
 182
 183 var normalizedManifestWithSubdirs = `. 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396c73c0abcdefgh11234567890@569fa8c3 0:9:file1_in_main.txt 9:18:file2_in_main.txt 0:27:zzzzz-8i9sb-bcdefghijkdhvnk.log.txt
 184 ./subdir1 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396cabcdefghij6419876543234@569fa8c4 0:9:file1_in_subdir1.txt 9:18:file2_in_subdir1.txt
 185 ./subdir1/subdir2 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396c73c0bcdefghijk544332211@569fa8c5 0:9:file1_in_subdir2.txt 9:18:file2_in_subdir2.txt
 186 `
 187
 188 var normalizedWithSubdirsPDH = "a0def87f80dd594d4675809e83bd4f15+367"
 189
 190 var denormalizedManifestWithSubdirs = ". 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396c73c0abcdefgh11234567890@569fa8c3 0:9:file1_in_main.txt 9:18:file2_in_main.txt 0:27:zzzzz-8i9sb-bcdefghijkdhvnk.log.txt 0:10:subdir1/file1_in_subdir1.txt 10:17:subdir1/file2_in_subdir1.txt\n"
 191 var denormalizedWithSubdirsPDH = "b0def87f80dd594d4675809e83bd4f15+367"
 192
 193 var fakeAuthUUID = "zzzzz-gj3su-55pqoyepgi2glem"
 194 var fakeAuthToken = "a3ltuwzqcu2u4sc0q7yhpc2w7s00fdcqecg5d6e0u3pfohmbjt"
 195
 196 func (client *ArvTestClient) Create(resourceType string,
 197         parameters arvadosclient.Dict,
 198         output interface{}) error {
 199
 200         client.Mutex.Lock()
 201         defer client.Mutex.Unlock()
 202
 203         client.Calls++
 204         client.Content = append(client.Content, parameters)
 205
 206         if resourceType == "logs" {
 207                 panic("logs.create called")
 208         }
 209
 210         if resourceType == "collections" && output != nil {
 211                 mt := parameters["collection"].(arvadosclient.Dict)["manifest_text"].(string)
 212                 md5sum := md5.Sum([]byte(mt))
 213                 outmap := output.(*arvados.Collection)
 214                 outmap.PortableDataHash = fmt.Sprintf("%x+%d", md5sum, len(mt))
 215                 outmap.UUID = fmt.Sprintf("zzzzz-4zz18-%015x", md5sum[:7])
 216         }
 217
 218         return nil
 219 }
 220
 221 func (client *ArvTestClient) Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error {
 222         switch {
 223         case method == "GET" && resourceType == "containers" && action == "auth":
 224                 return json.Unmarshal([]byte(`{
 225                         "kind": "arvados#api_client_authorization",
 226                         "uuid": "`+fakeAuthUUID+`",
 227                         "api_token": "`+fakeAuthToken+`"
 228                         }`), output)
 229         case method == "GET" && resourceType == "containers" && action == "secret_mounts":
 230                 if client.secretMounts != nil {
 231                         return json.Unmarshal(client.secretMounts, output)
 232                 }
 233                 return json.Unmarshal([]byte(`{"secret_mounts":{}}`), output)
 234         default:
 235                 return fmt.Errorf("Not found")
 236         }
 237 }
 238
 239 func (client *ArvTestClient) CallRaw(method, resourceType, uuid, action string,
 240         parameters arvadosclient.Dict) (reader io.ReadCloser, err error) {
 241         var j []byte
 242         if method == "GET" && resourceType == "nodes" && uuid == "" && action == "" {
 243                 j = []byte(`{
 244                         "kind": "arvados#nodeList",
 245                         "items": [{
 246                                 "uuid": "zzzzz-7ekkf-2z3mc76g2q73aio",
 247                                 "hostname": "compute2",
 248                                 "properties": {"total_cpu_cores": 16}
 249                         }]}`)
 250         } else if method == "GET" && resourceType == "containers" && action == "" && !client.callraw {
 251                 if uuid == "" {
 252                         j, err = json.Marshal(map[string]interface{}{
 253                                 "items": []interface{}{client.Container},
 254                                 "kind":  "arvados#nodeList",
 255                         })
 256                 } else {
 257                         j, err = json.Marshal(client.Container)
 258                 }
 259         } else {
 260                 j = []byte(`{
 261                         "command": ["sleep", "1"],
 262                         "container_image": "` + arvadostest.DockerImage112PDH + `",
 263                         "cwd": ".",
 264                         "environment": {},
 265                         "mounts": {"/tmp": {"kind": "tmp"}, "/json": {"kind": "json", "content": {"number": 123456789123456789}}},
 266                         "output_path": "/tmp",
 267                         "priority": 1,
 268                         "runtime_constraints": {}
 269                 }`)
 270         }
 271         return ioutil.NopCloser(bytes.NewReader(j)), err
 272 }
 273
 274 func (client *ArvTestClient) Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error {
 275         if resourceType == "collections" {
 276                 if uuid == hwPDH {
 277                         output.(*arvados.Collection).ManifestText = hwManifest
 278                 } else if uuid == otherPDH {
 279                         output.(*arvados.Collection).ManifestText = otherManifest
 280                 } else if uuid == normalizedWithSubdirsPDH {
 281                         output.(*arvados.Collection).ManifestText = normalizedManifestWithSubdirs
 282                 } else if uuid == denormalizedWithSubdirsPDH {
 283                         output.(*arvados.Collection).ManifestText = denormalizedManifestWithSubdirs
 284                 }
 285         }
 286         if resourceType == "containers" {
 287                 (*output.(*arvados.Container)) = client.Container
 288         }
 289         return nil
 290 }
 291
 292 func (client *ArvTestClient) Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) {
 293         client.Mutex.Lock()
 294         defer client.Mutex.Unlock()
 295         client.Calls++
 296         client.Content = append(client.Content, parameters)
 297         if resourceType == "containers" {
 298                 if parameters["container"].(arvadosclient.Dict)["state"] == "Running" {
 299                         client.WasSetRunning = true
 300                 }
 301         } else if resourceType == "collections" && output != nil {
 302                 mt := parameters["collection"].(arvadosclient.Dict)["manifest_text"].(string)
 303                 output.(*arvados.Collection).UUID = uuid
 304                 output.(*arvados.Collection).PortableDataHash = arvados.PortableDataHash(mt)
 305         }
 306         return nil
 307 }
 308
 309 var discoveryMap = map[string]interface{}{
 310         "crunchLogUpdateSize":   float64(crunchLogUpdateSize),
 311         "crunchLogUpdatePeriod": float64(crunchLogUpdatePeriod.Seconds()),
 312 }
 313
 314 func (client *ArvTestClient) Discovery(key string) (interface{}, error) {
 315         return discoveryMap[key], nil
 316 }
 317
 318 // CalledWith returns the parameters from the first API call whose
 319 // parameters match jpath/string. E.g., CalledWith(c, "foo.bar",
 320 // "baz") returns parameters with parameters["foo"]["bar"]=="baz". If
 321 // no call matches, it returns nil.
 322 func (client *ArvTestClient) CalledWith(jpath string, expect interface{}) arvadosclient.Dict {
 323 call:
 324         for _, content := range client.Content {
 325                 var v interface{} = content
 326                 for _, k := range strings.Split(jpath, ".") {
 327                         if dict, ok := v.(arvadosclient.Dict); !ok {
 328                                 continue call
 329                         } else {
 330                                 v = dict[k]
 331                         }
 332                 }
 333                 if v == expect {
 334                         return content
 335                 }
 336         }
 337         return nil
 338 }
 339
 340 func (client *KeepTestClient) LocalLocator(locator string) (string, error) {
 341         return locator, nil
 342 }
 343
 344 func (client *KeepTestClient) BlockWrite(_ context.Context, opts arvados.BlockWriteOptions) (arvados.BlockWriteResponse, error) {
 345         locator := fmt.Sprintf("%x+%d", md5.Sum(opts.Data), len(opts.Data))
 346         client.blocks.Store(locator, append([]byte(nil), opts.Data...))
 347         return arvados.BlockWriteResponse{
 348                 Locator: locator,
 349         }, nil
 350 }
 351
 352 func (client *KeepTestClient) BlockRead(_ context.Context, opts arvados.BlockReadOptions) (int, error) {
 353         loaded, ok := client.blocks.Load(opts.Locator)
 354         if !ok {
 355                 return 0, os.ErrNotExist
 356         }
 357         n, err := io.Copy(opts.WriteTo, bytes.NewReader(loaded.([]byte)))
 358         return int(n), err
 359 }
 360
 361 func (client *KeepTestClient) ReadAt(locator string, dst []byte, offset int) (int, error) {
 362         loaded, ok := client.blocks.Load(locator)
 363         if !ok {
 364                 return 0, os.ErrNotExist
 365         }
 366         data := loaded.([]byte)
 367         if offset >= len(data) {
 368                 return 0, io.EOF
 369         }
 370         return copy(dst, data[offset:]), nil
 371 }
 372
 373 func (client *KeepTestClient) Close() {
 374         client.blocks.Range(func(locator, value interface{}) bool {
 375                 client.blocks.Delete(locator)
 376                 return true
 377         })
 378 }
 379
 380 func (client *KeepTestClient) SetStorageClasses(sc []string) {
 381         client.StorageClasses = sc
 382 }
 383
 384 type FileWrapper struct {
 385         io.ReadCloser
 386         len int64
 387 }
 388
 389 func (fw FileWrapper) Readdir(n int) ([]os.FileInfo, error) {
 390         return nil, errors.New("not implemented")
 391 }
 392
 393 func (fw FileWrapper) Seek(int64, int) (int64, error) {
 394         return 0, errors.New("not implemented")
 395 }
 396
 397 func (fw FileWrapper) Size() int64 {
 398         return fw.len
 399 }
 400
 401 func (fw FileWrapper) Stat() (os.FileInfo, error) {
 402         return nil, errors.New("not implemented")
 403 }
 404
 405 func (fw FileWrapper) Truncate(int64) error {
 406         return errors.New("not implemented")
 407 }
 408
 409 func (fw FileWrapper) Write([]byte) (int, error) {
 410         return 0, errors.New("not implemented")
 411 }
 412
 413 func (fw FileWrapper) Sync() error {
 414         return errors.New("not implemented")
 415 }
 416
 417 func (fw FileWrapper) Snapshot() (*arvados.Subtree, error) {
 418         return nil, errors.New("not implemented")
 419 }
 420
 421 func (fw FileWrapper) Splice(*arvados.Subtree) error {
 422         return errors.New("not implemented")
 423 }
 424
 425 type apiStubServer struct {
 426         server    *httptest.Server
 427         proxy     *httputil.ReverseProxy
 428         intercept func(http.ResponseWriter, *http.Request) bool
 429
 430         container arvados.Container
 431 }
 432
 433 func apiStub() (*arvados.Client, *apiStubServer) {
 434         client := arvados.NewClientFromEnv()
 435         apistub := &apiStubServer{}
 436         apistub.server = httptest.NewTLSServer(apistub)
 437         apistub.proxy = httputil.NewSingleHostReverseProxy(&url.URL{Scheme: "https", Host: client.APIHost})
 438         if client.Insecure {
 439                 apistub.proxy.Transport = arvados.InsecureHTTPClient.Transport
 440         }
 441         client.APIHost = apistub.server.Listener.Addr().String()
 442         return client, apistub
 443 }
 444
 445 func (apistub *apiStubServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 446         if apistub.intercept != nil && apistub.intercept(w, r) {
 447                 return
 448         }
 449         if r.Method == "GET" && r.URL.Path == "/arvados/v1/collections/"+hwPDH {
 450                 json.NewEncoder(w).Encode(arvados.Collection{ManifestText: hwManifest})
 451                 return
 452         }
 453         if r.Method == "GET" && r.URL.Path == "/arvados/v1/collections/"+otherPDH {
 454                 json.NewEncoder(w).Encode(arvados.Collection{ManifestText: otherManifest})
 455                 return
 456         }
 457         if r.Method == "GET" && r.URL.Path == "/arvados/v1/collections/"+normalizedWithSubdirsPDH {
 458                 json.NewEncoder(w).Encode(arvados.Collection{ManifestText: normalizedManifestWithSubdirs})
 459                 return
 460         }
 461         if r.Method == "GET" && r.URL.Path == "/arvados/v1/collections/"+denormalizedWithSubdirsPDH {
 462                 json.NewEncoder(w).Encode(arvados.Collection{ManifestText: denormalizedManifestWithSubdirs})
 463                 return
 464         }
 465         if r.Method == "GET" && r.URL.Path == "/arvados/v1/containers/"+apistub.container.UUID {
 466                 json.NewEncoder(w).Encode(apistub.container)
 467                 return
 468         }
 469         apistub.proxy.ServeHTTP(w, r)
 470 }
 471
 472 func (s *TestSuite) TestLoadImage(c *C) {
 473         s.runner.Container.ContainerImage = arvadostest.DockerImage112PDH
 474         s.runner.Container.Mounts = map[string]arvados.Mount{
 475                 "/out": {Kind: "tmp", Writable: true},
 476         }
 477         s.runner.Container.OutputPath = "/out"
 478
 479         _, err := s.runner.SetupMounts()
 480         c.Assert(err, IsNil)
 481
 482         imageID, err := s.runner.LoadImage()
 483         c.Check(err, IsNil)
 484         c.Check(s.executor.loaded, Matches, ".*"+regexp.QuoteMeta(arvadostest.DockerImage112Filename))
 485         c.Check(imageID, Equals, strings.TrimSuffix(arvadostest.DockerImage112Filename, ".tar"))
 486
 487         s.runner.Container.ContainerImage = arvadostest.DockerImage112PDH
 488         s.executor.imageLoaded = false
 489         s.executor.loaded = ""
 490         s.executor.loadErr = errors.New("bork")
 491         imageID, err = s.runner.LoadImage()
 492         c.Check(err, ErrorMatches, ".*bork")
 493         c.Check(s.executor.loaded, Matches, ".*"+regexp.QuoteMeta(arvadostest.DockerImage112Filename))
 494
 495         s.runner.Container.ContainerImage = fakeInputCollectionPDH
 496         s.executor.imageLoaded = false
 497         s.executor.loaded = ""
 498         s.executor.loadErr = nil
 499         imageID, err = s.runner.LoadImage()
 500         c.Check(err, ErrorMatches, "image collection does not include a \\.tar image file")
 501         c.Check(s.executor.loaded, Equals, "")
 502 }
 503
 504 type ArvErrorTestClient struct{}
 505
 506 func (ArvErrorTestClient) Create(resourceType string,
 507         parameters arvadosclient.Dict,
 508         output interface{}) error {
 509         return nil
 510 }
 511
 512 func (ArvErrorTestClient) Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error {
 513         if method == "GET" && resourceType == "containers" && action == "auth" {
 514                 return nil
 515         }
 516         return errors.New("ArvError")
 517 }
 518
 519 func (ArvErrorTestClient) CallRaw(method, resourceType, uuid, action string,
 520         parameters arvadosclient.Dict) (reader io.ReadCloser, err error) {
 521         return nil, errors.New("ArvError")
 522 }
 523
 524 func (ArvErrorTestClient) Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error {
 525         return errors.New("ArvError")
 526 }
 527
 528 func (ArvErrorTestClient) Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) {
 529         return nil
 530 }
 531
 532 func (ArvErrorTestClient) Discovery(key string) (interface{}, error) {
 533         return discoveryMap[key], nil
 534 }
 535
 536 type KeepErrorTestClient struct {
 537         KeepTestClient
 538 }
 539
 540 func (*KeepErrorTestClient) BlockWrite(context.Context, arvados.BlockWriteOptions) (arvados.BlockWriteResponse, error) {
 541         return arvados.BlockWriteResponse{}, errors.New("KeepError")
 542 }
 543
 544 func (*KeepErrorTestClient) LocalLocator(string) (string, error) {
 545         return "", errors.New("KeepError")
 546 }
 547
 548 type KeepReadErrorTestClient struct {
 549         KeepTestClient
 550 }
 551
 552 func (*KeepReadErrorTestClient) ReadAt(string, []byte, int) (int, error) {
 553         return 0, errors.New("KeepError")
 554 }
 555
 556 func dockerLog(fd byte, msg string) []byte {
 557         by := []byte(msg)
 558         header := make([]byte, 8+len(by))
 559         header[0] = fd
 560         header[7] = byte(len(by))
 561         copy(header[8:], by)
 562         return header
 563 }
 564
 565 func (s *TestSuite) TestRunContainer(c *C) {
 566         s.executor.runFunc = func() int {
 567                 fmt.Fprintf(s.executor.created.Stdout, "Hello world\n")
 568                 return 0
 569         }
 570
 571         s.runner.Container.ContainerImage = arvadostest.DockerImage112PDH
 572         s.runner.Container.Command = []string{"./hw"}
 573         s.runner.Container.OutputStorageClasses = []string{"default"}
 574
 575         imageID, err := s.runner.LoadImage()
 576         c.Assert(err, IsNil)
 577
 578         err = s.runner.CreateContainer(imageID, nil)
 579         c.Assert(err, IsNil)
 580
 581         err = s.runner.StartContainer()
 582         c.Assert(err, IsNil)
 583
 584         err = s.runner.WaitFinish()
 585         c.Assert(err, IsNil)
 586
 587         c.Check(logFileContent(c, s.runner, "stdout.txt"), Matches, `2\S+Z Hello world\n`)
 588         c.Check(logFileContent(c, s.runner, "stderr.txt"), Matches, ``)
 589 }
 590
 591 func (s *TestSuite) TestCommitLogs(c *C) {
 592         api := &ArvTestClient{}
 593         kc := &KeepTestClient{}
 594         defer kc.Close()
 595         cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 596         c.Assert(err, IsNil)
 597         f, err := cr.openLogFile("crunch-run")
 598         c.Assert(err, IsNil)
 599         cr.CrunchLog = newLogWriter(newTestTimestamper(f))
 600
 601         cr.CrunchLog.Print("Hello world!")
 602         cr.CrunchLog.Print("Goodbye")
 603         cr.finalState = "Complete"
 604
 605         err = cr.CommitLogs()
 606         c.Check(err, IsNil)
 607
 608         c.Check(api.Calls, Equals, 1)
 609         c.Check(api.Content[0]["ensure_unique_name"], Equals, true)
 610         c.Check(api.Content[0]["collection"].(arvadosclient.Dict)["name"], Equals, "logs for zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 611         c.Check(api.Content[0]["collection"].(arvadosclient.Dict)["manifest_text"], Equals, ". 744b2e4553123b02fa7b452ec5c18993+123 0:123:crunch-run.txt\n")
 612         c.Check(*cr.LogsPDH, Equals, "63da7bdacf08c40f604daad80c261e9a+60")
 613 }
 614
 615 func (s *TestSuite) TestUpdateContainerRunning(c *C) {
 616         api := &ArvTestClient{}
 617         kc := &KeepTestClient{}
 618         defer kc.Close()
 619         cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 620         c.Assert(err, IsNil)
 621
 622         err = cr.UpdateContainerRunning("")
 623         c.Check(err, IsNil)
 624
 625         c.Check(api.Content[0]["container"].(arvadosclient.Dict)["state"], Equals, "Running")
 626 }
 627
 628 func (s *TestSuite) TestUpdateContainerComplete(c *C) {
 629         api := &ArvTestClient{}
 630         kc := &KeepTestClient{}
 631         defer kc.Close()
 632         cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 633         c.Assert(err, IsNil)
 634
 635         cr.LogsPDH = new(string)
 636         *cr.LogsPDH = "d3a229d2fe3690c2c3e75a71a153c6a3+60"
 637
 638         cr.ExitCode = new(int)
 639         *cr.ExitCode = 42
 640         cr.finalState = "Complete"
 641
 642         err = cr.UpdateContainerFinal()
 643         c.Check(err, IsNil)
 644
 645         c.Check(api.Content[0]["container"].(arvadosclient.Dict)["log"], Equals, *cr.LogsPDH)
 646         c.Check(api.Content[0]["container"].(arvadosclient.Dict)["exit_code"], Equals, *cr.ExitCode)
 647         c.Check(api.Content[0]["container"].(arvadosclient.Dict)["state"], Equals, "Complete")
 648 }
 649
 650 func (s *TestSuite) TestUpdateContainerCancelled(c *C) {
 651         api := &ArvTestClient{}
 652         kc := &KeepTestClient{}
 653         defer kc.Close()
 654         cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 655         c.Assert(err, IsNil)
 656         cr.cCancelled = true
 657         cr.finalState = "Cancelled"
 658
 659         err = cr.UpdateContainerFinal()
 660         c.Check(err, IsNil)
 661
 662         c.Check(api.Content[0]["container"].(arvadosclient.Dict)["log"], IsNil)
 663         c.Check(api.Content[0]["container"].(arvadosclient.Dict)["exit_code"], IsNil)
 664         c.Check(api.Content[0]["container"].(arvadosclient.Dict)["state"], Equals, "Cancelled")
 665 }
 666
 667 // Used by the TestFullRun*() test below to DRY up boilerplate setup to do full
 668 // dress rehearsal of the Run() function, starting from a JSON container record.
 669 func (s *TestSuite) fullRunHelper(c *C, record string, extraMounts []string, fn func() int) (*ArvTestClient, *ContainerRunner, string) {
 670         err := json.Unmarshal([]byte(record), &s.api.Container)
 671         c.Assert(err, IsNil)
 672         initialState := s.api.Container.State
 673
 674         var sm struct {
 675                 SecretMounts map[string]arvados.Mount `json:"secret_mounts"`
 676         }
 677         err = json.Unmarshal([]byte(record), &sm)
 678         c.Check(err, IsNil)
 679         secretMounts, err := json.Marshal(sm)
 680         c.Assert(err, IsNil)
 681         c.Logf("SecretMounts decoded %v json %q", sm, secretMounts)
 682
 683         s.executor.runFunc = fn
 684
 685         s.runner.statInterval = 100 * time.Millisecond
 686         s.runner.containerWatchdogInterval = time.Second
 687
 688         realTemp := c.MkDir()
 689         tempcount := 0
 690         s.runner.MkTempDir = func(_, prefix string) (string, error) {
 691                 tempcount++
 692                 d := fmt.Sprintf("%s/%s%d", realTemp, prefix, tempcount)
 693                 err := os.Mkdir(d, os.ModePerm)
 694                 if err != nil && strings.Contains(err.Error(), ": file exists") {
 695                         // Test case must have pre-populated the tempdir
 696                         err = nil
 697                 }
 698                 return d, err
 699         }
 700         client, _ := apiStub()
 701         s.runner.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) {
 702                 return &ArvTestClient{secretMounts: secretMounts}, &s.testContainerKeepClient, client, nil
 703         }
 704
 705         if extraMounts != nil && len(extraMounts) > 0 {
 706                 err := s.runner.SetupArvMountPoint("keep")
 707                 c.Check(err, IsNil)
 708
 709                 for _, m := range extraMounts {
 710                         os.MkdirAll(s.runner.ArvMountPoint+"/by_id/"+m, os.ModePerm)
 711                 }
 712         }
 713
 714         err = s.runner.Run()
 715         if s.api.CalledWith("container.state", "Complete") != nil {
 716                 c.Check(err, IsNil)
 717         }
 718         if s.executor.loadErr == nil && s.executor.createErr == nil && initialState != "Running" {
 719                 c.Check(s.api.WasSetRunning, Equals, true)
 720                 var lastupdate arvadosclient.Dict
 721                 for _, content := range s.api.Content {
 722                         if content["container"] != nil {
 723                                 lastupdate = content["container"].(arvadosclient.Dict)
 724                         }
 725                 }
 726                 if lastupdate["log"] == nil {
 727                         c.Errorf("no container update with non-nil log -- updates were: %v", s.api.Content)
 728                 }
 729         }
 730
 731         if err != nil {
 732                 dumpAllLogFiles(c, s.runner)
 733         }
 734
 735         return s.api, s.runner, realTemp
 736 }
 737
 738 func (s *TestSuite) TestFullRunHello(c *C) {
 739         s.runner.enableMemoryLimit = true
 740         s.runner.networkMode = "default"
 741         s.fullRunHelper(c, `{
 742     "command": ["echo", "hello world"],
 743     "container_image": "`+arvadostest.DockerImage112PDH+`",
 744     "cwd": ".",
 745     "environment": {"foo":"bar","baz":"waz"},
 746     "mounts": {"/tmp": {"kind": "tmp"} },
 747     "output_path": "/tmp",
 748     "priority": 1,
 749     "runtime_constraints": {"vcpus":1,"ram":1000000},
 750     "state": "Locked",
 751     "output_storage_classes": ["default"]
 752 }`, nil, func() int {
 753                 c.Check(s.executor.created.Command, DeepEquals, []string{"echo", "hello world"})
 754                 c.Check(s.executor.created.Image, Equals, "sha256:d8309758b8fe2c81034ffc8a10c36460b77db7bc5e7b448c4e5b684f9d95a678")
 755                 c.Check(s.executor.created.Env, DeepEquals, map[string]string{"foo": "bar", "baz": "waz"})
 756                 c.Check(s.executor.created.VCPUs, Equals, 1)
 757                 c.Check(s.executor.created.RAM, Equals, int64(1000000))
 758                 c.Check(s.executor.created.NetworkMode, Equals, "default")
 759                 c.Check(s.executor.created.EnableNetwork, Equals, false)
 760                 c.Check(s.executor.created.GPUDeviceCount, Equals, 0)
 761                 fmt.Fprintln(s.executor.created.Stdout, "hello world")
 762                 return 0
 763         })
 764
 765         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
 766         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
 767         c.Check(logFileContent(c, s.runner, "stdout.txt"), Matches, `2\S+Z hello world\n`)
 768         c.Check(s.testDispatcherKeepClient.StorageClasses, DeepEquals, []string{"default"})
 769         c.Check(s.testContainerKeepClient.StorageClasses, DeepEquals, []string{"default"})
 770 }
 771
 772 func (s *TestSuite) TestRunAlreadyRunning(c *C) {
 773         var ran bool
 774         s.fullRunHelper(c, `{
 775     "command": ["sleep", "3"],
 776     "container_image": "`+arvadostest.DockerImage112PDH+`",
 777     "cwd": ".",
 778     "environment": {},
 779     "mounts": {"/tmp": {"kind": "tmp"} },
 780     "output_path": "/tmp",
 781     "priority": 1,
 782     "runtime_constraints": {},
 783     "scheduling_parameters":{"max_run_time": 1},
 784     "state": "Running"
 785 }`, nil, func() int {
 786                 ran = true
 787                 return 2
 788         })
 789         c.Check(s.api.CalledWith("container.state", "Cancelled"), IsNil)
 790         c.Check(s.api.CalledWith("container.state", "Complete"), IsNil)
 791         c.Check(ran, Equals, false)
 792 }
 793
 794 func ec2MetadataServerStub(c *C, token *string, failureRate float64, stoptime *atomic.Value) *httptest.Server {
 795         failedOnce := false
 796         return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 797                 if !failedOnce || rand.Float64() < failureRate {
 798                         w.WriteHeader(http.StatusServiceUnavailable)
 799                         failedOnce = true
 800                         return
 801                 }
 802                 switch r.URL.Path {
 803                 case "/latest/api/token":
 804                         fmt.Fprintln(w, *token)
 805                 case "/latest/meta-data/spot/instance-action":
 806                         if r.Header.Get("X-aws-ec2-metadata-token") != *token {
 807                                 w.WriteHeader(http.StatusUnauthorized)
 808                         } else if t, _ := stoptime.Load().(time.Time); t.IsZero() {
 809                                 w.WriteHeader(http.StatusNotFound)
 810                         } else {
 811                                 fmt.Fprintf(w, `{"action":"stop","time":"%s"}`, t.Format(time.RFC3339))
 812                         }
 813                 default:
 814                         w.WriteHeader(http.StatusNotFound)
 815                 }
 816         }))
 817 }
 818
 819 func (s *TestSuite) TestSpotInterruptionNotice(c *C) {
 820         s.testSpotInterruptionNotice(c, 0.1)
 821 }
 822
 823 func (s *TestSuite) TestSpotInterruptionNoticeNotAvailable(c *C) {
 824         s.testSpotInterruptionNotice(c, 1)
 825 }
 826
 827 func (s *TestSuite) testSpotInterruptionNotice(c *C, failureRate float64) {
 828         var stoptime atomic.Value
 829         token := "fake-ec2-metadata-token"
 830         stub := ec2MetadataServerStub(c, &token, failureRate, &stoptime)
 831         defer stub.Close()
 832
 833         defer func(i time.Duration, u string) {
 834                 spotInterruptionCheckInterval = i
 835                 ec2MetadataBaseURL = u
 836         }(spotInterruptionCheckInterval, ec2MetadataBaseURL)
 837         spotInterruptionCheckInterval = time.Second / 8
 838         ec2MetadataBaseURL = stub.URL
 839
 840         checkedLogs := false
 841         checkLogs := func() {
 842                 checkedLogs = true
 843                 c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*Checking for spot instance interruptions every 125ms using instance metadata at http://.*`)
 844                 c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*Spot instance interruption check was inconclusive: 503 Service Unavailable -- will retry in 125ms.*`)
 845                 if failureRate == 1 {
 846                         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*Spot instance interruption check was inconclusive: 503 Service Unavailable -- now giving up after too many consecutive errors.*`)
 847                         return
 848                 }
 849                 text := `Cloud provider scheduled instance stop at ` + stoptime.Load().(time.Time).Format(time.RFC3339)
 850                 c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*`+text+`.*`)
 851                 c.Check(s.api.CalledWith("container.runtime_status.warning", "preemption notice"), NotNil)
 852                 c.Check(s.api.CalledWith("container.runtime_status.warningDetail", text), NotNil)
 853                 c.Check(s.api.CalledWith("container.runtime_status.preemptionNotice", text), NotNil)
 854
 855                 // Check that the log collection was saved, and the
 856                 // container record updated with the new PDH,
 857                 // immediately after the preemption notice was
 858                 // received -- i.e., while the container is still
 859                 // running.
 860                 lastpdh := ""
 861                 saved := make(map[string]string) // pdh => manifest_text
 862                 for _, call := range s.api.Content {
 863                         if ctr, ok := call["container"].(arvadosclient.Dict); ok {
 864                                 if pdh, ok := ctr["log"].(string); ok {
 865                                         lastpdh = pdh
 866                                 }
 867                         }
 868                         if coll, ok := call["collection"].(arvadosclient.Dict); ok {
 869                                 mt, _ := coll["manifest_text"].(string)
 870                                 if strings.Contains(mt, ":crunch-run.txt") {
 871                                         saved[arvados.PortableDataHash(mt)] = mt
 872                                 }
 873                         }
 874                 }
 875                 logfs, err := (&arvados.Collection{ManifestText: saved[lastpdh]}).FileSystem(s.runner.dispatcherClient, s.runner.DispatcherKeepClient)
 876                 c.Assert(err, IsNil)
 877                 log, err := fs.ReadFile(arvados.FS(logfs), "crunch-run.txt")
 878                 c.Check(err, IsNil)
 879                 c.Check(string(log), Matches, `(?ms).*\Q`+text+`\E.*`)
 880         }
 881
 882         go s.runner.checkSpotInterruptionNotices()
 883         s.fullRunHelper(c, `{
 884     "command": ["sleep", "3"],
 885     "container_image": "`+arvadostest.DockerImage112PDH+`",
 886     "cwd": ".",
 887     "environment": {},
 888     "mounts": {"/tmp": {"kind": "tmp"} },
 889     "output_path": "/tmp",
 890     "priority": 1,
 891     "runtime_constraints": {},
 892     "state": "Locked"
 893 }`, nil, func() int {
 894                 time.Sleep(time.Second)
 895                 stoptime.Store(time.Now().Add(time.Minute).UTC())
 896                 token = "different-fake-ec2-metadata-token"
 897                 time.Sleep(time.Second)
 898                 checkLogs()
 899                 return 0
 900         })
 901         c.Check(checkedLogs, Equals, true)
 902 }
 903
 904 func (s *TestSuite) TestRunTimeExceeded(c *C) {
 905         s.fullRunHelper(c, `{
 906     "command": ["sleep", "3"],
 907     "container_image": "`+arvadostest.DockerImage112PDH+`",
 908     "cwd": ".",
 909     "environment": {},
 910     "mounts": {"/tmp": {"kind": "tmp"} },
 911     "output_path": "/tmp",
 912     "priority": 1,
 913     "runtime_constraints": {},
 914     "scheduling_parameters":{"max_run_time": 1},
 915     "state": "Locked"
 916 }`, nil, func() int {
 917                 time.Sleep(3 * time.Second)
 918                 return 0
 919         })
 920
 921         c.Check(s.api.CalledWith("container.state", "Cancelled"), NotNil)
 922         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, "(?ms).*maximum run time exceeded.*")
 923 }
 924
 925 func (s *TestSuite) TestContainerWaitFails(c *C) {
 926         s.fullRunHelper(c, `{
 927     "command": ["sleep", "3"],
 928     "container_image": "`+arvadostest.DockerImage112PDH+`",
 929     "cwd": ".",
 930     "mounts": {"/tmp": {"kind": "tmp"} },
 931     "output_path": "/tmp",
 932     "priority": 1,
 933     "state": "Locked"
 934 }`, nil, func() int {
 935                 s.executor.waitErr = errors.New("Container is not running")
 936                 return 0
 937         })
 938
 939         c.Check(s.api.CalledWith("container.state", "Cancelled"), NotNil)
 940         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, "(?ms).*Container is not running.*")
 941 }
 942
 943 func (s *TestSuite) TestCrunchstat(c *C) {
 944         s.runner.crunchstatFakeFS = os.DirFS("../crunchstat/testdata/debian12")
 945         s.fullRunHelper(c, `{
 946                 "command": ["sleep", "1"],
 947                 "container_image": "`+arvadostest.DockerImage112PDH+`",
 948                 "cwd": ".",
 949                 "environment": {},
 950                 "mounts": {"/tmp": {"kind": "tmp"} },
 951                 "output_path": "/tmp",
 952                 "priority": 1,
 953                 "runtime_constraints": {},
 954                 "state": "Locked"
 955         }`, nil, func() int {
 956                 time.Sleep(time.Second)
 957                 return 0
 958         })
 959
 960         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
 961         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
 962
 963         c.Check(logFileContent(c, s.runner, "crunchstat.txt"), Matches, `(?ms).*`+reTimestamp+` mem \d+ swap \d+ pgmajfault \d+ rss.*`)
 964         c.Check(logFileContent(c, s.runner, "hoststat.txt"), Matches, `(?ms).*`+reTimestamp+` mem \d+ swap \d+ pgmajfault \d+ rss.*`)
 965
 966         // Check that we called (*crunchstat.Reporter)Stop().
 967         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*`+reTimestamp+` Maximum crunch-run memory rss usage was \d+ bytes\n.*`)
 968 }
 969
 970 func (s *TestSuite) TestNodeInfoLog(c *C) {
 971         s.fullRunHelper(c, `{
 972                 "command": ["true"],
 973                 "container_image": "`+arvadostest.DockerImage112PDH+`",
 974                 "cwd": ".",
 975                 "environment": {},
 976                 "mounts": {"/tmp": {"kind": "tmp"} },
 977                 "output_path": "/tmp",
 978                 "priority": 1,
 979                 "runtime_constraints": {},
 980                 "state": "Locked"
 981         }`, nil, func() int {
 982                 return 0
 983         })
 984
 985         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
 986         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
 987
 988         json := logFileContent(c, s.runner, "node.json")
 989         c.Check(json, Matches, `(?ms).*"ProviderType": *"a1\.2xlarge".*`)
 990         c.Check(json, Matches, `(?ms).*"Price": *1\.2.*`)
 991
 992         nodeinfo := logFileContent(c, s.runner, "node-info.txt")
 993         c.Check(nodeinfo, Matches, `(?ms).*Host Information.*`)
 994         c.Check(nodeinfo, Matches, `(?ms).*CPU Information.*`)
 995         c.Check(nodeinfo, Matches, `(?ms).*Memory Information.*`)
 996         c.Check(nodeinfo, Matches, `(?ms).*Disk Space.*`)
 997         c.Check(nodeinfo, Matches, `(?ms).*Disk INodes.*`)
 998 }
 999
1000 func (s *TestSuite) TestLogVersionAndRuntime(c *C) {
1001         s.fullRunHelper(c, `{
1002                 "command": ["sleep", "1"],
1003                 "container_image": "`+arvadostest.DockerImage112PDH+`",
1004                 "cwd": ".",
1005                 "environment": {},
1006                 "mounts": {"/tmp": {"kind": "tmp"} },
1007                 "output_path": "/tmp",
1008                 "priority": 1,
1009                 "runtime_constraints": {},
1010                 "state": "Locked"
1011         }`, nil, func() int {
1012                 return 0
1013         })
1014
1015         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*crunch-run \S+ \(go\S+\) start.*`)
1016         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*crunch-run process has uid=\d+\(.+\) gid=\d+\(.+\) groups=\d+\(.+\)(,\d+\(.+\))*\n.*`)
1017         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*Executing container: zzzzz-zzzzz-zzzzzzzzzzzzzzz.*`)
1018         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*Using container runtime: stub.*`)
1019 }
1020
1021 func (s *TestSuite) testLogRSSThresholds(c *C, ram int64, expected []int, notExpected int) {
1022         s.runner.crunchstatFakeFS = os.DirFS("../crunchstat/testdata/debian12")
1023         s.fullRunHelper(c, `{
1024                 "command": ["true"],
1025                 "container_image": "`+arvadostest.DockerImage112PDH+`",
1026                 "cwd": ".",
1027                 "environment": {},
1028                 "mounts": {"/tmp": {"kind": "tmp"} },
1029                 "output_path": "/tmp",
1030                 "priority": 1,
1031                 "runtime_constraints": {"ram": `+fmt.Sprintf("%d", ram)+`},
1032                 "state": "Locked"
1033         }`, nil, func() int { return 0 })
1034         logs := logFileContent(c, s.runner, "crunch-run.txt")
1035         c.Log("=== crunchstat logs")
1036         c.Log(logs)
1037         pattern := logLineStart + `Container using over %d%% of memory \(rss %d/%d bytes\)`
1038         var threshold int
1039         for _, threshold = range expected {
1040                 c.Check(logs, Matches, fmt.Sprintf(pattern, threshold, s.debian12MemoryCurrent, ram))
1041         }
1042         if notExpected > threshold {
1043                 c.Check(logs, Not(Matches), fmt.Sprintf(pattern, notExpected, s.debian12MemoryCurrent, ram))
1044         }
1045 }
1046
1047 func (s *TestSuite) TestLogNoRSSThresholds(c *C) {
1048         s.testLogRSSThresholds(c, s.debian12MemoryCurrent*10, []int{}, 90)
1049 }
1050
1051 func (s *TestSuite) TestLogSomeRSSThresholds(c *C) {
1052         onePercentRSS := s.debian12MemoryCurrent / 100
1053         s.testLogRSSThresholds(c, 102*onePercentRSS, []int{90, 95}, 99)
1054 }
1055
1056 func (s *TestSuite) TestLogAllRSSThresholds(c *C) {
1057         s.testLogRSSThresholds(c, s.debian12MemoryCurrent, []int{90, 95, 99}, 0)
1058 }
1059
1060 func (s *TestSuite) TestLogMaximaAfterRun(c *C) {
1061         s.runner.crunchstatFakeFS = os.DirFS("../crunchstat/testdata/debian12")
1062         s.runner.parentTemp = c.MkDir()
1063         s.fullRunHelper(c, `{
1064         "command": ["true"],
1065         "container_image": "`+arvadostest.DockerImage112PDH+`",
1066         "cwd": ".",
1067         "environment": {},
1068         "mounts": {"/tmp": {"kind": "tmp"} },
1069         "output_path": "/tmp",
1070         "priority": 1,
1071         "runtime_constraints": {"ram": `+fmt.Sprintf("%d", s.debian12MemoryCurrent*10)+`},
1072         "state": "Locked"
1073     }`, nil, func() int { return 0 })
1074         logs := logFileContent(c, s.runner, "crunch-run.txt")
1075         for _, expected := range []string{
1076                 `Maximum disk usage was \d+%, \d+/\d+ bytes`,
1077                 fmt.Sprintf(`Maximum container memory swap usage was %d bytes`, s.debian12SwapCurrent),
1078                 `Maximum container memory pgmajfault usage was \d+ faults`,
1079                 fmt.Sprintf(`Maximum container memory rss usage was 10%%, %d/%d bytes`, s.debian12MemoryCurrent, s.debian12MemoryCurrent*10),
1080                 `Maximum crunch-run memory rss usage was \d+ bytes`,
1081         } {
1082                 c.Check(logs, Matches, logLineStart+expected)
1083         }
1084 }
1085
1086 func (s *TestSuite) TestCommitNodeInfoBeforeStart(c *C) {
1087         var collection_create, container_update arvadosclient.Dict
1088         s.fullRunHelper(c, `{
1089                 "command": ["true"],
1090                 "container_image": "`+arvadostest.DockerImage112PDH+`",
1091                 "cwd": ".",
1092                 "environment": {},
1093                 "mounts": {"/tmp": {"kind": "tmp"} },
1094                 "output_path": "/tmp",
1095                 "priority": 1,
1096                 "runtime_constraints": {},
1097                 "state": "Locked",
1098                 "uuid": "zzzzz-dz642-202301121543210"
1099         }`, nil, func() int {
1100                 collection_create = s.api.CalledWith("ensure_unique_name", true)
1101                 container_update = s.api.CalledWith("container.state", "Running")
1102                 return 0
1103         })
1104
1105         c.Assert(collection_create, NotNil)
1106         log_collection := collection_create["collection"].(arvadosclient.Dict)
1107         c.Check(log_collection["name"], Equals, "logs for zzzzz-dz642-202301121543210")
1108         manifest_text := log_collection["manifest_text"].(string)
1109         // We check that the file size is at least two digits as an easy way to
1110         // check the file isn't empty.
1111         c.Check(manifest_text, Matches, `\. .+ \d+:\d{2,}:node-info\.txt( .+)?\n`)
1112         c.Check(manifest_text, Matches, `\. .+ \d+:\d{2,}:node\.json( .+)?\n`)
1113
1114         c.Assert(container_update, NotNil)
1115         // As of Arvados 2.5.0, the container update must specify its log in PDH
1116         // format for the API server to propagate it to container requests, which
1117         // is what we care about for this test.
1118         expect_pdh := fmt.Sprintf("%x+%d", md5.Sum([]byte(manifest_text)), len(manifest_text))
1119         c.Check(container_update["container"].(arvadosclient.Dict)["log"], Equals, expect_pdh)
1120 }
1121
1122 func (s *TestSuite) TestContainerRecordLog(c *C) {
1123         s.fullRunHelper(c, `{
1124                 "command": ["sleep", "1"],
1125                 "container_image": "`+arvadostest.DockerImage112PDH+`",
1126                 "cwd": ".",
1127                 "environment": {},
1128                 "mounts": {"/tmp": {"kind": "tmp"} },
1129                 "output_path": "/tmp",
1130                 "priority": 1,
1131                 "runtime_constraints": {},
1132                 "state": "Locked"
1133         }`, nil,
1134                 func() int {
1135                         time.Sleep(time.Second)
1136                         return 0
1137                 })
1138
1139         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1140         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1141
1142         c.Check(logFileContent(c, s.runner, "container.json"), Matches, `(?ms).*container_image.*`)
1143 }
1144
1145 func (s *TestSuite) TestFullRunStderr(c *C) {
1146         s.fullRunHelper(c, `{
1147     "command": ["/bin/sh", "-c", "echo hello ; echo world 1>&2 ; exit 1"],
1148     "container_image": "`+arvadostest.DockerImage112PDH+`",
1149     "cwd": ".",
1150     "environment": {},
1151     "mounts": {"/tmp": {"kind": "tmp"} },
1152     "output_path": "/tmp",
1153     "priority": 1,
1154     "runtime_constraints": {},
1155     "state": "Locked"
1156 }`, nil, func() int {
1157                 fmt.Fprintln(s.executor.created.Stdout, "hello")
1158                 fmt.Fprintln(s.executor.created.Stderr, "world")
1159                 return 1
1160         })
1161
1162         final := s.api.CalledWith("container.state", "Complete")
1163         c.Assert(final, NotNil)
1164         c.Check(final["container"].(arvadosclient.Dict)["exit_code"], Equals, 1)
1165         c.Check(final["container"].(arvadosclient.Dict)["log"], NotNil)
1166
1167         c.Check(logFileContent(c, s.runner, "stdout.txt"), Matches, ".*hello\n")
1168         c.Check(logFileContent(c, s.runner, "stderr.txt"), Matches, ".*world\n")
1169 }
1170
1171 func (s *TestSuite) TestFullRunDefaultCwd(c *C) {
1172         s.fullRunHelper(c, `{
1173     "command": ["pwd"],
1174     "container_image": "`+arvadostest.DockerImage112PDH+`",
1175     "cwd": ".",
1176     "environment": {},
1177     "mounts": {"/tmp": {"kind": "tmp"} },
1178     "output_path": "/tmp",
1179     "priority": 1,
1180     "runtime_constraints": {},
1181     "state": "Locked"
1182 }`, nil, func() int {
1183                 fmt.Fprintf(s.executor.created.Stdout, "workdir=%q", s.executor.created.WorkingDir)
1184                 return 0
1185         })
1186
1187         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1188         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1189         c.Check(logFileContent(c, s.runner, "stdout.txt"), Matches, `.*workdir=""`)
1190 }
1191
1192 func (s *TestSuite) TestFullRunSetCwd(c *C) {
1193         s.fullRunHelper(c, `{
1194     "command": ["pwd"],
1195     "container_image": "`+arvadostest.DockerImage112PDH+`",
1196     "cwd": "/bin",
1197     "environment": {},
1198     "mounts": {"/tmp": {"kind": "tmp"} },
1199     "output_path": "/tmp",
1200     "priority": 1,
1201     "runtime_constraints": {},
1202     "state": "Locked"
1203 }`, nil, func() int {
1204                 fmt.Fprintln(s.executor.created.Stdout, s.executor.created.WorkingDir)
1205                 return 0
1206         })
1207
1208         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1209         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1210         c.Check(logFileContent(c, s.runner, "stdout.txt"), Matches, ".*/bin\n")
1211 }
1212
1213 func (s *TestSuite) TestFullRunSetOutputStorageClasses(c *C) {
1214         s.fullRunHelper(c, `{
1215     "command": ["pwd"],
1216     "container_image": "`+arvadostest.DockerImage112PDH+`",
1217     "cwd": "/bin",
1218     "environment": {},
1219     "mounts": {"/tmp": {"kind": "tmp"} },
1220     "output_path": "/tmp",
1221     "priority": 1,
1222     "runtime_constraints": {},
1223     "state": "Locked",
1224     "output_storage_classes": ["foo", "bar"]
1225 }`, nil, func() int {
1226                 fmt.Fprintln(s.executor.created.Stdout, s.executor.created.WorkingDir)
1227                 return 0
1228         })
1229
1230         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1231         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1232         c.Check(logFileContent(c, s.runner, "stdout.txt"), Matches, ".*/bin\n")
1233         c.Check(s.testDispatcherKeepClient.StorageClasses, DeepEquals, []string{"foo", "bar"})
1234         c.Check(s.testContainerKeepClient.StorageClasses, DeepEquals, []string{"foo", "bar"})
1235 }
1236
1237 func (s *TestSuite) TestEnableCUDADeviceCount(c *C) {
1238         s.fullRunHelper(c, `{
1239     "command": ["pwd"],
1240     "container_image": "`+arvadostest.DockerImage112PDH+`",
1241     "cwd": "/bin",
1242     "environment": {},
1243     "mounts": {"/tmp": {"kind": "tmp"} },
1244     "output_path": "/tmp",
1245     "priority": 1,
1246     "runtime_constraints": {"gpu": {"device_count": 2, "stack": "cuda", "hardware_target": ["9.0"], "driver_version": "11.0", "vram": 8000000000}},
1247     "state": "Locked",
1248     "output_storage_classes": ["foo", "bar"]
1249 }`, nil, func() int {
1250                 fmt.Fprintln(s.executor.created.Stdout, "ok")
1251                 return 0
1252         })
1253         c.Check(s.executor.created.GPUDeviceCount, Equals, 2)
1254         c.Check(s.executor.created.GPUStack, Equals, "cuda")
1255 }
1256
1257 func (s *TestSuite) TestStopOnSignal(c *C) {
1258         s.executor.runFunc = func() int {
1259                 s.executor.created.Stdout.Write([]byte("foo\n"))
1260                 s.runner.SigChan <- syscall.SIGINT
1261                 time.Sleep(10 * time.Second)
1262                 return 0
1263         }
1264         s.testStopContainer(c)
1265 }
1266
1267 func (s *TestSuite) TestStopOnArvMountDeath(c *C) {
1268         s.executor.runFunc = func() int {
1269                 s.executor.created.Stdout.Write([]byte("foo\n"))
1270                 s.runner.ArvMountExit <- nil
1271                 close(s.runner.ArvMountExit)
1272                 time.Sleep(10 * time.Second)
1273                 return 0
1274         }
1275         s.runner.ArvMountExit = make(chan error)
1276         s.testStopContainer(c)
1277 }
1278
1279 func (s *TestSuite) testStopContainer(c *C) {
1280         record := `{
1281     "command": ["/bin/sh", "-c", "echo foo && sleep 30 && echo bar"],
1282     "container_image": "` + arvadostest.DockerImage112PDH + `",
1283     "cwd": ".",
1284     "environment": {},
1285     "mounts": {"/tmp": {"kind": "tmp"} },
1286     "output_path": "/tmp",
1287     "priority": 1,
1288     "runtime_constraints": {},
1289     "state": "Locked"
1290 }`
1291
1292         err := json.Unmarshal([]byte(record), &s.api.Container)
1293         c.Assert(err, IsNil)
1294
1295         s.runner.RunArvMount = func([]string, string) (*exec.Cmd, error) { return nil, nil }
1296         s.runner.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) {
1297                 return &ArvTestClient{}, &KeepTestClient{}, nil, nil
1298         }
1299
1300         done := make(chan error)
1301         go func() {
1302                 done <- s.runner.Run()
1303         }()
1304         select {
1305         case <-time.After(20 * time.Second):
1306                 pprof.Lookup("goroutine").WriteTo(os.Stderr, 1)
1307                 c.Fatal("timed out")
1308         case err = <-done:
1309                 c.Check(err, IsNil)
1310         }
1311         dumpAllLogFiles(c, s.runner)
1312
1313         c.Check(s.api.CalledWith("container.log", nil), NotNil)
1314         c.Check(s.api.CalledWith("container.state", "Cancelled"), NotNil)
1315         c.Check(logFileContent(c, s.runner, "stdout.txt"), Matches, "(?ms).*foo\n$")
1316 }
1317
1318 func (s *TestSuite) TestFullRunSetEnv(c *C) {
1319         s.fullRunHelper(c, `{
1320     "command": ["/bin/sh", "-c", "echo $FROBIZ"],
1321     "container_image": "`+arvadostest.DockerImage112PDH+`",
1322     "cwd": "/bin",
1323     "environment": {"FROBIZ": "bilbo"},
1324     "mounts": {"/tmp": {"kind": "tmp"} },
1325     "output_path": "/tmp",
1326     "priority": 1,
1327     "runtime_constraints": {},
1328     "state": "Locked"
1329 }`, nil, func() int {
1330                 fmt.Fprintf(s.executor.created.Stdout, "%v", s.executor.created.Env)
1331                 return 0
1332         })
1333
1334         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1335         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1336         c.Check(logFileContent(c, s.runner, "stdout.txt"), Matches, `.*map\[FROBIZ:bilbo\]`)
1337 }
1338
1339 type ArvMountCmdLine struct {
1340         Cmd   []string
1341         token string
1342 }
1343
1344 func (am *ArvMountCmdLine) ArvMountTest(c []string, token string) (*exec.Cmd, error) {
1345         am.Cmd = c
1346         am.token = token
1347         return nil, nil
1348 }
1349
1350 func stubCert(c *C, temp string) string {
1351         path := temp + "/ca-certificates.crt"
1352         err := os.WriteFile(path, []byte{}, 0666)
1353         c.Assert(err, IsNil)
1354         os.Setenv("SSL_CERT_FILE", path)
1355         return path
1356 }
1357
1358 func (s *TestSuite) TestSetupMounts(c *C) {
1359         cr := s.runner
1360         am := &ArvMountCmdLine{}
1361         cr.RunArvMount = am.ArvMountTest
1362         cr.containerClient, _ = apiStub()
1363         cr.ContainerArvClient = &ArvTestClient{}
1364         cr.ContainerKeepClient = &KeepTestClient{}
1365         cr.Container.OutputStorageClasses = []string{"default"}
1366
1367         realTemp := c.MkDir()
1368         certTemp := c.MkDir()
1369         stubCertPath := stubCert(c, certTemp)
1370         cr.parentTemp = realTemp
1371
1372         i := 0
1373         cr.MkTempDir = func(_ string, prefix string) (string, error) {
1374                 i++
1375                 d := fmt.Sprintf("%s/%s%d", realTemp, prefix, i)
1376                 err := os.Mkdir(d, os.ModePerm)
1377                 if err != nil && strings.Contains(err.Error(), ": file exists") {
1378                         // Test case must have pre-populated the tempdir
1379                         err = nil
1380                 }
1381                 return d, err
1382         }
1383
1384         checkEmpty := func() {
1385                 // Should be deleted.
1386                 _, err := os.Stat(realTemp)
1387                 c.Assert(os.IsNotExist(err), Equals, true)
1388
1389                 // Now recreate it for the next test.
1390                 c.Assert(os.Mkdir(realTemp, 0777), IsNil)
1391         }
1392
1393         {
1394                 i = 0
1395                 cr.ArvMountPoint = ""
1396                 cr.Container.Mounts = make(map[string]arvados.Mount)
1397                 cr.Container.Mounts["/tmp"] = arvados.Mount{Kind: "tmp"}
1398                 cr.Container.OutputPath = "/tmp"
1399                 cr.statInterval = 5 * time.Second
1400                 bindmounts, err := cr.SetupMounts()
1401                 c.Check(err, IsNil)
1402                 c.Check(am.Cmd, DeepEquals, []string{"arv-mount", "--foreground",
1403                         "--read-write", "--storage-classes", "default", "--crunchstat-interval=5",
1404                         "--mount-by-pdh", "by_id", "--disable-event-listening", "--mount-by-id", "by_uuid", realTemp + "/keep1"})
1405                 c.Check(bindmounts, DeepEquals, map[string]bindmount{"/tmp": {realTemp + "/tmp2", false}})
1406                 os.RemoveAll(cr.ArvMountPoint)
1407                 cr.CleanupDirs()
1408                 checkEmpty()
1409         }
1410
1411         {
1412                 i = 0
1413                 cr.ArvMountPoint = ""
1414                 cr.Container.Mounts = make(map[string]arvados.Mount)
1415                 cr.Container.Mounts["/out"] = arvados.Mount{Kind: "tmp"}
1416                 cr.Container.Mounts["/tmp"] = arvados.Mount{Kind: "tmp"}
1417                 cr.Container.OutputPath = "/out"
1418                 cr.Container.OutputStorageClasses = []string{"foo", "bar"}
1419
1420                 bindmounts, err := cr.SetupMounts()
1421                 c.Check(err, IsNil)
1422                 c.Check(am.Cmd, DeepEquals, []string{"arv-mount", "--foreground",
1423                         "--read-write", "--storage-classes", "foo,bar", "--crunchstat-interval=5",
1424                         "--mount-by-pdh", "by_id", "--disable-event-listening", "--mount-by-id", "by_uuid", realTemp + "/keep1"})
1425                 c.Check(bindmounts, DeepEquals, map[string]bindmount{"/out": {realTemp + "/tmp2", false}, "/tmp": {realTemp + "/tmp3", false}})
1426                 os.RemoveAll(cr.ArvMountPoint)
1427                 cr.CleanupDirs()
1428                 checkEmpty()
1429         }
1430
1431         {
1432                 i = 0
1433                 cr.ArvMountPoint = ""
1434                 cr.Container.Mounts = make(map[string]arvados.Mount)
1435                 cr.Container.Mounts["/tmp"] = arvados.Mount{Kind: "tmp"}
1436                 cr.Container.OutputPath = "/tmp"
1437                 cr.Container.RuntimeConstraints.API = true
1438                 cr.Container.OutputStorageClasses = []string{"default"}
1439
1440                 bindmounts, err := cr.SetupMounts()
1441                 c.Check(err, IsNil)
1442                 c.Check(am.Cmd, DeepEquals, []string{"arv-mount", "--foreground",
1443                         "--read-write", "--storage-classes", "default", "--crunchstat-interval=5",
1444                         "--mount-by-pdh", "by_id", "--disable-event-listening", "--mount-by-id", "by_uuid", realTemp + "/keep1"})
1445                 c.Check(bindmounts, DeepEquals, map[string]bindmount{"/tmp": {realTemp + "/tmp2", false}, "/etc/arvados/ca-certificates.crt": {stubCertPath, true}})
1446                 os.RemoveAll(cr.ArvMountPoint)
1447                 cr.CleanupDirs()
1448                 checkEmpty()
1449
1450                 cr.Container.RuntimeConstraints.API = false
1451         }
1452
1453         {
1454                 i = 0
1455                 cr.ArvMountPoint = ""
1456                 cr.Container.Mounts = map[string]arvados.Mount{
1457                         "/keeptmp": {Kind: "collection", Writable: true},
1458                 }
1459                 cr.Container.OutputPath = "/keeptmp"
1460
1461                 os.MkdirAll(realTemp+"/keep1/tmp0", os.ModePerm)
1462
1463                 bindmounts, err := cr.SetupMounts()
1464                 c.Check(err, IsNil)
1465                 c.Check(am.Cmd, DeepEquals, []string{"arv-mount", "--foreground",
1466                         "--read-write", "--storage-classes", "default", "--crunchstat-interval=5",
1467                         "--mount-tmp", "tmp0", "--mount-by-pdh", "by_id", "--disable-event-listening", "--mount-by-id", "by_uuid", realTemp + "/keep1"})
1468                 c.Check(bindmounts, DeepEquals, map[string]bindmount{"/keeptmp": {realTemp + "/keep1/tmp0", false}})
1469                 os.RemoveAll(cr.ArvMountPoint)
1470                 cr.CleanupDirs()
1471                 checkEmpty()
1472         }
1473
1474         {
1475                 i = 0
1476                 cr.ArvMountPoint = ""
1477                 cr.Container.Mounts = map[string]arvados.Mount{
1478                         "/keepinp": {Kind: "collection", PortableDataHash: "59389a8f9ee9d399be35462a0f92541c+53"},
1479                         "/keepout": {Kind: "collection", Writable: true},
1480                 }
1481                 cr.Container.OutputPath = "/keepout"
1482
1483                 os.MkdirAll(realTemp+"/keep1/by_id/59389a8f9ee9d399be35462a0f92541c+53", os.ModePerm)
1484                 os.MkdirAll(realTemp+"/keep1/tmp0", os.ModePerm)
1485
1486                 bindmounts, err := cr.SetupMounts()
1487                 c.Check(err, IsNil)
1488                 c.Check(am.Cmd, DeepEquals, []string{"arv-mount", "--foreground",
1489                         "--read-write", "--storage-classes", "default", "--crunchstat-interval=5",
1490                         "--mount-tmp", "tmp0", "--mount-by-pdh", "by_id", "--disable-event-listening", "--mount-by-id", "by_uuid", realTemp + "/keep1"})
1491                 c.Check(bindmounts, DeepEquals, map[string]bindmount{
1492                         "/keepinp": {realTemp + "/keep1/by_id/59389a8f9ee9d399be35462a0f92541c+53", true},
1493                         "/keepout": {realTemp + "/keep1/tmp0", false},
1494                 })
1495                 os.RemoveAll(cr.ArvMountPoint)
1496                 cr.CleanupDirs()
1497                 checkEmpty()
1498         }
1499
1500         {
1501                 i = 0
1502                 cr.ArvMountPoint = ""
1503                 cr.Container.RuntimeConstraints.KeepCacheRAM = 512
1504                 cr.Container.Mounts = map[string]arvados.Mount{
1505                         "/keepinp": {Kind: "collection", PortableDataHash: "59389a8f9ee9d399be35462a0f92541c+53"},
1506                         "/keepout": {Kind: "collection", Writable: true},
1507                 }
1508                 cr.Container.OutputPath = "/keepout"
1509
1510                 os.MkdirAll(realTemp+"/keep1/by_id/59389a8f9ee9d399be35462a0f92541c+53", os.ModePerm)
1511                 os.MkdirAll(realTemp+"/keep1/tmp0", os.ModePerm)
1512
1513                 bindmounts, err := cr.SetupMounts()
1514                 c.Check(err, IsNil)
1515                 c.Check(am.Cmd, DeepEquals, []string{"arv-mount", "--foreground",
1516                         "--read-write", "--storage-classes", "default", "--crunchstat-interval=5", "--ram-cache",
1517                         "--file-cache", "512", "--mount-tmp", "tmp0", "--mount-by-pdh", "by_id", "--disable-event-listening", "--mount-by-id", "by_uuid", realTemp + "/keep1"})
1518                 c.Check(bindmounts, DeepEquals, map[string]bindmount{
1519                         "/keepinp": {realTemp + "/keep1/by_id/59389a8f9ee9d399be35462a0f92541c+53", true},
1520                         "/keepout": {realTemp + "/keep1/tmp0", false},
1521                 })
1522                 os.RemoveAll(cr.ArvMountPoint)
1523                 cr.CleanupDirs()
1524                 checkEmpty()
1525         }
1526
1527         for _, test := range []struct {
1528                 in  interface{}
1529                 out string
1530         }{
1531                 {in: "foo", out: `"foo"`},
1532                 {in: nil, out: `null`},
1533                 {in: map[string]int64{"foo": 123456789123456789}, out: `{"foo":123456789123456789}`},
1534         } {
1535                 i = 0
1536                 cr.ArvMountPoint = ""
1537                 cr.Container.Mounts = map[string]arvados.Mount{
1538                         "/mnt/test.json": {Kind: "json", Content: test.in},
1539                 }
1540                 bindmounts, err := cr.SetupMounts()
1541                 c.Check(err, IsNil)
1542                 c.Check(bindmounts, DeepEquals, map[string]bindmount{
1543                         "/mnt/test.json": {realTemp + "/json2/mountdata.json", true},
1544                 })
1545                 content, err := ioutil.ReadFile(realTemp + "/json2/mountdata.json")
1546                 c.Check(err, IsNil)
1547                 c.Check(content, DeepEquals, []byte(test.out))
1548                 os.RemoveAll(cr.ArvMountPoint)
1549                 cr.CleanupDirs()
1550                 checkEmpty()
1551         }
1552
1553         for _, test := range []struct {
1554                 in  interface{}
1555                 out string
1556         }{
1557                 {in: "foo", out: `foo`},
1558                 {in: nil, out: "error"},
1559                 {in: map[string]int64{"foo": 123456789123456789}, out: "error"},
1560         } {
1561                 i = 0
1562                 cr.ArvMountPoint = ""
1563                 cr.Container.Mounts = map[string]arvados.Mount{
1564                         "/mnt/test.txt": {Kind: "text", Content: test.in},
1565                 }
1566                 bindmounts, err := cr.SetupMounts()
1567                 if test.out == "error" {
1568                         c.Check(err.Error(), Equals, "content for mount \"/mnt/test.txt\" must be a string")
1569                 } else {
1570                         c.Check(err, IsNil)
1571                         c.Check(bindmounts, DeepEquals, map[string]bindmount{
1572                                 "/mnt/test.txt": {realTemp + "/text2/mountdata.text", true},
1573                         })
1574                         content, err := ioutil.ReadFile(realTemp + "/text2/mountdata.text")
1575                         c.Check(err, IsNil)
1576                         c.Check(content, DeepEquals, []byte(test.out))
1577                 }
1578                 os.RemoveAll(cr.ArvMountPoint)
1579                 cr.CleanupDirs()
1580                 checkEmpty()
1581         }
1582
1583         // Read-only mount points are allowed underneath output_dir mount point
1584         {
1585                 i = 0
1586                 cr.ArvMountPoint = ""
1587                 cr.Container.Mounts = make(map[string]arvados.Mount)
1588                 cr.Container.Mounts = map[string]arvados.Mount{
1589                         "/tmp":     {Kind: "tmp"},
1590                         "/tmp/foo": {Kind: "collection"},
1591                 }
1592                 cr.Container.OutputPath = "/tmp"
1593
1594                 os.MkdirAll(realTemp+"/keep1/tmp0", os.ModePerm)
1595
1596                 bindmounts, err := cr.SetupMounts()
1597                 c.Check(err, IsNil)
1598                 c.Check(am.Cmd, DeepEquals, []string{"arv-mount", "--foreground",
1599                         "--read-write", "--storage-classes", "default", "--crunchstat-interval=5", "--ram-cache",
1600                         "--file-cache", "512", "--mount-tmp", "tmp0", "--mount-by-pdh", "by_id", "--disable-event-listening", "--mount-by-id", "by_uuid", realTemp + "/keep1"})
1601                 c.Check(bindmounts, DeepEquals, map[string]bindmount{
1602                         "/tmp":     {realTemp + "/tmp2", false},
1603                         "/tmp/foo": {realTemp + "/keep1/tmp0", true},
1604                 })
1605                 os.RemoveAll(cr.ArvMountPoint)
1606                 cr.CleanupDirs()
1607                 checkEmpty()
1608         }
1609
1610         // Writable mount points copied to output_dir mount point
1611         {
1612                 i = 0
1613                 cr.ArvMountPoint = ""
1614                 cr.Container.Mounts = make(map[string]arvados.Mount)
1615                 cr.Container.Mounts = map[string]arvados.Mount{
1616                         "/tmp": {Kind: "tmp"},
1617                         "/tmp/foo": {Kind: "collection",
1618                                 PortableDataHash: "59389a8f9ee9d399be35462a0f92541c+53",
1619                                 Writable:         true},
1620                         "/tmp/bar": {Kind: "collection",
1621                                 PortableDataHash: "59389a8f9ee9d399be35462a0f92541d+53",
1622                                 Path:             "baz",
1623                                 Writable:         true},
1624                 }
1625                 cr.Container.OutputPath = "/tmp"
1626
1627                 os.MkdirAll(realTemp+"/keep1/by_id/59389a8f9ee9d399be35462a0f92541c+53", os.ModePerm)
1628                 os.MkdirAll(realTemp+"/keep1/by_id/59389a8f9ee9d399be35462a0f92541d+53/baz", os.ModePerm)
1629
1630                 rf, _ := os.Create(realTemp + "/keep1/by_id/59389a8f9ee9d399be35462a0f92541d+53/baz/quux")
1631                 rf.Write([]byte("bar"))
1632                 rf.Close()
1633
1634                 _, err := cr.SetupMounts()
1635                 c.Check(err, IsNil)
1636                 _, err = os.Stat(cr.HostOutputDir + "/foo")
1637                 c.Check(err, IsNil)
1638                 _, err = os.Stat(cr.HostOutputDir + "/bar/quux")
1639                 c.Check(err, IsNil)
1640                 os.RemoveAll(cr.ArvMountPoint)
1641                 cr.CleanupDirs()
1642                 checkEmpty()
1643         }
1644
1645         // Only mount points of kind 'collection' are allowed underneath output_dir mount point
1646         {
1647                 i = 0
1648                 cr.ArvMountPoint = ""
1649                 cr.Container.Mounts = make(map[string]arvados.Mount)
1650                 cr.Container.Mounts = map[string]arvados.Mount{
1651                         "/tmp":     {Kind: "tmp"},
1652                         "/tmp/foo": {Kind: "tmp"},
1653                 }
1654                 cr.Container.OutputPath = "/tmp"
1655
1656                 _, err := cr.SetupMounts()
1657                 c.Check(err, NotNil)
1658                 c.Check(err, ErrorMatches, `only mount points of kind 'collection', 'text' or 'json' are supported underneath the output_path.*`)
1659                 os.RemoveAll(cr.ArvMountPoint)
1660                 cr.CleanupDirs()
1661                 checkEmpty()
1662         }
1663
1664         // Only mount point of kind 'collection' is allowed for stdin
1665         {
1666                 i = 0
1667                 cr.ArvMountPoint = ""
1668                 cr.Container.Mounts = make(map[string]arvados.Mount)
1669                 cr.Container.Mounts = map[string]arvados.Mount{
1670                         "stdin": {Kind: "tmp"},
1671                 }
1672
1673                 _, err := cr.SetupMounts()
1674                 c.Check(err, NotNil)
1675                 c.Check(err, ErrorMatches, `unsupported mount kind 'tmp' for stdin.*`)
1676                 os.RemoveAll(cr.ArvMountPoint)
1677                 cr.CleanupDirs()
1678                 checkEmpty()
1679         }
1680 }
1681
1682 func (s *TestSuite) TestStdout(c *C) {
1683         helperRecord := `{
1684                 "command": ["/bin/sh", "-c", "echo $FROBIZ"],
1685                 "container_image": "` + arvadostest.DockerImage112PDH + `",
1686                 "cwd": "/bin",
1687                 "environment": {"FROBIZ": "bilbo"},
1688                 "mounts": {"/tmp": {"kind": "tmp"}, "stdout": {"kind": "file", "path": "/tmp/a/b/c.out"} },
1689                 "output_path": "/tmp",
1690                 "priority": 1,
1691                 "runtime_constraints": {},
1692                 "state": "Locked"
1693         }`
1694
1695         s.fullRunHelper(c, helperRecord, nil, func() int {
1696                 fmt.Fprintln(s.executor.created.Stdout, s.executor.created.Env["FROBIZ"])
1697                 return 0
1698         })
1699
1700         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1701         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1702         c.Check(s.runner.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", "./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out\n"), NotNil)
1703 }
1704
1705 // Used by the TestStdoutWithWrongPath*()
1706 func (s *TestSuite) stdoutErrorRunHelper(c *C, record string, fn func() int) (*ArvTestClient, *ContainerRunner, error) {
1707         err := json.Unmarshal([]byte(record), &s.api.Container)
1708         c.Assert(err, IsNil)
1709         s.executor.runFunc = fn
1710         s.runner.RunArvMount = (&ArvMountCmdLine{}).ArvMountTest
1711         s.runner.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) {
1712                 return s.api, &KeepTestClient{}, nil, nil
1713         }
1714         return s.api, s.runner, s.runner.Run()
1715 }
1716
1717 func (s *TestSuite) TestStdoutWithWrongPath(c *C) {
1718         _, _, err := s.stdoutErrorRunHelper(c, `{
1719     "mounts": {"/tmp": {"kind": "tmp"}, "stdout": {"kind": "file", "path":"/tmpa.out"} },
1720     "output_path": "/tmp",
1721     "state": "Locked"
1722 }`, func() int { return 0 })
1723         c.Check(err, ErrorMatches, ".*Stdout path does not start with OutputPath.*")
1724 }
1725
1726 func (s *TestSuite) TestStdoutWithWrongKindTmp(c *C) {
1727         _, _, err := s.stdoutErrorRunHelper(c, `{
1728     "mounts": {"/tmp": {"kind": "tmp"}, "stdout": {"kind": "tmp", "path":"/tmp/a.out"} },
1729     "output_path": "/tmp",
1730     "state": "Locked"
1731 }`, func() int { return 0 })
1732         c.Check(err, ErrorMatches, ".*unsupported mount kind 'tmp' for stdout.*")
1733 }
1734
1735 func (s *TestSuite) TestStdoutWithWrongKindCollection(c *C) {
1736         _, _, err := s.stdoutErrorRunHelper(c, `{
1737     "mounts": {"/tmp": {"kind": "tmp"}, "stdout": {"kind": "collection", "path":"/tmp/a.out"} },
1738     "output_path": "/tmp",
1739     "state": "Locked"
1740 }`, func() int { return 0 })
1741         c.Check(err, ErrorMatches, ".*unsupported mount kind 'collection' for stdout.*")
1742 }
1743
1744 func (s *TestSuite) TestFullRunWithAPI(c *C) {
1745         s.fullRunHelper(c, `{
1746     "command": ["/bin/sh", "-c", "true $ARVADOS_API_HOST"],
1747     "container_image": "`+arvadostest.DockerImage112PDH+`",
1748     "cwd": "/bin",
1749     "environment": {},
1750     "mounts": {"/tmp": {"kind": "tmp"} },
1751     "output_path": "/tmp",
1752     "priority": 1,
1753     "runtime_constraints": {"API": true},
1754     "state": "Locked"
1755 }`, nil, func() int {
1756                 c.Check(s.executor.created.Env["ARVADOS_API_HOST"], Equals, os.Getenv("ARVADOS_API_HOST"))
1757                 return 3
1758         })
1759         c.Check(s.api.CalledWith("container.exit_code", 3), NotNil)
1760         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1761         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*`+reTimestamp+` Container exited with status code 3\n.*`)
1762 }
1763
1764 func (s *TestSuite) TestFullRunSetOutput(c *C) {
1765         defer os.Setenv("ARVADOS_API_HOST", os.Getenv("ARVADOS_API_HOST"))
1766         os.Setenv("ARVADOS_API_HOST", "test.arvados.org")
1767         s.fullRunHelper(c, `{
1768     "command": ["/bin/sh", "-c", "echo $ARVADOS_API_HOST"],
1769     "container_image": "`+arvadostest.DockerImage112PDH+`",
1770     "cwd": "/bin",
1771     "environment": {},
1772     "mounts": {"/tmp": {"kind": "tmp"} },
1773     "output_path": "/tmp",
1774     "priority": 1,
1775     "runtime_constraints": {"API": true},
1776     "state": "Locked"
1777 }`, nil, func() int {
1778                 s.api.Container.Output = arvadostest.DockerImage112PDH
1779                 return 0
1780         })
1781
1782         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1783         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1784         c.Check(s.api.CalledWith("container.output", arvadostest.DockerImage112PDH), NotNil)
1785 }
1786
1787 func (s *TestSuite) TestArvMountRuntimeStatusWarning(c *C) {
1788         s.runner.RunArvMount = func([]string, string) (*exec.Cmd, error) {
1789                 os.Mkdir(s.runner.ArvMountPoint+"/by_id", 0666)
1790                 ioutil.WriteFile(s.runner.ArvMountPoint+"/by_id/README", nil, 0666)
1791                 return s.runner.ArvMountCmd([]string{"bash", "-c", "echo >&2 Test: Keep write error: I am a teapot; sleep 3"}, "")
1792         }
1793         s.executor.runFunc = func() int {
1794                 time.Sleep(time.Second)
1795                 return 137
1796         }
1797         record := `{
1798     "command": ["sleep", "1"],
1799     "container_image": "` + arvadostest.DockerImage112PDH + `",
1800     "cwd": "/bin",
1801     "environment": {},
1802     "mounts": {"/tmp": {"kind": "tmp"} },
1803     "output_path": "/tmp",
1804     "priority": 1,
1805     "runtime_constraints": {"API": true},
1806     "state": "Locked"
1807 }`
1808         err := json.Unmarshal([]byte(record), &s.api.Container)
1809         c.Assert(err, IsNil)
1810         err = s.runner.Run()
1811         c.Assert(err, IsNil)
1812         c.Check(s.api.CalledWith("container.exit_code", 137), NotNil)
1813         c.Check(s.api.CalledWith("container.runtime_status.warning", "arv-mount: Keep write error"), NotNil)
1814         c.Check(s.api.CalledWith("container.runtime_status.warningDetail", "Test: Keep write error: I am a teapot"), NotNil)
1815         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1816         c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, `(?ms).*`+reTimestamp+` Container exited with status code 137 \(signal 9, SIGKILL\).*`)
1817         c.Check(logFileContent(c, s.runner, "arv-mount.txt"), Matches, reTimestamp+` Test: Keep write error: I am a teapot\n`)
1818 }
1819
1820 func (s *TestSuite) TestStdoutWithExcludeFromOutputMountPointUnderOutputDir(c *C) {
1821         helperRecord := `{
1822                 "command": ["/bin/sh", "-c", "echo $FROBIZ"],
1823                 "container_image": "` + arvadostest.DockerImage112PDH + `",
1824                 "cwd": "/bin",
1825                 "environment": {"FROBIZ": "bilbo"},
1826                 "mounts": {
1827         "/tmp": {"kind": "tmp"},
1828         "/tmp/foo": {"kind": "collection",
1829                      "portable_data_hash": "a3e8f74c6f101eae01fa08bfb4e49b3a+54",
1830                      "exclude_from_output": true
1831         },
1832         "stdout": {"kind": "file", "path": "/tmp/a/b/c.out"}
1833     },
1834                 "output_path": "/tmp",
1835                 "priority": 1,
1836                 "runtime_constraints": {},
1837                 "state": "Locked"
1838         }`
1839
1840         extraMounts := []string{"a3e8f74c6f101eae01fa08bfb4e49b3a+54"}
1841
1842         s.fullRunHelper(c, helperRecord, extraMounts, func() int {
1843                 fmt.Fprintln(s.executor.created.Stdout, s.executor.created.Env["FROBIZ"])
1844                 return 0
1845         })
1846
1847         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1848         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1849         c.Check(s.runner.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", "./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out\n"), NotNil)
1850 }
1851
1852 func (s *TestSuite) TestStdoutWithMultipleMountPointsUnderOutputDir(c *C) {
1853         helperRecord := `{
1854                 "command": ["/bin/sh", "-c", "echo $FROBIZ"],
1855                 "container_image": "` + arvadostest.DockerImage112PDH + `",
1856                 "cwd": "/bin",
1857                 "environment": {"FROBIZ": "bilbo"},
1858                 "mounts": {
1859         "/tmp": {"kind": "tmp"},
1860         "/tmp/foo/bar": {"kind": "collection", "portable_data_hash": "a0def87f80dd594d4675809e83bd4f15+367", "path":"/file2_in_main.txt"},
1861         "/tmp/foo/sub1": {"kind": "collection", "portable_data_hash": "a0def87f80dd594d4675809e83bd4f15+367", "path":"/subdir1"},
1862         "/tmp/foo/sub1file2": {"kind": "collection", "portable_data_hash": "a0def87f80dd594d4675809e83bd4f15+367", "path":"/subdir1/file2_in_subdir1.txt"},
1863         "/tmp/foo/baz/sub2file2": {"kind": "collection", "portable_data_hash": "a0def87f80dd594d4675809e83bd4f15+367", "path":"/subdir1/subdir2/file2_in_subdir2.txt"},
1864         "stdout": {"kind": "file", "path": "/tmp/a/b/c.out"}
1865     },
1866                 "output_path": "/tmp",
1867                 "priority": 1,
1868                 "runtime_constraints": {},
1869                 "state": "Locked",
1870                 "uuid": "zzzzz-dz642-202301130848001"
1871         }`
1872
1873         extraMounts := []string{
1874                 "a0def87f80dd594d4675809e83bd4f15+367/file2_in_main.txt",
1875                 "a0def87f80dd594d4675809e83bd4f15+367/subdir1/file2_in_subdir1.txt",
1876                 "a0def87f80dd594d4675809e83bd4f15+367/subdir1/subdir2/file2_in_subdir2.txt",
1877         }
1878
1879         api, _, realtemp := s.fullRunHelper(c, helperRecord, extraMounts, func() int {
1880                 fmt.Fprintln(s.executor.created.Stdout, s.executor.created.Env["FROBIZ"])
1881                 return 0
1882         })
1883
1884         c.Check(s.executor.created.BindMounts, DeepEquals, map[string]bindmount{
1885                 "/tmp":                   {realtemp + "/tmp1", false},
1886                 "/tmp/foo/bar":           {s.keepmount + "/by_id/a0def87f80dd594d4675809e83bd4f15+367/file2_in_main.txt", true},
1887                 "/tmp/foo/baz/sub2file2": {s.keepmount + "/by_id/a0def87f80dd594d4675809e83bd4f15+367/subdir1/subdir2/file2_in_subdir2.txt", true},
1888                 "/tmp/foo/sub1":          {s.keepmount + "/by_id/a0def87f80dd594d4675809e83bd4f15+367/subdir1", true},
1889                 "/tmp/foo/sub1file2":     {s.keepmount + "/by_id/a0def87f80dd594d4675809e83bd4f15+367/subdir1/file2_in_subdir1.txt", true},
1890         })
1891
1892         c.Check(api.CalledWith("container.exit_code", 0), NotNil)
1893         c.Check(api.CalledWith("container.state", "Complete"), NotNil)
1894         output_count := uint(0)
1895         for _, v := range s.runner.ContainerArvClient.(*ArvTestClient).Content {
1896                 if v["collection"] == nil {
1897                         continue
1898                 }
1899                 collection := v["collection"].(arvadosclient.Dict)
1900                 if collection["name"].(string) != "output for zzzzz-dz642-202301130848001" {
1901                         continue
1902                 }
1903                 c.Check(v["ensure_unique_name"], Equals, true)
1904                 c.Check(collection["manifest_text"].(string), Equals, `./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out
1905 ./foo 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396c73c0abcdefgh11234567890@569fa8c3 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396cabcdefghij6419876543234@569fa8c4 9:18:bar 36:18:sub1file2
1906 ./foo/baz 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396c73c0bcdefghijk544332211@569fa8c5 9:18:sub2file2
1907 ./foo/sub1 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396cabcdefghij6419876543234@569fa8c4 0:9:file1_in_subdir1.txt 9:18:file2_in_subdir1.txt
1908 ./foo/sub1/subdir2 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396c73c0bcdefghijk544332211@569fa8c5 0:9:file1_in_subdir2.txt 9:18:file2_in_subdir2.txt
1909 `)
1910                 output_count++
1911         }
1912         c.Check(output_count, Not(Equals), uint(0))
1913 }
1914
1915 func (s *TestSuite) TestStdoutWithMountPointsUnderOutputDirDenormalizedManifest(c *C) {
1916         helperRecord := `{
1917                 "command": ["/bin/sh", "-c", "echo $FROBIZ"],
1918                 "container_image": "` + arvadostest.DockerImage112PDH + `",
1919                 "cwd": "/bin",
1920                 "environment": {"FROBIZ": "bilbo"},
1921                 "mounts": {
1922         "/tmp": {"kind": "tmp"},
1923         "/tmp/foo/bar": {"kind": "collection", "portable_data_hash": "b0def87f80dd594d4675809e83bd4f15+367", "path": "/subdir1/file2_in_subdir1.txt"},
1924         "stdout": {"kind": "file", "path": "/tmp/a/b/c.out"}
1925     },
1926                 "output_path": "/tmp",
1927                 "priority": 1,
1928                 "runtime_constraints": {},
1929                 "state": "Locked",
1930                 "uuid": "zzzzz-dz642-202301130848002"
1931         }`
1932
1933         extraMounts := []string{
1934                 "b0def87f80dd594d4675809e83bd4f15+367/subdir1/file2_in_subdir1.txt",
1935         }
1936
1937         s.fullRunHelper(c, helperRecord, extraMounts, func() int {
1938                 fmt.Fprintln(s.executor.created.Stdout, s.executor.created.Env["FROBIZ"])
1939                 return 0
1940         })
1941
1942         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
1943         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
1944         output_count := uint(0)
1945         for _, v := range s.runner.ContainerArvClient.(*ArvTestClient).Content {
1946                 if v["collection"] == nil {
1947                         continue
1948                 }
1949                 collection := v["collection"].(arvadosclient.Dict)
1950                 if collection["name"].(string) != "output for zzzzz-dz642-202301130848002" {
1951                         continue
1952                 }
1953                 c.Check(collection["manifest_text"].(string), Equals, `./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out
1954 ./foo 3e426d509afffb85e06c4c96a7c15e91+27+Aa124ac75e5168396c73c0abcdefgh11234567890@569fa8c3 10:17:bar
1955 `)
1956                 output_count++
1957         }
1958         c.Check(output_count, Not(Equals), uint(0))
1959 }
1960
1961 func (s *TestSuite) TestOutputError(c *C) {
1962         helperRecord := `{
1963                 "command": ["/bin/sh", "-c", "echo $FROBIZ"],
1964                 "container_image": "` + arvadostest.DockerImage112PDH + `",
1965                 "cwd": "/bin",
1966                 "environment": {"FROBIZ": "bilbo"},
1967                 "mounts": {
1968                         "/tmp": {"kind": "tmp"}
1969                 },
1970                 "output_path": "/tmp",
1971                 "priority": 1,
1972                 "runtime_constraints": {},
1973                 "state": "Locked"
1974         }`
1975         s.fullRunHelper(c, helperRecord, nil, func() int {
1976                 os.Symlink("/etc/hosts", s.runner.HostOutputDir+"/baz")
1977                 return 0
1978         })
1979
1980         c.Check(s.api.CalledWith("container.state", "Cancelled"), NotNil)
1981 }
1982
1983 func (s *TestSuite) TestStdinCollectionMountPoint(c *C) {
1984         helperRecord := `{
1985                 "command": ["/bin/sh", "-c", "echo $FROBIZ"],
1986                 "container_image": "` + arvadostest.DockerImage112PDH + `",
1987                 "cwd": "/bin",
1988                 "environment": {"FROBIZ": "bilbo"},
1989                 "mounts": {
1990         "/tmp": {"kind": "tmp"},
1991         "stdin": {"kind": "collection", "portable_data_hash": "b0def87f80dd594d4675809e83bd4f15+367", "path": "/file1_in_main.txt"},
1992         "stdout": {"kind": "file", "path": "/tmp/a/b/c.out"}
1993     },
1994                 "output_path": "/tmp",
1995                 "priority": 1,
1996                 "runtime_constraints": {},
1997                 "state": "Locked"
1998         }`
1999
2000         extraMounts := []string{
2001                 "b0def87f80dd594d4675809e83bd4f15+367/file1_in_main.txt",
2002         }
2003
2004         api, _, _ := s.fullRunHelper(c, helperRecord, extraMounts, func() int {
2005                 fmt.Fprintln(s.executor.created.Stdout, s.executor.created.Env["FROBIZ"])
2006                 return 0
2007         })
2008
2009         c.Check(api.CalledWith("container.exit_code", 0), NotNil)
2010         c.Check(api.CalledWith("container.state", "Complete"), NotNil)
2011         for _, v := range api.Content {
2012                 if v["collection"] != nil {
2013                         collection := v["collection"].(arvadosclient.Dict)
2014                         if strings.Index(collection["name"].(string), "output") == 0 {
2015                                 manifest := collection["manifest_text"].(string)
2016                                 c.Check(manifest, Equals, `./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out
2017 `)
2018                         }
2019                 }
2020         }
2021 }
2022
2023 func (s *TestSuite) TestStdinJsonMountPoint(c *C) {
2024         helperRecord := `{
2025                 "command": ["/bin/sh", "-c", "echo $FROBIZ"],
2026                 "container_image": "` + arvadostest.DockerImage112PDH + `",
2027                 "cwd": "/bin",
2028                 "environment": {"FROBIZ": "bilbo"},
2029                 "mounts": {
2030         "/tmp": {"kind": "tmp"},
2031         "stdin": {"kind": "json", "content": "foo"},
2032         "stdout": {"kind": "file", "path": "/tmp/a/b/c.out"}
2033     },
2034                 "output_path": "/tmp",
2035                 "priority": 1,
2036                 "runtime_constraints": {},
2037                 "state": "Locked"
2038         }`
2039
2040         api, _, _ := s.fullRunHelper(c, helperRecord, nil, func() int {
2041                 fmt.Fprintln(s.executor.created.Stdout, s.executor.created.Env["FROBIZ"])
2042                 return 0
2043         })
2044
2045         c.Check(api.CalledWith("container.exit_code", 0), NotNil)
2046         c.Check(api.CalledWith("container.state", "Complete"), NotNil)
2047         for _, v := range api.Content {
2048                 if v["collection"] != nil {
2049                         collection := v["collection"].(arvadosclient.Dict)
2050                         if strings.Index(collection["name"].(string), "output") == 0 {
2051                                 manifest := collection["manifest_text"].(string)
2052                                 c.Check(manifest, Equals, `./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out
2053 `)
2054                         }
2055                 }
2056         }
2057 }
2058
2059 func (s *TestSuite) TestStderrMount(c *C) {
2060         api, cr, _ := s.fullRunHelper(c, `{
2061     "command": ["/bin/sh", "-c", "echo hello;exit 1"],
2062     "container_image": "`+arvadostest.DockerImage112PDH+`",
2063     "cwd": ".",
2064     "environment": {},
2065     "mounts": {"/tmp": {"kind": "tmp"},
2066                "stdout": {"kind": "file", "path": "/tmp/a/out.txt"},
2067                "stderr": {"kind": "file", "path": "/tmp/b/err.txt"}},
2068     "output_path": "/tmp",
2069     "priority": 1,
2070     "runtime_constraints": {},
2071     "state": "Locked"
2072 }`, nil, func() int {
2073                 fmt.Fprintln(s.executor.created.Stdout, "hello")
2074                 fmt.Fprintln(s.executor.created.Stderr, "oops")
2075                 return 1
2076         })
2077
2078         final := api.CalledWith("container.state", "Complete")
2079         c.Assert(final, NotNil)
2080         c.Check(final["container"].(arvadosclient.Dict)["exit_code"], Equals, 1)
2081         c.Check(final["container"].(arvadosclient.Dict)["log"], NotNil)
2082
2083         c.Check(cr.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", "./a b1946ac92492d2347c6235b4d2611184+6 0:6:out.txt\n./b 38af5c54926b620264ab1501150cf189+5 0:5:err.txt\n"), NotNil)
2084 }
2085
2086 func (s *TestSuite) TestNumberRoundTrip(c *C) {
2087         s.api.callraw = true
2088         err := s.runner.fetchContainerRecord()
2089         c.Assert(err, IsNil)
2090         jsondata, err := json.Marshal(s.runner.Container.Mounts["/json"].Content)
2091         c.Logf("%#v", s.runner.Container)
2092         c.Check(err, IsNil)
2093         c.Check(string(jsondata), Equals, `{"number":123456789123456789}`)
2094 }
2095
2096 func (s *TestSuite) TestFullBrokenDocker(c *C) {
2097         nextState := ""
2098         for _, setup := range []func(){
2099                 func() {
2100                         c.Log("// waitErr = ocl runtime error")
2101                         s.executor.waitErr = errors.New(`Error response from daemon: oci runtime error: container_linux.go:247: starting container process caused "process_linux.go:359: container init caused \"rootfs_linux.go:54: mounting \\\"/tmp/keep453790790/by_id/99999999999999999999999999999999+99999/myGenome\\\" to rootfs \\\"/tmp/docker/overlay2/9999999999999999999999999999999999999999999999999999999999999999/merged\\\" at \\\"/tmp/docker/overlay2/9999999999999999999999999999999999999999999999999999999999999999/merged/keep/99999999999999999999999999999999+99999/myGenome\\\" caused \\\"no such file or directory\\\"\""`)
2102                         nextState = "Cancelled"
2103                 },
2104                 func() {
2105                         c.Log("// loadErr = cannot connect")
2106                         s.executor.loadErr = errors.New("Cannot connect to the Docker daemon at unix:///var/run/docker.sock. Is the docker daemon running?")
2107                         s.runner.brokenNodeHook = c.MkDir() + "/broken-node-hook"
2108                         err := ioutil.WriteFile(s.runner.brokenNodeHook, []byte("#!/bin/sh\nexec echo killme\n"), 0700)
2109                         c.Assert(err, IsNil)
2110                         nextState = "Queued"
2111                 },
2112         } {
2113                 s.SetUpTest(c)
2114                 setup()
2115                 s.fullRunHelper(c, `{
2116     "command": ["echo", "hello world"],
2117     "container_image": "`+arvadostest.DockerImage112PDH+`",
2118     "cwd": ".",
2119     "environment": {},
2120     "mounts": {"/tmp": {"kind": "tmp"} },
2121     "output_path": "/tmp",
2122     "priority": 1,
2123     "runtime_constraints": {},
2124     "state": "Locked"
2125 }`, nil, func() int { return 0 })
2126                 c.Check(s.api.CalledWith("container.state", nextState), NotNil)
2127                 logs := logFileContent(c, s.runner, "crunch-run.txt")
2128                 c.Check(logs, Matches, "(?ms).*unable to run containers.*")
2129                 if s.runner.brokenNodeHook != "" {
2130                         c.Check(logs, Matches, "(?ms).*Running broken node hook.*")
2131                         c.Check(logs, Matches, "(?ms).*killme.*")
2132                         c.Check(logs, Not(Matches), "(?ms).*Writing /var/lock/crunch-run-broken to mark node as broken.*")
2133                 } else {
2134                         c.Check(logs, Matches, "(?ms).*Writing /var/lock/crunch-run-broken to mark node as broken.*")
2135                 }
2136         }
2137 }
2138
2139 func (s *TestSuite) TestBadCommand(c *C) {
2140         for _, startError := range []string{
2141                 `panic: standard_init_linux.go:175: exec user process caused "no such file or directory"`,
2142                 `Error response from daemon: Cannot start container 41f26cbc43bcc1280f4323efb1830a394ba8660c9d1c2b564ba42bf7f7694845: [8] System error: no such file or directory`,
2143                 `Error response from daemon: Cannot start container 58099cd76c834f3dc2a4fb76c8028f049ae6d4fdf0ec373e1f2cfea030670c2d: [8] System error: exec: "foobar": executable file not found in $PATH`,
2144         } {
2145                 s.SetUpTest(c)
2146                 s.executor.startErr = errors.New(startError)
2147                 s.fullRunHelper(c, `{
2148     "command": ["echo", "hello world"],
2149     "container_image": "`+arvadostest.DockerImage112PDH+`",
2150     "cwd": ".",
2151     "environment": {},
2152     "mounts": {"/tmp": {"kind": "tmp"} },
2153     "output_path": "/tmp",
2154     "priority": 1,
2155     "runtime_constraints": {},
2156     "state": "Locked"
2157 }`, nil, func() int { return 0 })
2158                 c.Check(s.api.CalledWith("container.state", "Cancelled"), NotNil)
2159                 c.Check(logFileContent(c, s.runner, "crunch-run.txt"), Matches, "(?ms).*Possible causes:.*is missing.*")
2160         }
2161 }
2162
2163 func (s *TestSuite) TestSecretTextMountPoint(c *C) {
2164         helperRecord := `{
2165                 "command": ["true"],
2166                 "container_image": "` + arvadostest.DockerImage112PDH + `",
2167                 "cwd": "/bin",
2168                 "mounts": {
2169                     "/tmp": {"kind": "tmp"},
2170                     "/tmp/secret.conf": {"kind": "text", "content": "mypassword"}
2171                 },
2172                 "secret_mounts": {
2173                 },
2174                 "output_path": "/tmp",
2175                 "priority": 1,
2176                 "runtime_constraints": {},
2177                 "state": "Locked"
2178         }`
2179
2180         s.fullRunHelper(c, helperRecord, nil, func() int {
2181                 content, err := ioutil.ReadFile(s.runner.HostOutputDir + "/secret.conf")
2182                 c.Check(err, IsNil)
2183                 c.Check(string(content), Equals, "mypassword")
2184                 return 0
2185         })
2186
2187         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
2188         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
2189         c.Check(s.runner.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ". 34819d7beeabb9260a5c854bc85b3e44+10 0:10:secret.conf\n"), NotNil)
2190         c.Check(s.runner.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ""), IsNil)
2191
2192         // under secret mounts, not captured in output
2193         helperRecord = `{
2194                 "command": ["true"],
2195                 "container_image": "` + arvadostest.DockerImage112PDH + `",
2196                 "cwd": "/bin",
2197                 "mounts": {
2198                     "/tmp": {"kind": "tmp"}
2199                 },
2200                 "secret_mounts": {
2201                     "/tmp/secret.conf": {"kind": "text", "content": "mypassword"}
2202                 },
2203                 "output_path": "/tmp",
2204                 "priority": 1,
2205                 "runtime_constraints": {},
2206                 "state": "Locked"
2207         }`
2208
2209         s.SetUpTest(c)
2210         s.fullRunHelper(c, helperRecord, nil, func() int {
2211                 content, err := ioutil.ReadFile(s.runner.HostOutputDir + "/secret.conf")
2212                 c.Check(err, IsNil)
2213                 c.Check(string(content), Equals, "mypassword")
2214                 return 0
2215         })
2216
2217         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
2218         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
2219         c.Check(s.runner.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ". 34819d7beeabb9260a5c854bc85b3e44+10 0:10:secret.conf\n"), IsNil)
2220         c.Check(s.runner.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ""), NotNil)
2221
2222         // under secret mounts, output dir is a collection, not captured in output
2223         helperRecord = `{
2224                 "command": ["true"],
2225                 "container_image": "` + arvadostest.DockerImage112PDH + `",
2226                 "cwd": "/bin",
2227                 "mounts": {
2228                     "/tmp": {"kind": "collection", "writable": true}
2229                 },
2230                 "secret_mounts": {
2231                     "/tmp/secret.conf": {"kind": "text", "content": "mypassword"}
2232                 },
2233                 "output_path": "/tmp",
2234                 "priority": 1,
2235                 "runtime_constraints": {},
2236                 "state": "Locked"
2237         }`
2238
2239         s.SetUpTest(c)
2240         _, _, realtemp := s.fullRunHelper(c, helperRecord, nil, func() int {
2241                 // secret.conf should be provisioned as a separate
2242                 // bind mount, i.e., it should not appear in the
2243                 // (fake) fuse filesystem as viewed from the host.
2244                 content, err := ioutil.ReadFile(s.runner.HostOutputDir + "/secret.conf")
2245                 if !c.Check(errors.Is(err, os.ErrNotExist), Equals, true) {
2246                         c.Logf("secret.conf: content %q, err %#v", content, err)
2247                 }
2248                 err = ioutil.WriteFile(s.runner.HostOutputDir+"/.arvados#collection", []byte(`{"manifest_text":". acbd18db4cc2f85cedef654fccc4a4d8+3 0:3:foo.txt\n"}`), 0700)
2249                 c.Check(err, IsNil)
2250                 return 0
2251         })
2252
2253         content, err := ioutil.ReadFile(realtemp + "/text1/mountdata.text")
2254         c.Check(err, IsNil)
2255         c.Check(string(content), Equals, "mypassword")
2256         c.Check(s.executor.created.BindMounts["/tmp/secret.conf"], DeepEquals, bindmount{realtemp + "/text1/mountdata.text", true})
2257         c.Check(s.api.CalledWith("container.exit_code", 0), NotNil)
2258         c.Check(s.api.CalledWith("container.state", "Complete"), NotNil)
2259         c.Check(s.runner.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ". acbd18db4cc2f85cedef654fccc4a4d8+3 0:3:foo.txt\n"), NotNil)
2260 }
2261
2262 func (s *TestSuite) TestCalculateCost(c *C) {
2263         defer func(s string) { lockdir = s }(lockdir)
2264         lockdir = c.MkDir()
2265         now := time.Now()
2266         cr := s.runner
2267         cr.costStartTime = now.Add(-time.Hour)
2268         var logbuf bytes.Buffer
2269         cr.CrunchLog = newLogWriter(&logbuf)
2270
2271         // if there's no InstanceType env var, cost is calculated as 0
2272         os.Unsetenv("InstanceType")
2273         cost := cr.calculateCost(now)
2274         c.Check(cost, Equals, 0.0)
2275
2276         // with InstanceType env var and loadPrices() hasn't run (or
2277         // hasn't found any data), cost is calculated based on
2278         // InstanceType env var
2279         os.Setenv("InstanceType", `{"Price":1.2}`)
2280         cost = cr.calculateCost(now)
2281         c.Check(cost, Equals, 1.2)
2282
2283         // first update tells us the spot price was $1/h until 30
2284         // minutes ago when it increased to $2/h
2285         j, err := json.Marshal([]cloud.InstancePrice{
2286                 {StartTime: now.Add(-4 * time.Hour), Price: 1.0},
2287                 {StartTime: now.Add(-time.Hour / 2), Price: 2.0},
2288         })
2289         c.Assert(err, IsNil)
2290         os.WriteFile(lockdir+"/"+pricesfile, j, 0777)
2291         cr.loadPrices()
2292         cost = cr.calculateCost(now)
2293         c.Check(cost, Equals, 1.5)
2294
2295         // next update (via --list + SIGUSR2) tells us the spot price
2296         // increased to $3/h 15 minutes ago
2297         j, err = json.Marshal([]cloud.InstancePrice{
2298                 {StartTime: now.Add(-time.Hour / 3), Price: 2.0}, // dup of -time.Hour/2 price
2299                 {StartTime: now.Add(-time.Hour / 4), Price: 3.0},
2300         })
2301         c.Assert(err, IsNil)
2302         os.WriteFile(lockdir+"/"+pricesfile, j, 0777)
2303         cr.loadPrices()
2304         cost = cr.calculateCost(now)
2305         c.Check(cost, Equals, 1.0/2+2.0/4+3.0/4)
2306
2307         cost = cr.calculateCost(now.Add(-time.Hour / 2))
2308         c.Check(cost, Equals, 0.5)
2309
2310         c.Logf("%s", logbuf.String())
2311         c.Check(logbuf.String(), Matches, `(?ms).*Instance price changed to 1\.00 at 20.* changed to 2\.00 .* changed to 3\.00 .*`)
2312         c.Check(logbuf.String(), Not(Matches), `(?ms).*changed to 2\.00 .* changed to 2\.00 .*`)
2313 }
2314
2315 func (s *TestSuite) TestSIGUSR2CostUpdate(c *C) {
2316         pid := os.Getpid()
2317         now := time.Now()
2318         pricesJSON, err := json.Marshal([]cloud.InstancePrice{
2319                 {StartTime: now.Add(-4 * time.Hour), Price: 2.4},
2320                 {StartTime: now.Add(-2 * time.Hour), Price: 2.6},
2321         })
2322         c.Assert(err, IsNil)
2323
2324         os.Setenv("InstanceType", `{"Price":2.2}`)
2325         defer func(s string) { lockdir = s }(lockdir)
2326         lockdir = c.MkDir()
2327
2328         // We can't use s.api.CalledWith because timing differences will yield
2329         // different cost values across runs. getCostUpdate iterates over API
2330         // calls until it finds one that sets the cost, then writes that value
2331         // to the next index of costUpdates.
2332         deadline := now.Add(time.Second)
2333         costUpdates := make([]float64, 2)
2334         costIndex := 0
2335         apiIndex := 0
2336         getCostUpdate := func() {
2337                 for ; time.Now().Before(deadline); time.Sleep(time.Second / 10) {
2338                         for apiIndex < len(s.api.Content) {
2339                                 update := s.api.Content[apiIndex]
2340                                 apiIndex++
2341                                 var ok bool
2342                                 var cost float64
2343                                 if update, ok = update["container"].(arvadosclient.Dict); !ok {
2344                                         continue
2345                                 }
2346                                 if cost, ok = update["cost"].(float64); !ok {
2347                                         continue
2348                                 }
2349                                 c.Logf("API call #%d updates cost to %v", apiIndex-1, cost)
2350                                 costUpdates[costIndex] = cost
2351                                 costIndex++
2352                                 return
2353                         }
2354                 }
2355         }
2356
2357         s.fullRunHelper(c, `{
2358                 "command": ["true"],
2359                 "container_image": "`+arvadostest.DockerImage112PDH+`",
2360                 "cwd": ".",
2361                 "environment": {},
2362                 "mounts": {"/tmp": {"kind": "tmp"} },
2363                 "output_path": "/tmp",
2364                 "priority": 1,
2365                 "runtime_constraints": {},
2366                 "state": "Locked",
2367                 "uuid": "zzzzz-dz642-20230320101530a"
2368         }`, nil, func() int {
2369                 s.runner.costStartTime = now.Add(-3 * time.Hour)
2370                 err := syscall.Kill(pid, syscall.SIGUSR2)
2371                 c.Check(err, IsNil, Commentf("error sending first SIGUSR2 to runner"))
2372                 getCostUpdate()
2373
2374                 err = os.WriteFile(path.Join(lockdir, pricesfile), pricesJSON, 0o700)
2375                 c.Check(err, IsNil, Commentf("error writing JSON prices file"))
2376                 err = syscall.Kill(pid, syscall.SIGUSR2)
2377                 c.Check(err, IsNil, Commentf("error sending second SIGUSR2 to runner"))
2378                 getCostUpdate()
2379
2380                 return 0
2381         })
2382         // Comparing with format strings makes it easy to ignore minor variations
2383         // in cost across runs while keeping diagnostics pretty.
2384         c.Check(fmt.Sprintf("%.3f", costUpdates[0]), Equals, "6.600")
2385         c.Check(fmt.Sprintf("%.3f", costUpdates[1]), Equals, "7.600")
2386 }
2387
2388 type FakeProcess struct {
2389         cmdLine []string
2390 }
2391
2392 func (fp FakeProcess) CmdlineSlice() ([]string, error) {
2393         return fp.cmdLine, nil
2394 }
2395
2396 func logFileContent(c *C, cr *ContainerRunner, fnm string) string {
2397         buf, err := fs.ReadFile(arvados.FS(cr.LogCollection), fnm)
2398         c.Assert(err, IsNil)
2399         return string(buf)
2400 }
2401
2402 func dumpAllLogFiles(c *C, cr *ContainerRunner) {
2403         d, err := cr.LogCollection.OpenFile("/", os.O_RDONLY, 0)
2404         c.Assert(err, IsNil)
2405         fis, err := d.Readdir(-1)
2406         c.Assert(err, IsNil)
2407         for _, fi := range fis {
2408                 c.Logf("=== %s", fi.Name())
2409                 c.Log(logFileContent(c, cr, fi.Name()))
2410         }
2411 }