X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/a7da936b6106bff17a7830124b1fd87ff01ad91e..2a748e79c3a72454d70e40f39fcad9dabf4943cc:/lib/dispatchcloud/worker/pool.go diff --git a/lib/dispatchcloud/worker/pool.go b/lib/dispatchcloud/worker/pool.go index 25b57c6e5b..e90935e2aa 100644 --- a/lib/dispatchcloud/worker/pool.go +++ b/lib/dispatchcloud/worker/pool.go @@ -5,6 +5,9 @@ package worker import ( + "crypto/rand" + "errors" + "fmt" "io" "sort" "strings" @@ -15,22 +18,26 @@ import ( "git.curoverse.com/arvados.git/sdk/go/arvados" "github.com/prometheus/client_golang/prometheus" "github.com/sirupsen/logrus" + "golang.org/x/crypto/ssh" ) const ( - tagKeyInstanceType = "InstanceType" - tagKeyHold = "Hold" + tagKeyInstanceType = "InstanceType" + tagKeyIdleBehavior = "IdleBehavior" + tagKeyInstanceSecret = "InstanceSecret" ) // An InstanceView shows a worker's current state and recent activity. type InstanceView struct { - Instance string - Price float64 - ArvadosInstanceType string - ProviderInstanceType string - LastContainerUUID string - LastBusy time.Time - WorkerState string + Instance cloud.InstanceID `json:"instance"` + Address string `json:"address"` + Price float64 `json:"price"` + ArvadosInstanceType string `json:"arvados_instance_type"` + ProviderInstanceType string `json:"provider_instance_type"` + LastContainerUUID string `json:"last_container_uuid"` + LastBusy time.Time `json:"last_busy"` + WorkerState string `json:"worker_state"` + IdleBehavior IdleBehavior `json:"idle_behavior"` } // An Executor executes shell commands on a remote host. @@ -61,6 +68,13 @@ const ( defaultTimeoutBooting = time.Minute * 10 defaultTimeoutProbe = time.Minute * 10 defaultTimeoutShutdown = time.Second * 10 + + // Time after a quota error to try again anyway, even if no + // instances have been shutdown. + quotaErrorTTL = time.Minute + + // Time between "X failed because rate limiting" messages + logRateLimitErrorInterval = time.Second * 10 ) func duration(conf arvados.Duration, def time.Duration) time.Duration { @@ -75,11 +89,11 @@ func duration(conf arvados.Duration, def time.Duration) time.Duration { // // New instances are configured and set up according to the given // cluster configuration. -func NewPool(logger logrus.FieldLogger, arvClient *arvados.Client, reg *prometheus.Registry, instanceSet cloud.InstanceSet, newExecutor func(cloud.Instance) Executor, cluster *arvados.Cluster) *Pool { +func NewPool(logger logrus.FieldLogger, arvClient *arvados.Client, reg *prometheus.Registry, instanceSet cloud.InstanceSet, newExecutor func(cloud.Instance) Executor, installPublicKey ssh.PublicKey, cluster *arvados.Cluster) *Pool { wp := &Pool{ logger: logger, arvClient: arvClient, - instanceSet: instanceSet, + instanceSet: &throttledInstanceSet{InstanceSet: instanceSet}, newExecutor: newExecutor, bootProbeCommand: cluster.CloudVMs.BootProbeCommand, imageID: cloud.ImageID(cluster.CloudVMs.ImageID), @@ -91,6 +105,7 @@ func NewPool(logger logrus.FieldLogger, arvClient *arvados.Client, reg *promethe timeoutBooting: duration(cluster.CloudVMs.TimeoutBooting, defaultTimeoutBooting), timeoutProbe: duration(cluster.CloudVMs.TimeoutProbe, defaultTimeoutProbe), timeoutShutdown: duration(cluster.CloudVMs.TimeoutShutdown, defaultTimeoutShutdown), + installPublicKey: installPublicKey, stop: make(chan bool), } wp.registerMetrics(reg) @@ -109,7 +124,7 @@ type Pool struct { // configuration logger logrus.FieldLogger arvClient *arvados.Client - instanceSet cloud.InstanceSet + instanceSet *throttledInstanceSet newExecutor func(cloud.Instance) Executor bootProbeCommand string imageID cloud.ImageID @@ -121,10 +136,11 @@ type Pool struct { timeoutBooting time.Duration timeoutProbe time.Duration timeoutShutdown time.Duration + installPublicKey ssh.PublicKey // private state subscribers map[<-chan struct{}]chan<- struct{} - creating map[arvados.InstanceType][]time.Time // start times of unfinished (InstanceSet)Create calls + creating map[string]createCall // unfinished (cloud.InstanceSet)Create calls (key is instance secret) workers map[cloud.InstanceID]*worker loaded bool // loaded list of instances from InstanceSet at least once exited map[string]time.Time // containers whose crunch-run proc has exited, but KillContainer has not been called @@ -134,23 +150,36 @@ type Pool struct { mtx sync.RWMutex setupOnce sync.Once - mInstances prometheus.Gauge + throttleCreate throttle + throttleInstances throttle + mContainersRunning prometheus.Gauge - mVCPUs prometheus.Gauge - mVCPUsInuse prometheus.Gauge - mMemory prometheus.Gauge - mMemoryInuse prometheus.Gauge + mInstances *prometheus.GaugeVec + mInstancesPrice *prometheus.GaugeVec + mVCPUs *prometheus.GaugeVec + mMemory *prometheus.GaugeVec +} + +type createCall struct { + time time.Time + instanceType arvados.InstanceType } -// Subscribe returns a channel that becomes ready whenever a worker's -// state changes. +// Subscribe returns a buffered channel that becomes ready after any +// change to the pool's state that could have scheduling implications: +// a worker's state changes, a new worker appears, the cloud +// provider's API rate limiting period ends, etc. +// +// Additional events that occur while the channel is already ready +// will be dropped, so it is OK if the caller services the channel +// slowly. // // Example: // // ch := wp.Subscribe() // defer wp.Unsubscribe(ch) // for range ch { -// // ...try scheduling some work... +// tryScheduling(wp) // if done { // break // } @@ -173,23 +202,36 @@ func (wp *Pool) Unsubscribe(ch <-chan struct{}) { } // Unallocated returns the number of unallocated (creating + booting + -// idle + unknown) workers for each instance type. +// idle + unknown) workers for each instance type. Workers in +// hold/drain mode are not included. func (wp *Pool) Unallocated() map[arvados.InstanceType]int { wp.setupOnce.Do(wp.setup) wp.mtx.RLock() defer wp.mtx.RUnlock() unalloc := map[arvados.InstanceType]int{} creating := map[arvados.InstanceType]int{} - for it, times := range wp.creating { - creating[it] = len(times) + oldestCreate := map[arvados.InstanceType]time.Time{} + for _, cc := range wp.creating { + it := cc.instanceType + creating[it]++ + if t, ok := oldestCreate[it]; !ok || t.After(cc.time) { + oldestCreate[it] = cc.time + } } for _, wkr := range wp.workers { - if !(wkr.state == StateIdle || wkr.state == StateBooting || wkr.state == StateUnknown) { + // Skip workers that are not expected to become + // available soon. Note len(wkr.running)>0 is not + // redundant here: it can be true even in + // StateUnknown. + if wkr.state == StateShutdown || + wkr.state == StateRunning || + wkr.idleBehavior != IdleBehaviorRun || + len(wkr.running) > 0 { continue } it := wkr.instType unalloc[it]++ - if wkr.state == StateUnknown && creating[it] > 0 && wkr.appeared.After(wp.creating[it][0]) { + if wkr.state == StateUnknown && creating[it] > 0 && wkr.appeared.After(oldestCreate[it]) { // If up to N new workers appear in // Instances() while we are waiting for N // Create() calls to complete, we assume we're @@ -214,41 +256,50 @@ func (wp *Pool) Unallocated() map[arvados.InstanceType]int { // Create a new instance with the given type, and add it to the worker // pool. The worker is added immediately; instance creation runs in // the background. -func (wp *Pool) Create(it arvados.InstanceType) error { +// +// Create returns false if a pre-existing error state prevents it from +// even attempting to create a new instance. Those errors are logged +// by the Pool, so the caller does not need to log anything in such +// cases. +func (wp *Pool) Create(it arvados.InstanceType) bool { logger := wp.logger.WithField("InstanceType", it.Name) wp.setupOnce.Do(wp.setup) wp.mtx.Lock() defer wp.mtx.Unlock() - if time.Now().Before(wp.atQuotaUntil) { - return wp.atQuotaErr + if time.Now().Before(wp.atQuotaUntil) || wp.throttleCreate.Error() != nil { + return false } - tags := cloud.InstanceTags{tagKeyInstanceType: it.Name} now := time.Now() - wp.creating[it] = append(wp.creating[it], now) + secret := randomHex(instanceSecretLength) + wp.creating[secret] = createCall{time: now, instanceType: it} go func() { defer wp.notify() - inst, err := wp.instanceSet.Create(it, wp.imageID, tags, nil) + tags := cloud.InstanceTags{ + tagKeyInstanceType: it.Name, + tagKeyIdleBehavior: string(IdleBehaviorRun), + tagKeyInstanceSecret: secret, + } + initCmd := cloud.InitCommand(fmt.Sprintf("umask 0177 && echo -n %q >%s", secret, instanceSecretFilename)) + inst, err := wp.instanceSet.Create(it, wp.imageID, tags, initCmd, wp.installPublicKey) wp.mtx.Lock() defer wp.mtx.Unlock() - // Remove our timestamp marker from wp.creating - for i, t := range wp.creating[it] { - if t == now { - copy(wp.creating[it][i:], wp.creating[it][i+1:]) - wp.creating[it] = wp.creating[it][:len(wp.creating[it])-1] - break - } - } - if err, ok := err.(cloud.QuotaError); ok && err.IsQuotaError() { - wp.atQuotaErr = err - wp.atQuotaUntil = time.Now().Add(time.Minute) - } + // delete() is deferred so the updateWorker() call + // below knows to use StateBooting when adding a new + // worker. + defer delete(wp.creating, secret) if err != nil { + if err, ok := err.(cloud.QuotaError); ok && err.IsQuotaError() { + wp.atQuotaErr = err + wp.atQuotaUntil = time.Now().Add(quotaErrorTTL) + time.AfterFunc(quotaErrorTTL, wp.notify) + } logger.WithError(err).Error("create failed") + wp.instanceSet.throttleCreate.CheckRateLimitError(err, wp.logger, "create instance", wp.notify) return } - wp.updateWorker(inst, it, StateBooting) + wp.updateWorker(inst, it) }() - return nil + return true } // AtQuota returns true if Create is not expected to work at the @@ -259,47 +310,82 @@ func (wp *Pool) AtQuota() bool { return time.Now().Before(wp.atQuotaUntil) } -// Add or update worker attached to the given instance. Use -// initialState if a new worker is created. +// SetIdleBehavior determines how the indicated instance will behave +// when it has no containers running. +func (wp *Pool) SetIdleBehavior(id cloud.InstanceID, idleBehavior IdleBehavior) error { + wp.mtx.Lock() + defer wp.mtx.Unlock() + wkr, ok := wp.workers[id] + if !ok { + return errors.New("requested instance does not exist") + } + wkr.idleBehavior = idleBehavior + wkr.saveTags() + wkr.shutdownIfIdle() + return nil +} + +// Add or update worker attached to the given instance. // // The second return value is true if a new worker is created. // +// A newly added instance has state=StateBooting if its tags match an +// entry in wp.creating, otherwise StateUnknown. +// // Caller must have lock. -func (wp *Pool) updateWorker(inst cloud.Instance, it arvados.InstanceType, initialState State) (*worker, bool) { +func (wp *Pool) updateWorker(inst cloud.Instance, it arvados.InstanceType) (*worker, bool) { + inst = tagVerifier{inst} id := inst.ID() if wkr := wp.workers[id]; wkr != nil { wkr.executor.SetTarget(inst) wkr.instance = inst wkr.updated = time.Now() - if initialState == StateBooting && wkr.state == StateUnknown { - wkr.state = StateBooting - } + wkr.saveTags() return wkr, false } - if initialState == StateUnknown && inst.Tags()[tagKeyHold] != "" { - initialState = StateHold + + state := StateUnknown + if _, ok := wp.creating[inst.Tags()[tagKeyInstanceSecret]]; ok { + state = StateBooting } + + // If an instance has a valid IdleBehavior tag when it first + // appears, initialize the new worker accordingly (this is how + // we restore IdleBehavior that was set by a prior dispatch + // process); otherwise, default to "run". After this, + // wkr.idleBehavior is the source of truth, and will only be + // changed via SetIdleBehavior(). + idleBehavior := IdleBehavior(inst.Tags()[tagKeyIdleBehavior]) + if !validIdleBehavior[idleBehavior] { + idleBehavior = IdleBehaviorRun + } + logger := wp.logger.WithFields(logrus.Fields{ "InstanceType": it.Name, - "Instance": inst, + "Instance": inst.ID(), + "Address": inst.Address(), }) - logger.WithField("State", initialState).Infof("instance appeared in cloud") + logger.WithFields(logrus.Fields{ + "State": state, + "IdleBehavior": idleBehavior, + }).Infof("instance appeared in cloud") now := time.Now() wkr := &worker{ - mtx: &wp.mtx, - wp: wp, - logger: logger, - executor: wp.newExecutor(inst), - state: initialState, - instance: inst, - instType: it, - appeared: now, - probed: now, - busy: now, - updated: now, - running: make(map[string]struct{}), - starting: make(map[string]struct{}), - probing: make(chan struct{}, 1), + mtx: &wp.mtx, + wp: wp, + logger: logger, + executor: wp.newExecutor(inst), + state: state, + idleBehavior: idleBehavior, + instance: inst, + instType: it, + appeared: now, + probed: now, + busy: now, + updated: now, + running: make(map[string]struct{}), + starting: make(map[string]struct{}), + probing: make(chan struct{}, 1), } wp.workers[id] = wkr return wkr, true @@ -322,7 +408,7 @@ func (wp *Pool) Shutdown(it arvados.InstanceType) bool { // TODO: shutdown the worker with the longest idle // time (Idle) or the earliest create time (Booting) for _, wkr := range wp.workers { - if wkr.state == tryState && wkr.instType == it { + if wkr.idleBehavior != IdleBehaviorHold && wkr.state == tryState && wkr.instType == it { logger.WithField("Instance", wkr.instance).Info("shutting down") wkr.shutdown() return true @@ -333,8 +419,12 @@ func (wp *Pool) Shutdown(it arvados.InstanceType) bool { } // CountWorkers returns the current number of workers in each state. +// +// CountWorkers blocks, if necessary, until the initial instance list +// has been loaded from the cloud provider. func (wp *Pool) CountWorkers() map[State]int { wp.setupOnce.Do(wp.setup) + wp.waitUntilLoaded() wp.mtx.Lock() defer wp.mtx.Unlock() r := map[State]int{} @@ -345,6 +435,12 @@ func (wp *Pool) CountWorkers() map[State]int { } // Running returns the container UUIDs being prepared/run on workers. +// +// In the returned map, the time value indicates when the Pool +// observed that the container process had exited. A container that +// has not yet exited has a zero time value. The caller should use +// KillContainer() to garbage-collect the entries for exited +// containers. func (wp *Pool) Running() map[string]time.Time { wp.setupOnce.Do(wp.setup) wp.mtx.Lock() @@ -410,10 +506,14 @@ func (wp *Pool) KillContainer(uuid string) { func (wp *Pool) kill(wkr *worker, uuid string) { logger := wp.logger.WithFields(logrus.Fields{ "ContainerUUID": uuid, - "Instance": wkr.instance, + "Instance": wkr.instance.ID(), }) logger.Debug("killing process") - stdout, stderr, err := wkr.executor.Execute(nil, "crunch-run --kill 15 "+uuid, nil) + cmd := "crunch-run --kill 15 " + uuid + if u := wkr.instance.RemoteUser(); u != "root" { + cmd = "sudo " + cmd + } + stdout, stderr, err := wkr.executor.Execute(nil, cmd, nil) if err != nil { logger.WithFields(logrus.Fields{ "stderr": string(stderr), @@ -439,13 +539,6 @@ func (wp *Pool) registerMetrics(reg *prometheus.Registry) { if reg == nil { reg = prometheus.NewRegistry() } - wp.mInstances = prometheus.NewGauge(prometheus.GaugeOpts{ - Namespace: "arvados", - Subsystem: "dispatchcloud", - Name: "instances_total", - Help: "Number of cloud VMs including pending, booting, running, held, and shutting down.", - }) - reg.MustRegister(wp.mInstances) wp.mContainersRunning = prometheus.NewGauge(prometheus.GaugeOpts{ Namespace: "arvados", Subsystem: "dispatchcloud", @@ -453,40 +546,40 @@ func (wp *Pool) registerMetrics(reg *prometheus.Registry) { Help: "Number of containers reported running by cloud VMs.", }) reg.MustRegister(wp.mContainersRunning) - - wp.mVCPUs = prometheus.NewGauge(prometheus.GaugeOpts{ + wp.mInstances = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "arvados", + Subsystem: "dispatchcloud", + Name: "instances_total", + Help: "Number of cloud VMs.", + }, []string{"category"}) + reg.MustRegister(wp.mInstances) + wp.mInstancesPrice = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "arvados", + Subsystem: "dispatchcloud", + Name: "instances_price", + Help: "Price of cloud VMs.", + }, []string{"category"}) + reg.MustRegister(wp.mInstancesPrice) + wp.mVCPUs = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: "arvados", Subsystem: "dispatchcloud", Name: "vcpus_total", Help: "Total VCPUs on all cloud VMs.", - }) + }, []string{"category"}) reg.MustRegister(wp.mVCPUs) - wp.mVCPUsInuse = prometheus.NewGauge(prometheus.GaugeOpts{ - Namespace: "arvados", - Subsystem: "dispatchcloud", - Name: "vcpus_inuse", - Help: "VCPUs on cloud VMs that are running containers.", - }) - reg.MustRegister(wp.mVCPUsInuse) - wp.mMemory = prometheus.NewGauge(prometheus.GaugeOpts{ + wp.mMemory = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: "arvados", Subsystem: "dispatchcloud", Name: "memory_bytes_total", Help: "Total memory on all cloud VMs.", - }) + }, []string{"category"}) reg.MustRegister(wp.mMemory) - wp.mMemoryInuse = prometheus.NewGauge(prometheus.GaugeOpts{ - Namespace: "arvados", - Subsystem: "dispatchcloud", - Name: "memory_bytes_inuse", - Help: "Memory on cloud VMs that are running containers.", - }) - reg.MustRegister(wp.mMemoryInuse) } func (wp *Pool) runMetrics() { ch := wp.Subscribe() defer wp.Unsubscribe(ch) + wp.updateMetrics() for range ch { wp.updateMetrics() } @@ -496,23 +589,38 @@ func (wp *Pool) updateMetrics() { wp.mtx.RLock() defer wp.mtx.RUnlock() - var alloc, cpu, cpuInuse, mem, memInuse int64 + instances := map[string]int64{} + price := map[string]float64{} + cpu := map[string]int64{} + mem := map[string]int64{} + var running int64 for _, wkr := range wp.workers { - cpu += int64(wkr.instType.VCPUs) - mem += int64(wkr.instType.RAM) - if len(wkr.running)+len(wkr.starting) == 0 { - continue + var cat string + switch { + case len(wkr.running)+len(wkr.starting) > 0: + cat = "inuse" + case wkr.idleBehavior == IdleBehaviorHold: + cat = "hold" + case wkr.state == StateBooting: + cat = "booting" + case wkr.state == StateUnknown: + cat = "unknown" + default: + cat = "idle" } - alloc += int64(len(wkr.running) + len(wkr.starting)) - cpuInuse += int64(wkr.instType.VCPUs) - memInuse += int64(wkr.instType.RAM) - } - wp.mInstances.Set(float64(len(wp.workers))) - wp.mContainersRunning.Set(float64(alloc)) - wp.mVCPUs.Set(float64(cpu)) - wp.mMemory.Set(float64(mem)) - wp.mVCPUsInuse.Set(float64(cpuInuse)) - wp.mMemoryInuse.Set(float64(memInuse)) + instances[cat]++ + price[cat] += wkr.instType.Price + cpu[cat] += int64(wkr.instType.VCPUs) + mem[cat] += int64(wkr.instType.RAM) + running += int64(len(wkr.running) + len(wkr.starting)) + } + for _, cat := range []string{"inuse", "hold", "booting", "unknown", "idle"} { + wp.mInstances.WithLabelValues(cat).Set(float64(instances[cat])) + wp.mInstancesPrice.WithLabelValues(cat).Set(price[cat]) + wp.mVCPUs.WithLabelValues(cat).Set(float64(cpu[cat])) + wp.mMemory.WithLabelValues(cat).Set(float64(mem[cat])) + } + wp.mContainersRunning.Set(float64(running)) } func (wp *Pool) runProbes() { @@ -590,24 +698,26 @@ func (wp *Pool) Instances() []InstanceView { wp.mtx.Lock() for _, w := range wp.workers { r = append(r, InstanceView{ - Instance: w.instance.String(), + Instance: w.instance.ID(), + Address: w.instance.Address(), Price: w.instType.Price, ArvadosInstanceType: w.instType.Name, ProviderInstanceType: w.instType.ProviderType, LastContainerUUID: w.lastUUID, LastBusy: w.busy, WorkerState: w.state.String(), + IdleBehavior: w.idleBehavior, }) } wp.mtx.Unlock() sort.Slice(r, func(i, j int) bool { - return strings.Compare(r[i].Instance, r[j].Instance) < 0 + return strings.Compare(string(r[i].Instance), string(r[j].Instance)) < 0 }) return r } func (wp *Pool) setup() { - wp.creating = map[arvados.InstanceType][]time.Time{} + wp.creating = map[string]createCall{} wp.exited = map[string]time.Time{} wp.workers = map[cloud.InstanceID]*worker{} wp.subscribers = map[<-chan struct{}]chan<- struct{}{} @@ -626,10 +736,14 @@ func (wp *Pool) notify() { func (wp *Pool) getInstancesAndSync() error { wp.setupOnce.Do(wp.setup) + if err := wp.instanceSet.throttleInstances.Error(); err != nil { + return err + } wp.logger.Debug("getting instance list") threshold := time.Now() instances, err := wp.instanceSet.Instances(cloud.InstanceTags{}) if err != nil { + wp.instanceSet.throttleInstances.CheckRateLimitError(err, wp.logger, "list instances", wp.notify) return err } wp.sync(threshold, instances) @@ -653,7 +767,7 @@ func (wp *Pool) sync(threshold time.Time, instances []cloud.Instance) { wp.logger.WithField("Instance", inst).Errorf("unknown InstanceType tag %q --- ignoring", itTag) continue } - if wkr, isNew := wp.updateWorker(inst, it, StateUnknown); isNew { + if wkr, isNew := wp.updateWorker(inst, it); isNew { notify = true } else if wkr.state == StateShutdown && time.Since(wkr.destroyed) > wp.timeoutShutdown { wp.logger.WithField("Instance", inst).Info("worker still listed after shutdown; retrying") @@ -666,7 +780,7 @@ func (wp *Pool) sync(threshold time.Time, instances []cloud.Instance) { continue } logger := wp.logger.WithFields(logrus.Fields{ - "Instance": wkr.instance, + "Instance": wkr.instance.ID(), "WorkerState": wkr.state, }) logger.Info("instance disappeared in cloud") @@ -676,6 +790,7 @@ func (wp *Pool) sync(threshold time.Time, instances []cloud.Instance) { } if !wp.loaded { + notify = true wp.loaded = true wp.logger.WithField("N", len(wp.workers)).Info("loaded initial instance list") } @@ -684,3 +799,25 @@ func (wp *Pool) sync(threshold time.Time, instances []cloud.Instance) { go wp.notify() } } + +func (wp *Pool) waitUntilLoaded() { + ch := wp.Subscribe() + wp.mtx.RLock() + defer wp.mtx.RUnlock() + for !wp.loaded { + wp.mtx.RUnlock() + <-ch + wp.mtx.RLock() + } +} + +// Return a random string of n hexadecimal digits (n*4 random bits). n +// must be even. +func randomHex(n int) string { + buf := make([]byte, n/2) + _, err := rand.Read(buf) + if err != nil { + panic(err) + } + return fmt.Sprintf("%x", buf) +}