X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/4ed6626cfba06e924f77dc54e1fe704eff1d113c..90944740f40ab0dbfc4bdfc1b16accfbf6559e4f:/services/crunch-run/crunchrun.go diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index 800556866a..84b578a3e2 100644 --- a/services/crunch-run/crunchrun.go +++ b/services/crunch-run/crunchrun.go @@ -32,7 +32,6 @@ import ( "git.curoverse.com/arvados.git/sdk/go/arvadosclient" "git.curoverse.com/arvados.git/sdk/go/keepclient" "git.curoverse.com/arvados.git/sdk/go/manifest" - "github.com/shirou/gopsutil/process" "golang.org/x/net/context" dockertypes "github.com/docker/docker/api/types" @@ -93,10 +92,30 @@ type PsProcess interface { // ContainerRunner is the main stateful struct used for a single execution of a // container. type ContainerRunner struct { - Docker ThinDockerClient - client *arvados.Client - ArvClient IArvadosClient - Kc IKeepClient + Docker ThinDockerClient + + // Dispatcher client is initialized with the Dispatcher token. + // This is a priviledged token used to manage container status + // and logs. + // + // We have both dispatcherClient and DispatcherArvClient + // because there are two different incompatible Arvados Go + // SDKs and we have to use both (hopefully this gets fixed in + // #14467) + dispatcherClient *arvados.Client + DispatcherArvClient IArvadosClient + DispatcherKeepClient IKeepClient + + // Container client is initialized with the Container token + // This token controls the permissions of the container, and + // must be used for operations such as reading collections. + // + // Same comment as above applies to + // containerClient/ContainerArvClient. + containerClient *arvados.Client + ContainerArvClient IArvadosClient + ContainerKeepClient IKeepClient + Container arvados.Container ContainerConfig dockercontainer.Config HostConfig dockercontainer.HostConfig @@ -123,12 +142,10 @@ type ContainerRunner struct { SigChan chan os.Signal ArvMountExit chan error SecretMounts map[string]arvados.Mount - MkArvClient func(token string) (IArvadosClient, error) + MkArvClient func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) finalState string parentTemp string - ListProcesses func() ([]PsProcess, error) - statLogger io.WriteCloser statReporter *crunchstat.Reporter hoststatLogger io.WriteCloser @@ -153,10 +170,9 @@ type ContainerRunner struct { cCancelled bool // StopContainer() invoked cRemoved bool // docker confirmed the container no longer exists - enableNetwork string // one of "default" or "always" - networkMode string // passed through to HostConfig.NetworkMode - arvMountLog *ThrottledLogger - checkContainerd time.Duration + enableNetwork string // one of "default" or "always" + networkMode string // passed through to HostConfig.NetworkMode + arvMountLog *ThrottledLogger containerWatchdogInterval time.Duration } @@ -206,7 +222,14 @@ var brokenNodeHook *string = flag.String("broken-node-hook", "", "Script to run func (runner *ContainerRunner) runBrokenNodeHook() { if *brokenNodeHook == "" { - runner.CrunchLog.Printf("No broken node hook provided, cannot mark node as broken.") + path := filepath.Join(lockdir, brokenfile) + runner.CrunchLog.Printf("Writing %s to mark node as broken", path) + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0700) + if err != nil { + runner.CrunchLog.Printf("Error writing %s: %s", path, err) + return + } + f.Close() } else { runner.CrunchLog.Printf("Running broken node hook %q", *brokenNodeHook) // run killme script @@ -239,7 +262,7 @@ func (runner *ContainerRunner) LoadImage() (err error) { runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.Container.ContainerImage) var collection arvados.Collection - err = runner.ArvClient.Get("collections", runner.Container.ContainerImage, nil, &collection) + err = runner.ContainerArvClient.Get("collections", runner.Container.ContainerImage, nil, &collection) if err != nil { return fmt.Errorf("While getting container image collection: %v", err) } @@ -260,7 +283,7 @@ func (runner *ContainerRunner) LoadImage() (err error) { runner.CrunchLog.Print("Loading Docker image from keep") var readCloser io.ReadCloser - readCloser, err = runner.Kc.ManifestFileReader(manifest, img) + readCloser, err = runner.ContainerKeepClient.ManifestFileReader(manifest, img) if err != nil { return fmt.Errorf("While creating ManifestFileReader for container image: %v", err) } @@ -282,7 +305,7 @@ func (runner *ContainerRunner) LoadImage() (err error) { runner.ContainerConfig.Image = imageID - runner.Kc.ClearBlockCache() + runner.ContainerKeepClient.ClearBlockCache() return nil } @@ -583,7 +606,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { if err != nil { return fmt.Errorf("creating temp dir: %v", err) } - err = gitMount(mnt).extractTree(runner.ArvClient, tmpdir, token) + err = gitMount(mnt).extractTree(runner.ContainerArvClient, tmpdir, token) if err != nil { return err } @@ -850,13 +873,13 @@ func (runner *ContainerRunner) logAPIResponse(label, path string, params map[str return false, err } w := &ArvLogWriter{ - ArvClient: runner.ArvClient, + ArvClient: runner.DispatcherArvClient, UUID: runner.Container.UUID, loggingStream: label, writeCloser: writer, } - reader, err := runner.ArvClient.CallRaw("GET", path, "", "", arvadosclient.Dict(params)) + reader, err := runner.DispatcherArvClient.CallRaw("GET", path, "", "", arvadosclient.Dict(params)) if err != nil { return false, fmt.Errorf("error getting %s record: %v", label, err) } @@ -906,12 +929,14 @@ func (runner *ContainerRunner) AttachStreams() (err error) { if collId == "" { collId = stdinMnt.PortableDataHash } - err = runner.ArvClient.Get("collections", collId, nil, &stdinColl) + err = runner.ContainerArvClient.Get("collections", collId, nil, &stdinColl) if err != nil { - return fmt.Errorf("While getting stding collection: %v", err) + return fmt.Errorf("While getting stdin collection: %v", err) } - stdinRdr, err = runner.Kc.ManifestFileReader(manifest.Manifest{Text: stdinColl.ManifestText}, stdinMnt.Path) + stdinRdr, err = runner.ContainerKeepClient.ManifestFileReader( + manifest.Manifest{Text: stdinColl.ManifestText}, + stdinMnt.Path) if os.IsNotExist(err) { return fmt.Errorf("stdin collection path not found: %v", stdinMnt.Path) } else if err != nil { @@ -962,7 +987,7 @@ func (runner *ContainerRunner) AttachStreams() (err error) { go func() { _, err := io.Copy(response.Conn, stdinRdr) if err != nil { - runner.CrunchLog.Print("While writing stdin collection to docker container %q", err) + runner.CrunchLog.Printf("While writing stdin collection to docker container %q", err) runner.stop(nil) } stdinRdr.Close() @@ -972,7 +997,7 @@ func (runner *ContainerRunner) AttachStreams() (err error) { go func() { _, err := io.Copy(response.Conn, bytes.NewReader(stdinJson)) if err != nil { - runner.CrunchLog.Print("While writing stdin json to docker container %q", err) + runner.CrunchLog.Printf("While writing stdin json to docker container %q", err) runner.stop(nil) } response.CloseWrite() @@ -1099,27 +1124,6 @@ func (runner *ContainerRunner) StartContainer() error { return nil } -// checkContainerd checks if "containerd" is present in the process list. -func (runner *ContainerRunner) CheckContainerd() error { - if runner.checkContainerd == 0 { - return nil - } - p, _ := runner.ListProcesses() - for _, i := range p { - e, _ := i.CmdlineSlice() - if len(e) > 0 { - if strings.Index(e[0], "containerd") > -1 { - return nil - } - } - } - - // Not found - runner.runBrokenNodeHook() - runner.stop(nil) - return fmt.Errorf("'containerd' not found in process list.") -} - // WaitFinish waits for the container to terminate, capture the exit code, and // close the stdout/stderr logging. func (runner *ContainerRunner) WaitFinish() error { @@ -1158,27 +1162,6 @@ func (runner *ContainerRunner) WaitFinish() error { } }() - containerdGone := make(chan error) - defer close(containerdGone) - if runner.checkContainerd > 0 { - go func() { - ticker := time.NewTicker(time.Duration(runner.checkContainerd)) - defer ticker.Stop() - for { - select { - case <-ticker.C: - if ck := runner.CheckContainerd(); ck != nil { - containerdGone <- ck - return - } - case <-containerdGone: - // Channel closed, quit goroutine - return - } - } - }() - } - for { select { case waitBody := <-waitOk: @@ -1207,9 +1190,6 @@ func (runner *ContainerRunner) WaitFinish() error { case <-containerGone: return errors.New("docker client never returned status") - - case err := <-containerdGone: - return err } } } @@ -1250,7 +1230,7 @@ func (runner *ContainerRunner) updateLogs() { } var updated arvados.Container - err = runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{ + err = runner.DispatcherArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{ "container": arvadosclient.Dict{"log": saved.PortableDataHash}, }, &updated) if err != nil { @@ -1268,7 +1248,7 @@ func (runner *ContainerRunner) CaptureOutput() error { if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI { // Output may have been set directly by the container, so // refresh the container record to check. - err := runner.ArvClient.Get("containers", runner.Container.UUID, + err := runner.DispatcherArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container) if err != nil { return err @@ -1281,9 +1261,9 @@ func (runner *ContainerRunner) CaptureOutput() error { } txt, err := (&copier{ - client: runner.client, - arvClient: runner.ArvClient, - keepClient: runner.Kc, + client: runner.containerClient, + arvClient: runner.ContainerArvClient, + keepClient: runner.ContainerKeepClient, hostOutputDir: runner.HostOutputDir, ctrOutputDir: runner.Container.OutputPath, binds: runner.Binds, @@ -1296,7 +1276,7 @@ func (runner *ContainerRunner) CaptureOutput() error { } if n := len(regexp.MustCompile(` [0-9a-f]+\+\S*\+R`).FindAllStringIndex(txt, -1)); n > 0 { runner.CrunchLog.Printf("Copying %d data blocks from remote input collections...", n) - fs, err := (&arvados.Collection{ManifestText: txt}).FileSystem(runner.client, runner.Kc) + fs, err := (&arvados.Collection{ManifestText: txt}).FileSystem(runner.containerClient, runner.ContainerKeepClient) if err != nil { return err } @@ -1306,7 +1286,7 @@ func (runner *ContainerRunner) CaptureOutput() error { } } var resp arvados.Collection - err = runner.ArvClient.Create("collections", arvadosclient.Dict{ + err = runner.ContainerArvClient.Create("collections", arvadosclient.Dict{ "ensure_unique_name": true, "collection": arvadosclient.Dict{ "is_trashed": true, @@ -1394,7 +1374,7 @@ func (runner *ContainerRunner) CommitLogs() error { // other further errors (such as failing to write the log to Keep!) // while shutting down runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{ - ArvClient: runner.ArvClient, + ArvClient: runner.DispatcherArvClient, UUID: runner.Container.UUID, loggingStream: "crunch-run", writeCloser: nil, @@ -1446,9 +1426,9 @@ func (runner *ContainerRunner) saveLogCollection(final bool) (response arvados.C reqBody := arvadosclient.Dict{"collection": updates} if runner.logUUID == "" { reqBody["ensure_unique_name"] = true - err = runner.ArvClient.Create("collections", reqBody, &response) + err = runner.DispatcherArvClient.Create("collections", reqBody, &response) } else { - err = runner.ArvClient.Update("collections", runner.logUUID, reqBody, &response) + err = runner.DispatcherArvClient.Update("collections", runner.logUUID, reqBody, &response) } if err != nil { return @@ -1464,7 +1444,7 @@ func (runner *ContainerRunner) UpdateContainerRunning() error { if runner.cCancelled { return ErrCancelled } - return runner.ArvClient.Update("containers", runner.Container.UUID, + return runner.DispatcherArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil) } @@ -1476,7 +1456,7 @@ func (runner *ContainerRunner) ContainerToken() (string, error) { } var auth arvados.APIClientAuthorization - err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth) + err := runner.DispatcherArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth) if err != nil { return "", err } @@ -1500,7 +1480,7 @@ func (runner *ContainerRunner) UpdateContainerFinal() error { update["output"] = *runner.OutputPDH } } - return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil) + return runner.DispatcherArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil) } // IsCancelled returns the value of Cancelled, with goroutine safety. @@ -1517,7 +1497,7 @@ func (runner *ContainerRunner) NewArvLogWriter(name string) (io.WriteCloser, err return nil, err } return &ArvLogWriter{ - ArvClient: runner.ArvClient, + ArvClient: runner.DispatcherArvClient, UUID: runner.Container.UUID, loggingStream: name, writeCloser: writer, @@ -1545,6 +1525,14 @@ func (runner *ContainerRunner) Run() (err error) { runner.CrunchLog.Close() }() + err = runner.fetchContainerRecord() + if err != nil { + return + } + if runner.Container.State != "Locked" { + return fmt.Errorf("dispatch error detected: container %q has state %q", runner.Container.UUID, runner.Container.State) + } + defer func() { // checkErr prints e (unless it's nil) and sets err to // e (unless err is already non-nil). Thus, if err @@ -1585,22 +1573,12 @@ func (runner *ContainerRunner) Run() (err error) { checkErr("UpdateContainerFinal", runner.UpdateContainerFinal()) }() - err = runner.fetchContainerRecord() - if err != nil { - return - } runner.setupSignals() err = runner.startHoststat() if err != nil { return } - // Sanity check that containerd is running. - err = runner.CheckContainerd() - if err != nil { - return - } - // check for and/or load image err = runner.LoadImage() if err != nil { @@ -1669,7 +1647,7 @@ func (runner *ContainerRunner) Run() (err error) { // Fetch the current container record (uuid = runner.Container.UUID) // into runner.Container. func (runner *ContainerRunner) fetchContainerRecord() error { - reader, err := runner.ArvClient.CallRaw("GET", "containers", runner.Container.UUID, "", nil) + reader, err := runner.DispatcherArvClient.CallRaw("GET", "containers", runner.Container.UUID, "", nil) if err != nil { return fmt.Errorf("error fetching container record: %v", err) } @@ -1691,12 +1669,13 @@ func (runner *ContainerRunner) fetchContainerRecord() error { return fmt.Errorf("error getting container token: %v", err) } - containerClient, err := runner.MkArvClient(containerToken) + runner.ContainerArvClient, runner.ContainerKeepClient, + runner.containerClient, err = runner.MkArvClient(containerToken) if err != nil { return fmt.Errorf("error creating container API client: %v", err) } - err = containerClient.Call("GET", "containers", runner.Container.UUID, "secret_mounts", nil, &sm) + err = runner.ContainerArvClient.Call("GET", "containers", runner.Container.UUID, "secret_mounts", nil, &sm) if err != nil { if apierr, ok := err.(arvadosclient.APIServerError); !ok || apierr.HttpStatusCode != 404 { return fmt.Errorf("error fetching secret_mounts: %v", err) @@ -1710,37 +1689,37 @@ func (runner *ContainerRunner) fetchContainerRecord() error { } // NewContainerRunner creates a new container runner. -func NewContainerRunner(client *arvados.Client, api IArvadosClient, kc IKeepClient, docker ThinDockerClient, containerUUID string) (*ContainerRunner, error) { +func NewContainerRunner(dispatcherClient *arvados.Client, + dispatcherArvClient IArvadosClient, + dispatcherKeepClient IKeepClient, + docker ThinDockerClient, + containerUUID string) (*ContainerRunner, error) { + cr := &ContainerRunner{ - client: client, - ArvClient: api, - Kc: kc, - Docker: docker, + dispatcherClient: dispatcherClient, + DispatcherArvClient: dispatcherArvClient, + DispatcherKeepClient: dispatcherKeepClient, + Docker: docker, } cr.NewLogWriter = cr.NewArvLogWriter cr.RunArvMount = cr.ArvMountCmd cr.MkTempDir = ioutil.TempDir - cr.ListProcesses = func() ([]PsProcess, error) { - pr, err := process.Processes() - if err != nil { - return nil, err - } - ps := make([]PsProcess, len(pr)) - for i, j := range pr { - ps[i] = j - } - return ps, nil - } - cr.MkArvClient = func(token string) (IArvadosClient, error) { + cr.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) { cl, err := arvadosclient.MakeArvadosClient() if err != nil { - return nil, err + return nil, nil, nil, err } cl.ApiToken = token - return cl, nil + kc, err := keepclient.MakeKeepClient(cl) + if err != nil { + return nil, nil, nil, err + } + c2 := arvados.NewClientFromEnv() + c2.AuthToken = token + return cl, kc, c2, nil } var err error - cr.LogCollection, err = (&arvados.Collection{}).FileSystem(cr.client, cr.Kc) + cr.LogCollection, err = (&arvados.Collection{}).FileSystem(cr.dispatcherClient, cr.DispatcherKeepClient) if err != nil { return nil, err } @@ -1752,7 +1731,7 @@ func NewContainerRunner(client *arvados.Client, api IArvadosClient, kc IKeepClie cr.CrunchLog = NewThrottledLogger(w) cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0) - loadLogThrottleParams(api) + loadLogThrottleParams(dispatcherArvClient) go cr.updateLogs() return cr, nil @@ -1764,6 +1743,11 @@ func main() { cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)") cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container") caCertsPath := flag.String("ca-certs", "", "Path to TLS root certificates") + detach := flag.Bool("detach", false, "Detach from parent process and run in the background") + stdinEnv := flag.Bool("stdin-env", false, "Load environment variables from JSON message on stdin") + sleep := flag.Duration("sleep", 0, "Delay before starting (testing use only)") + kill := flag.Int("kill", -1, "Send signal to an existing crunch-run process for given UUID") + list := flag.Bool("list", false, "List UUIDs of existing crunch-run processes") enableNetwork := flag.String("container-enable-networking", "default", `Specify if networking should be enabled for container. One of 'default', 'always': default: only enable networking if container requests it. @@ -1774,9 +1758,38 @@ func main() { `) memprofile := flag.String("memprofile", "", "write memory profile to `file` after running container") getVersion := flag.Bool("version", false, "Print version information and exit.") - checkContainerd := flag.Duration("check-containerd", 60*time.Second, "Periodic check if (docker-)containerd is running (use 0s to disable).") + flag.Duration("check-containerd", 0, "Ignored. Exists for compatibility with older versions.") + + ignoreDetachFlag := false + if len(os.Args) > 1 && os.Args[1] == "-no-detach" { + // This process was invoked by a parent process, which + // has passed along its own arguments, including + // -detach, after the leading -no-detach flag. Strip + // the leading -no-detach flag (it's not recognized by + // flag.Parse()) and ignore the -detach flag that + // comes later. + os.Args = append([]string{os.Args[0]}, os.Args[2:]...) + ignoreDetachFlag = true + } + flag.Parse() + if *stdinEnv && !ignoreDetachFlag { + // Load env vars on stdin if asked (but not in a + // detached child process, in which case stdin is + // /dev/null). + loadEnv(os.Stdin) + } + + switch { + case *detach && !ignoreDetachFlag: + os.Exit(Detach(flag.Arg(0), os.Args, os.Stdout, os.Stderr)) + case *kill >= 0: + os.Exit(KillProcess(flag.Arg(0), syscall.Signal(*kill), os.Stdout, os.Stderr)) + case *list: + os.Exit(ListProcesses(os.Stdout, os.Stderr)) + } + // Print version information if requested if *getVersion { fmt.Printf("crunch-run %s\n", version) @@ -1784,6 +1797,7 @@ func main() { } log.Printf("crunch-run %s started", version) + time.Sleep(*sleep) containerId := flag.Arg(0) @@ -1830,7 +1844,6 @@ func main() { cr.expectCgroupParent = *cgroupParent cr.enableNetwork = *enableNetwork cr.networkMode = *networkMode - cr.checkContainerd = *checkContainerd if *cgroupParentSubsystem != "" { p := findCgroup(*cgroupParentSubsystem) cr.setCgroupParent = p @@ -1858,3 +1871,21 @@ func main() { log.Fatalf("%s: %v", containerId, runerr) } } + +func loadEnv(rdr io.Reader) { + buf, err := ioutil.ReadAll(rdr) + if err != nil { + log.Fatalf("read stdin: %s", err) + } + var env map[string]string + err = json.Unmarshal(buf, &env) + if err != nil { + log.Fatalf("decode stdin: %s", err) + } + for k, v := range env { + err = os.Setenv(k, v) + if err != nil { + log.Fatalf("setenv(%q): %s", k, err) + } + } +}