lib/dispatchcloud/worker/worker.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package worker
   6
   7 import (
   8         "bytes"
   9         "fmt"
  10         "path/filepath"
  11         "strings"
  12         "sync"
  13         "time"
  14
  15         "git.arvados.org/arvados.git/lib/cloud"
  16         "git.arvados.org/arvados.git/sdk/go/arvados"
  17         "git.arvados.org/arvados.git/sdk/go/stats"
  18         "github.com/sirupsen/logrus"
  19 )
  20
  21 const (
  22         // TODO: configurable
  23         maxPingFailTime = 10 * time.Minute
  24 )
  25
  26 // State indicates whether a worker is available to do work, and (if
  27 // not) whether/when it is expected to become ready.
  28 type State int
  29
  30 const (
  31         StateUnknown  State = iota // might be running a container already
  32         StateBooting               // instance is booting
  33         StateIdle                  // instance booted, no containers are running
  34         StateRunning               // instance is running one or more containers
  35         StateShutdown              // worker has stopped monitoring the instance
  36 )
  37
  38 var stateString = map[State]string{
  39         StateUnknown:  "unknown",
  40         StateBooting:  "booting",
  41         StateIdle:     "idle",
  42         StateRunning:  "running",
  43         StateShutdown: "shutdown",
  44 }
  45
  46 // String implements fmt.Stringer.
  47 func (s State) String() string {
  48         return stateString[s]
  49 }
  50
  51 // MarshalText implements encoding.TextMarshaler so a JSON encoding of
  52 // map[State]anything uses the state's string representation.
  53 func (s State) MarshalText() ([]byte, error) {
  54         return []byte(stateString[s]), nil
  55 }
  56
  57 // BootOutcome is the result of a worker boot. It is used as a label in a metric.
  58 type BootOutcome string
  59
  60 const (
  61         BootOutcomeFailed      BootOutcome = "failure"
  62         BootOutcomeSucceeded   BootOutcome = "success"
  63         BootOutcomeAborted     BootOutcome = "aborted"
  64         BootOutcomeDisappeared BootOutcome = "disappeared"
  65 )
  66
  67 var validBootOutcomes = map[BootOutcome]bool{
  68         BootOutcomeFailed:      true,
  69         BootOutcomeSucceeded:   true,
  70         BootOutcomeAborted:     true,
  71         BootOutcomeDisappeared: true,
  72 }
  73
  74 // IdleBehavior indicates the behavior desired when a node becomes idle.
  75 type IdleBehavior string
  76
  77 const (
  78         IdleBehaviorRun   IdleBehavior = "run"   // run containers, or shutdown on idle timeout
  79         IdleBehaviorHold  IdleBehavior = "hold"  // don't shutdown or run more containers
  80         IdleBehaviorDrain IdleBehavior = "drain" // shutdown immediately when idle
  81 )
  82
  83 var validIdleBehavior = map[IdleBehavior]bool{
  84         IdleBehaviorRun:   true,
  85         IdleBehaviorHold:  true,
  86         IdleBehaviorDrain: true,
  87 }
  88
  89 type worker struct {
  90         logger   logrus.FieldLogger
  91         executor Executor
  92         wp       *Pool
  93
  94         mtx                 sync.Locker // must be wp's Locker.
  95         state               State
  96         idleBehavior        IdleBehavior
  97         instance            cloud.Instance
  98         instType            arvados.InstanceType
  99         vcpus               int64
 100         memory              int64
 101         appeared            time.Time
 102         probed              time.Time
 103         updated             time.Time
 104         busy                time.Time
 105         destroyed           time.Time
 106         firstSSHConnection  time.Time
 107         lastUUID            string
 108         running             map[string]*remoteRunner // remember to update state idle<->running when this changes
 109         starting            map[string]*remoteRunner // remember to update state idle<->running when this changes
 110         probing             chan struct{}
 111         bootOutcomeReported bool
 112         timeToReadyReported bool
 113         staleRunLockSince   time.Time
 114 }
 115
 116 func (wkr *worker) onUnkillable(uuid string) {
 117         wkr.mtx.Lock()
 118         defer wkr.mtx.Unlock()
 119         logger := wkr.logger.WithField("ContainerUUID", uuid)
 120         if wkr.idleBehavior == IdleBehaviorHold {
 121                 logger.Warn("unkillable container, but worker has IdleBehavior=Hold")
 122                 return
 123         }
 124         logger.Warn("unkillable container, draining worker")
 125         wkr.setIdleBehavior(IdleBehaviorDrain)
 126 }
 127
 128 func (wkr *worker) onKilled(uuid string) {
 129         wkr.mtx.Lock()
 130         defer wkr.mtx.Unlock()
 131         wkr.closeRunner(uuid)
 132         go wkr.wp.notify()
 133 }
 134
 135 // caller must have lock.
 136 func (wkr *worker) reportBootOutcome(outcome BootOutcome) {
 137         if wkr.bootOutcomeReported {
 138                 return
 139         }
 140         if wkr.wp.mBootOutcomes != nil {
 141                 wkr.wp.mBootOutcomes.WithLabelValues(string(outcome)).Inc()
 142         }
 143         wkr.bootOutcomeReported = true
 144 }
 145
 146 // caller must have lock.
 147 func (wkr *worker) reportTimeBetweenFirstSSHAndReadyForContainer() {
 148         if wkr.timeToReadyReported {
 149                 return
 150         }
 151         if wkr.wp.mTimeToSSH != nil {
 152                 wkr.wp.mTimeToReadyForContainer.Observe(time.Since(wkr.firstSSHConnection).Seconds())
 153         }
 154         wkr.timeToReadyReported = true
 155 }
 156
 157 // caller must have lock.
 158 func (wkr *worker) setIdleBehavior(idleBehavior IdleBehavior) {
 159         wkr.logger.WithField("IdleBehavior", idleBehavior).Info("set idle behavior")
 160         wkr.idleBehavior = idleBehavior
 161         wkr.saveTags()
 162         wkr.shutdownIfIdle()
 163 }
 164
 165 // caller must have lock.
 166 func (wkr *worker) startContainer(ctr arvados.Container) {
 167         logger := wkr.logger.WithFields(logrus.Fields{
 168                 "ContainerUUID": ctr.UUID,
 169                 "Priority":      ctr.Priority,
 170         })
 171         logger.Debug("starting container")
 172         rr := newRemoteRunner(ctr.UUID, wkr)
 173         wkr.starting[ctr.UUID] = rr
 174         if wkr.state != StateRunning {
 175                 wkr.state = StateRunning
 176                 go wkr.wp.notify()
 177         }
 178         go func() {
 179                 rr.Start()
 180                 if wkr.wp.mTimeFromQueueToCrunchRun != nil {
 181                         wkr.wp.mTimeFromQueueToCrunchRun.Observe(time.Since(ctr.CreatedAt).Seconds())
 182                 }
 183                 wkr.mtx.Lock()
 184                 defer wkr.mtx.Unlock()
 185                 now := time.Now()
 186                 wkr.updated = now
 187                 wkr.busy = now
 188                 delete(wkr.starting, ctr.UUID)
 189                 wkr.running[ctr.UUID] = rr
 190                 wkr.lastUUID = ctr.UUID
 191         }()
 192 }
 193
 194 // ProbeAndUpdate conducts appropriate boot/running probes (if any)
 195 // for the worker's curent state. If a previous probe is still
 196 // running, it does nothing.
 197 //
 198 // It should be called in a new goroutine.
 199 func (wkr *worker) ProbeAndUpdate() {
 200         select {
 201         case wkr.probing <- struct{}{}:
 202                 wkr.probeAndUpdate()
 203                 <-wkr.probing
 204         default:
 205                 wkr.logger.Debug("still waiting for last probe to finish")
 206         }
 207 }
 208
 209 // probeAndUpdate calls probeBooted and/or probeRunning if needed, and
 210 // updates state accordingly.
 211 //
 212 // In StateUnknown: Call both probeBooted and probeRunning.
 213 // In StateBooting: Call probeBooted; if successful, call probeRunning.
 214 // In StateRunning: Call probeRunning.
 215 // In StateIdle: Call probeRunning.
 216 // In StateShutdown: Do nothing.
 217 //
 218 // If both probes succeed, wkr.state changes to
 219 // StateIdle/StateRunning.
 220 //
 221 // If probeRunning succeeds, wkr.running is updated. (This means
 222 // wkr.running might be non-empty even in StateUnknown, if the boot
 223 // probe failed.)
 224 //
 225 // probeAndUpdate should be called in a new goroutine.
 226 func (wkr *worker) probeAndUpdate() {
 227         wkr.mtx.Lock()
 228         updated := wkr.updated
 229         initialState := wkr.state
 230         wkr.mtx.Unlock()
 231
 232         var (
 233                 booted   bool
 234                 ctrUUIDs []string
 235                 ok       bool
 236                 stderr   []byte // from probeBooted
 237         )
 238
 239         switch initialState {
 240         case StateShutdown:
 241                 return
 242         case StateIdle, StateRunning:
 243                 booted = true
 244         case StateUnknown, StateBooting:
 245         default:
 246                 panic(fmt.Sprintf("unknown state %s", initialState))
 247         }
 248
 249         probeStart := time.Now()
 250         logger := wkr.logger.WithField("ProbeStart", probeStart)
 251
 252         if !booted {
 253                 booted, stderr = wkr.probeBooted()
 254                 if !booted {
 255                         // Pretend this probe succeeded if another
 256                         // concurrent attempt succeeded.
 257                         wkr.mtx.Lock()
 258                         booted = wkr.state == StateRunning || wkr.state == StateIdle
 259                         wkr.mtx.Unlock()
 260                 }
 261                 if booted {
 262                         logger.Info("instance booted; will try probeRunning")
 263                 }
 264         }
 265         reportedBroken := false
 266         if booted || wkr.state == StateUnknown {
 267                 ctrUUIDs, reportedBroken, ok = wkr.probeRunning()
 268         }
 269         wkr.mtx.Lock()
 270         defer wkr.mtx.Unlock()
 271         if reportedBroken && wkr.idleBehavior == IdleBehaviorRun {
 272                 logger.Info("probe reported broken instance")
 273                 wkr.reportBootOutcome(BootOutcomeFailed)
 274                 wkr.setIdleBehavior(IdleBehaviorDrain)
 275         }
 276         if !ok || (!booted && len(ctrUUIDs) == 0 && len(wkr.running) == 0) {
 277                 if wkr.state == StateShutdown && wkr.updated.After(updated) {
 278                         // Skip the logging noise if shutdown was
 279                         // initiated during probe.
 280                         return
 281                 }
 282                 // Using the start time of the probe as the timeout
 283                 // threshold ensures we always initiate at least one
 284                 // probe attempt after the boot/probe timeout expires
 285                 // (otherwise, a slow probe failure could cause us to
 286                 // shutdown an instance even though it did in fact
 287                 // boot/recover before the timeout expired).
 288                 dur := probeStart.Sub(wkr.probed)
 289                 if wkr.shutdownIfBroken(dur) {
 290                         // stderr from failed run-probes will have
 291                         // been logged already, but boot-probe
 292                         // failures are normal so they are logged only
 293                         // at Debug level. This is our chance to log
 294                         // some evidence about why the node never
 295                         // booted, even in non-debug mode.
 296                         if !booted {
 297                                 wkr.reportBootOutcome(BootOutcomeFailed)
 298                                 logger.WithFields(logrus.Fields{
 299                                         "Duration": dur,
 300                                         "stderr":   string(stderr),
 301                                 }).Info("boot failed")
 302                         }
 303                 }
 304                 return
 305         }
 306
 307         updateTime := time.Now()
 308         wkr.probed = updateTime
 309
 310         if updated != wkr.updated {
 311                 // Worker was updated after the probe began, so
 312                 // wkr.running might have a container UUID that was
 313                 // not yet running when ctrUUIDs was generated. Leave
 314                 // wkr.running alone and wait for the next probe to
 315                 // catch up on any changes.
 316                 return
 317         }
 318
 319         if len(ctrUUIDs) > 0 {
 320                 wkr.busy = updateTime
 321                 wkr.lastUUID = ctrUUIDs[0]
 322         } else if len(wkr.running) > 0 {
 323                 // Actual last-busy time was sometime between wkr.busy
 324                 // and now. Now is the earliest opportunity to take
 325                 // advantage of the non-busy state, though.
 326                 wkr.busy = updateTime
 327         }
 328
 329         changed := wkr.updateRunning(ctrUUIDs)
 330
 331         // Update state if this was the first successful boot-probe.
 332         if booted && (wkr.state == StateUnknown || wkr.state == StateBooting) {
 333                 if wkr.state == StateBooting {
 334                         wkr.reportTimeBetweenFirstSSHAndReadyForContainer()
 335                 }
 336                 // Note: this will change again below if
 337                 // len(wkr.starting)+len(wkr.running) > 0.
 338                 wkr.state = StateIdle
 339                 changed = true
 340         }
 341
 342         // If wkr.state and wkr.running aren't changing then there's
 343         // no need to log anything, notify the scheduler, move state
 344         // back and forth between idle/running, etc.
 345         if !changed {
 346                 return
 347         }
 348
 349         // Log whenever a run-probe reveals crunch-run processes
 350         // appearing/disappearing before boot-probe succeeds.
 351         if wkr.state == StateUnknown && changed {
 352                 logger.WithFields(logrus.Fields{
 353                         "RunningContainers": len(wkr.running),
 354                         "State":             wkr.state,
 355                 }).Info("crunch-run probe succeeded, but boot probe is still failing")
 356         }
 357
 358         if wkr.state == StateIdle && len(wkr.starting)+len(wkr.running) > 0 {
 359                 wkr.state = StateRunning
 360         } else if wkr.state == StateRunning && len(wkr.starting)+len(wkr.running) == 0 {
 361                 wkr.state = StateIdle
 362         }
 363         wkr.updated = updateTime
 364         if booted && (initialState == StateUnknown || initialState == StateBooting) {
 365                 wkr.reportBootOutcome(BootOutcomeSucceeded)
 366                 logger.WithFields(logrus.Fields{
 367                         "RunningContainers": len(wkr.running),
 368                         "State":             wkr.state,
 369                 }).Info("probes succeeded, instance is in service")
 370         }
 371         go wkr.wp.notify()
 372 }
 373
 374 func (wkr *worker) probeRunning() (running []string, reportsBroken, ok bool) {
 375         cmd := wkr.wp.runnerCmd + " --list"
 376         if u := wkr.instance.RemoteUser(); u != "root" {
 377                 cmd = "sudo " + cmd
 378         }
 379         stdout, stderr, err := wkr.executor.Execute(nil, cmd, nil)
 380         if err != nil {
 381                 wkr.logger.WithFields(logrus.Fields{
 382                         "Command": cmd,
 383                         "stdout":  string(stdout),
 384                         "stderr":  string(stderr),
 385                 }).WithError(err).Warn("probe failed")
 386                 return
 387         }
 388         ok = true
 389
 390         staleRunLock := false
 391         for _, s := range strings.Split(string(stdout), "\n") {
 392                 // Each line of the "crunch-run --list" output is one
 393                 // of the following:
 394                 //
 395                 // * a container UUID, indicating that processes
 396                 //   related to that container are currently running.
 397                 //   Optionally followed by " stale", indicating that
 398                 //   the crunch-run process itself has exited (the
 399                 //   remaining process is probably arv-mount).
 400                 //
 401                 // * the string "broken", indicating that the instance
 402                 //   appears incapable of starting containers.
 403                 //
 404                 // See ListProcesses() in lib/crunchrun/background.go.
 405                 if s == "" {
 406                         // empty string following final newline
 407                 } else if s == "broken" {
 408                         reportsBroken = true
 409                 } else if toks := strings.Split(s, " "); len(toks) == 1 {
 410                         running = append(running, s)
 411                 } else if toks[1] == "stale" {
 412                         wkr.logger.WithField("ContainerUUID", toks[0]).Info("probe reported stale run lock")
 413                         staleRunLock = true
 414                 }
 415         }
 416         wkr.mtx.Lock()
 417         defer wkr.mtx.Unlock()
 418         if !staleRunLock {
 419                 wkr.staleRunLockSince = time.Time{}
 420         } else if wkr.staleRunLockSince.IsZero() {
 421                 wkr.staleRunLockSince = time.Now()
 422         } else if dur := time.Now().Sub(wkr.staleRunLockSince); dur > wkr.wp.timeoutStaleRunLock {
 423                 wkr.logger.WithField("Duration", dur).Warn("reporting broken after reporting stale run lock for too long")
 424                 reportsBroken = true
 425         }
 426         return
 427 }
 428
 429 func (wkr *worker) probeBooted() (ok bool, stderr []byte) {
 430         cmd := wkr.wp.bootProbeCommand
 431         if cmd == "" {
 432                 cmd = "true"
 433         }
 434         stdout, stderr, err := wkr.executor.Execute(nil, cmd, nil)
 435         logger := wkr.logger.WithFields(logrus.Fields{
 436                 "Command": cmd,
 437                 "stdout":  string(stdout),
 438                 "stderr":  string(stderr),
 439         })
 440         if err != nil {
 441                 logger.WithError(err).Debug("boot probe failed")
 442                 return false, stderr
 443         }
 444         logger.Info("boot probe succeeded")
 445         if err = wkr.wp.loadRunnerData(); err != nil {
 446                 wkr.logger.WithError(err).Warn("cannot boot worker: error loading runner binary")
 447                 return false, stderr
 448         } else if len(wkr.wp.runnerData) == 0 {
 449                 // Assume crunch-run is already installed
 450         } else if _, stderr2, err := wkr.copyRunnerData(); err != nil {
 451                 wkr.logger.WithError(err).WithField("stderr", string(stderr2)).Warn("error copying runner binary")
 452                 return false, stderr2
 453         } else {
 454                 stderr = append(stderr, stderr2...)
 455         }
 456         return true, stderr
 457 }
 458
 459 func (wkr *worker) copyRunnerData() (stdout, stderr []byte, err error) {
 460         hash := fmt.Sprintf("%x", wkr.wp.runnerMD5)
 461         dstdir, _ := filepath.Split(wkr.wp.runnerCmd)
 462         logger := wkr.logger.WithFields(logrus.Fields{
 463                 "hash": hash,
 464                 "path": wkr.wp.runnerCmd,
 465         })
 466
 467         stdout, stderr, err = wkr.executor.Execute(nil, `md5sum `+wkr.wp.runnerCmd, nil)
 468         if err == nil && len(stderr) == 0 && bytes.Equal(stdout, []byte(hash+"  "+wkr.wp.runnerCmd+"\n")) {
 469                 logger.Info("runner binary already exists on worker, with correct hash")
 470                 return
 471         }
 472
 473         // Note touch+chmod come before writing data, to avoid the
 474         // possibility of md5 being correct while file mode is
 475         // incorrect.
 476         cmd := `set -e; dstdir="` + dstdir + `"; dstfile="` + wkr.wp.runnerCmd + `"; mkdir -p "$dstdir"; touch "$dstfile"; chmod 0755 "$dstdir" "$dstfile"; cat >"$dstfile"`
 477         if wkr.instance.RemoteUser() != "root" {
 478                 cmd = `sudo sh -c '` + strings.Replace(cmd, "'", "'\\''", -1) + `'`
 479         }
 480         logger.WithField("cmd", cmd).Info("installing runner binary on worker")
 481         stdout, stderr, err = wkr.executor.Execute(nil, cmd, bytes.NewReader(wkr.wp.runnerData))
 482         return
 483 }
 484
 485 // caller must have lock.
 486 func (wkr *worker) shutdownIfBroken(dur time.Duration) bool {
 487         if wkr.idleBehavior == IdleBehaviorHold {
 488                 // Never shut down.
 489                 return false
 490         }
 491         label, threshold := "", wkr.wp.timeoutProbe
 492         if wkr.state == StateUnknown || wkr.state == StateBooting {
 493                 label, threshold = "new ", wkr.wp.timeoutBooting
 494         }
 495         if dur < threshold {
 496                 return false
 497         }
 498         wkr.logger.WithFields(logrus.Fields{
 499                 "Duration": dur,
 500                 "Since":    wkr.probed,
 501                 "State":    wkr.state,
 502         }).Warnf("%sinstance unresponsive, shutting down", label)
 503         wkr.shutdown()
 504         return true
 505 }
 506
 507 // Returns true if the instance is eligible for shutdown: either it's
 508 // been idle too long, or idleBehavior=Drain and nothing is running.
 509 //
 510 // caller must have lock.
 511 func (wkr *worker) eligibleForShutdown() bool {
 512         if wkr.idleBehavior == IdleBehaviorHold {
 513                 return false
 514         }
 515         draining := wkr.idleBehavior == IdleBehaviorDrain
 516         switch wkr.state {
 517         case StateBooting:
 518                 return draining
 519         case StateIdle:
 520                 return draining || time.Since(wkr.busy) >= wkr.wp.timeoutIdle
 521         case StateRunning:
 522                 if !draining {
 523                         return false
 524                 }
 525                 for _, rr := range wkr.running {
 526                         if !rr.givenup {
 527                                 return false
 528                         }
 529                 }
 530                 for _, rr := range wkr.starting {
 531                         if !rr.givenup {
 532                                 return false
 533                         }
 534                 }
 535                 // draining, and all remaining runners are just trying
 536                 // to force-kill their crunch-run procs
 537                 return true
 538         default:
 539                 return false
 540         }
 541 }
 542
 543 // caller must have lock.
 544 func (wkr *worker) shutdownIfIdle() bool {
 545         if !wkr.eligibleForShutdown() {
 546                 return false
 547         }
 548         wkr.logger.WithFields(logrus.Fields{
 549                 "State":        wkr.state,
 550                 "IdleDuration": stats.Duration(time.Since(wkr.busy)),
 551                 "IdleBehavior": wkr.idleBehavior,
 552         }).Info("shutdown worker")
 553         wkr.reportBootOutcome(BootOutcomeAborted)
 554         wkr.shutdown()
 555         return true
 556 }
 557
 558 // caller must have lock.
 559 func (wkr *worker) shutdown() {
 560         now := time.Now()
 561         wkr.updated = now
 562         wkr.destroyed = now
 563         wkr.state = StateShutdown
 564         go wkr.wp.notify()
 565         go func() {
 566                 err := wkr.instance.Destroy()
 567                 if err != nil {
 568                         wkr.logger.WithError(err).Warn("shutdown failed")
 569                         return
 570                 }
 571         }()
 572 }
 573
 574 // Save worker tags to cloud provider metadata, if they don't already
 575 // match. Caller must have lock.
 576 func (wkr *worker) saveTags() {
 577         instance := wkr.instance
 578         tags := instance.Tags()
 579         update := cloud.InstanceTags{
 580                 wkr.wp.tagKeyPrefix + tagKeyInstanceType: wkr.instType.Name,
 581                 wkr.wp.tagKeyPrefix + tagKeyIdleBehavior: string(wkr.idleBehavior),
 582         }
 583         save := false
 584         for k, v := range update {
 585                 if tags[k] != v {
 586                         tags[k] = v
 587                         save = true
 588                 }
 589         }
 590         if save {
 591                 go func() {
 592                         err := instance.SetTags(tags)
 593                         if err != nil {
 594                                 wkr.wp.logger.WithField("Instance", instance.ID()).WithError(err).Warnf("error updating tags")
 595                         }
 596                 }()
 597         }
 598 }
 599
 600 func (wkr *worker) Close() {
 601         // This might take time, so do it after unlocking mtx.
 602         defer wkr.executor.Close()
 603
 604         wkr.mtx.Lock()
 605         defer wkr.mtx.Unlock()
 606         for uuid, rr := range wkr.running {
 607                 wkr.logger.WithField("ContainerUUID", uuid).Info("crunch-run process abandoned")
 608                 rr.Close()
 609         }
 610         for uuid, rr := range wkr.starting {
 611                 wkr.logger.WithField("ContainerUUID", uuid).Info("crunch-run process abandoned")
 612                 rr.Close()
 613         }
 614 }
 615
 616 // Add/remove entries in wkr.running to match ctrUUIDs returned by a
 617 // probe. Returns true if anything was added or removed.
 618 //
 619 // Caller must have lock.
 620 func (wkr *worker) updateRunning(ctrUUIDs []string) (changed bool) {
 621         alive := map[string]bool{}
 622         for _, uuid := range ctrUUIDs {
 623                 alive[uuid] = true
 624                 if _, ok := wkr.running[uuid]; ok {
 625                         // unchanged
 626                 } else if rr, ok := wkr.starting[uuid]; ok {
 627                         wkr.running[uuid] = rr
 628                         delete(wkr.starting, uuid)
 629                         changed = true
 630                 } else {
 631                         // We didn't start it -- it must have been
 632                         // started by a previous dispatcher process.
 633                         wkr.logger.WithField("ContainerUUID", uuid).Info("crunch-run process detected")
 634                         wkr.running[uuid] = newRemoteRunner(uuid, wkr)
 635                         changed = true
 636                 }
 637         }
 638         for uuid := range wkr.running {
 639                 if !alive[uuid] {
 640                         wkr.closeRunner(uuid)
 641                         changed = true
 642                 }
 643         }
 644         return
 645 }
 646
 647 // caller must have lock.
 648 func (wkr *worker) closeRunner(uuid string) {
 649         rr := wkr.running[uuid]
 650         if rr == nil {
 651                 return
 652         }
 653         wkr.logger.WithField("ContainerUUID", uuid).Info("crunch-run process ended")
 654         delete(wkr.running, uuid)
 655         rr.Close()
 656
 657         now := time.Now()
 658         wkr.updated = now
 659         wkr.wp.exited[uuid] = now
 660         if wkr.state == StateRunning && len(wkr.running)+len(wkr.starting) == 0 {
 661                 wkr.state = StateIdle
 662         }
 663 }