X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/60bbffb69b908f7a1146fc0c1904aae2775cdf1e..96d8b9e1afecccae803ec4b956ada745dbe71d9f:/lib/dispatchcloud/test/queue.go diff --git a/lib/dispatchcloud/test/queue.go b/lib/dispatchcloud/test/queue.go index 74b84122f2..2be8246bd6 100644 --- a/lib/dispatchcloud/test/queue.go +++ b/lib/dispatchcloud/test/queue.go @@ -24,15 +24,32 @@ type Queue struct { // must not be nil. ChooseType func(*arvados.Container) (arvados.InstanceType, error) + // Mimic railsapi implementation of MaxDispatchAttempts config + MaxDispatchAttempts int + Logger logrus.FieldLogger - entries map[string]container.QueueEnt - updTime time.Time - subscribers map[<-chan struct{}]chan struct{} + entries map[string]container.QueueEnt + updTime time.Time + subscribers map[<-chan struct{}]chan struct{} + stateChanges []QueueStateChange mtx sync.Mutex } +type QueueStateChange struct { + UUID string + From arvados.ContainerState + To arvados.ContainerState +} + +// All calls to Lock/Unlock/Cancel to date. +func (q *Queue) StateChanges() []QueueStateChange { + q.mtx.Lock() + defer q.mtx.Unlock() + return q.stateChanges +} + // Entries returns the containers that were queued when Update was // last called. func (q *Queue) Entries() (map[string]container.QueueEnt, time.Time) { @@ -111,6 +128,7 @@ func (q *Queue) notify() { // caller must have lock. func (q *Queue) changeState(uuid string, from, to arvados.ContainerState) error { ent := q.entries[uuid] + q.stateChanges = append(q.stateChanges, QueueStateChange{uuid, from, to}) if ent.Container.State != from { return fmt.Errorf("changeState failed: state=%q", ent.Container.State) } @@ -118,7 +136,15 @@ func (q *Queue) changeState(uuid string, from, to arvados.ContainerState) error q.entries[uuid] = ent for i, ctr := range q.Containers { if ctr.UUID == uuid { - q.Containers[i].State = to + if max := q.MaxDispatchAttempts; max > 0 && ctr.LockCount >= max && to == arvados.ContainerStateQueued { + q.Containers[i].State = arvados.ContainerStateCancelled + q.Containers[i].RuntimeStatus = map[string]interface{}{"error": fmt.Sprintf("Failed to start: lock_count == %d", ctr.LockCount)} + } else { + q.Containers[i].State = to + if to == arvados.ContainerStateLocked { + q.Containers[i].LockCount++ + } + } break } } @@ -142,9 +168,11 @@ func (q *Queue) Update() error { upd[ctr.UUID] = ent } else { it, _ := q.ChooseType(&ctr) + ctr.Mounts = nil upd[ctr.UUID] = container.QueueEnt{ Container: ctr, InstanceType: it, + FirstSeenAt: time.Now(), } } } @@ -172,12 +200,11 @@ func (q *Queue) Notify(upd arvados.Container) bool { if allowContainerUpdate[ctr.State][upd.State] { q.Containers[i] = upd return true - } else { - if q.Logger != nil { - q.Logger.WithField("ContainerUUID", ctr.UUID).Infof("test.Queue rejected update from %s to %s", ctr.State, upd.State) - } - return false } + if q.Logger != nil { + q.Logger.WithField("ContainerUUID", ctr.UUID).Infof("test.Queue rejected update from %s to %s", ctr.State, upd.State) + } + return false } } q.Containers = append(q.Containers, upd) @@ -185,18 +212,18 @@ func (q *Queue) Notify(upd arvados.Container) bool { } var allowContainerUpdate = map[arvados.ContainerState]map[arvados.ContainerState]bool{ - arvados.ContainerStateQueued: map[arvados.ContainerState]bool{ + arvados.ContainerStateQueued: { arvados.ContainerStateQueued: true, arvados.ContainerStateLocked: true, arvados.ContainerStateCancelled: true, }, - arvados.ContainerStateLocked: map[arvados.ContainerState]bool{ + arvados.ContainerStateLocked: { arvados.ContainerStateQueued: true, arvados.ContainerStateLocked: true, arvados.ContainerStateRunning: true, arvados.ContainerStateCancelled: true, }, - arvados.ContainerStateRunning: map[arvados.ContainerState]bool{ + arvados.ContainerStateRunning: { arvados.ContainerStateRunning: true, arvados.ContainerStateCancelled: true, arvados.ContainerStateComplete: true,