21123: Add container status API to cloud dispatcher.

[arvados.git] / lib / dispatchcloud / scheduler / scheduler.go

diff --git a/lib/dispatchcloud/scheduler/scheduler.go b/lib/dispatchcloud/scheduler/scheduler.go
index 589aa3ec1140774a5446fdcf0331cb5d781734f7..bc6574a21a538134c618320f9e97511b84d9b307 100644 (file)
--- a/lib/dispatchcloud/scheduler/scheduler.go
+++ b/lib/dispatchcloud/scheduler/scheduler.go
@@ -9,6 +9,7 @@ package scheduler
 import (
        "context"
        "sync"
+       "sync/atomic"
        "time"
 
        "git.arvados.org/arvados.git/sdk/go/arvados"
@@ -46,19 +47,26 @@ type Scheduler struct {
        stop    chan struct{}
        stopped chan struct{}
 
-       last503time    time.Time // last time API responded 503
-       maxConcurrency int       // dynamic container limit (0 = unlimited), see runQueue()
+       last503time          time.Time // last time API responded 503
+       maxConcurrency       int       // dynamic container limit (0 = unlimited), see runQueue()
+       supervisorFraction   float64   // maximum fraction of "supervisor" containers (these are containers who's main job is to launch other containers, e.g. workflow runners)
+       maxInstances         int       // maximum number of instances the pool will bring up (0 = unlimited)
+       instancesWithinQuota int       // max concurrency achieved since last quota error (0 = no quota error yet)
 
        mContainersAllocatedNotStarted   prometheus.Gauge
        mContainersNotAllocatedOverQuota prometheus.Gauge
        mLongestWaitTimeSinceQueue       prometheus.Gauge
+       mLast503Time                     prometheus.Gauge
+       mMaxContainerConcurrency         prometheus.Gauge
+
+       lastQueue atomic.Value // stores a []QueueEnt
 }
 
 // New returns a new unstarted Scheduler.
 //
 // Any given queue and pool should not be used by more than one
 // scheduler at a time.
-func New(ctx context.Context, client *arvados.Client, queue ContainerQueue, pool WorkerPool, reg *prometheus.Registry, staleLockTimeout, queueUpdateInterval time.Duration) *Scheduler {
+func New(ctx context.Context, client *arvados.Client, queue ContainerQueue, pool WorkerPool, reg *prometheus.Registry, staleLockTimeout, queueUpdateInterval time.Duration, minQuota, maxInstances int, supervisorFraction float64) *Scheduler {
        sch := &Scheduler{
                logger:              ctxlog.FromContext(ctx),
                client:              client,
@@ -71,6 +79,13 @@ func New(ctx context.Context, client *arvados.Client, queue ContainerQueue, pool
                stop:                make(chan struct{}),
                stopped:             make(chan struct{}),
                uuidOp:              map[string]string{},
+               supervisorFraction:  supervisorFraction,
+               maxInstances:        maxInstances,
+       }
+       if minQuota > 0 {
+               sch.maxConcurrency = minQuota
+       } else {
+               sch.maxConcurrency = maxInstances
        }
        sch.registerMetrics(reg)
        return sch
@@ -101,6 +116,32 @@ func (sch *Scheduler) registerMetrics(reg *prometheus.Registry) {
                Help:      "Current longest wait time of any container since queuing, and before the start of crunch-run.",
        })
        reg.MustRegister(sch.mLongestWaitTimeSinceQueue)
+       sch.mLast503Time = prometheus.NewGauge(prometheus.GaugeOpts{
+               Namespace: "arvados",
+               Subsystem: "dispatchcloud",
+               Name:      "last_503_time",
+               Help:      "Time of most recent 503 error received from API.",
+       })
+       reg.MustRegister(sch.mLast503Time)
+       sch.mMaxContainerConcurrency = prometheus.NewGauge(prometheus.GaugeOpts{
+               Namespace: "arvados",
+               Subsystem: "dispatchcloud",
+               Name:      "max_concurrent_containers",
+               Help:      "Dynamically assigned limit on number of containers scheduled concurrency, set after receiving 503 errors from API.",
+       })
+       reg.MustRegister(sch.mMaxContainerConcurrency)
+       reg.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+               Namespace: "arvados",
+               Subsystem: "dispatchcloud",
+               Name:      "at_quota",
+               Help:      "Flag indicating the cloud driver is reporting an at-quota condition.",
+       }, func() float64 {
+               if sch.pool.AtQuota() {
+                       return 1
+               } else {
+                       return 0
+               }
+       }))
 }
 
 func (sch *Scheduler) updateMetrics() {
@@ -154,14 +195,23 @@ func (sch *Scheduler) run() {
        }
 
        // Keep the queue up to date.
-       poll := time.NewTicker(sch.queueUpdateInterval)
-       defer poll.Stop()
        go func() {
-               for range poll.C {
+               for {
+                       starttime := time.Now()
                        err := sch.queue.Update()
                        if err != nil {
                                sch.logger.Errorf("error updating queue: %s", err)
                        }
+                       // If the previous update took a long time,
+                       // that probably means the server is
+                       // overloaded, so wait that long before doing
+                       // another. Otherwise, wait for the configured
+                       // poll interval.
+                       delay := time.Since(starttime)
+                       if delay < sch.queueUpdateInterval {
+                               delay = sch.queueUpdateInterval
+                       }
+                       time.Sleep(delay)
                }
        }()