From c01221e70fc71f5bcff5f699095fbcbfc1e2b4a3 Mon Sep 17 00:00:00 2001 From: zeripath Date: Wed, 29 Jan 2020 01:01:06 +0000 Subject: Queue: Make WorkerPools and Queues flushable (#10001) * Make WorkerPools and Queues flushable Adds Flush methods to Queues and the WorkerPool Further abstracts the WorkerPool Adds a final step to Flush the queues in the defer from PrintCurrentTest Fixes an issue with Settings inheritance in queues Signed-off-by: Andrew Thornton * Change to for loop * Add IsEmpty and begin just making the queues composed WorkerPools * subsume workerpool into the queues and create a flushable interface * Add manager command * Move flushall to queue.Manager and add to testlogger * As per @guillep2k * as per @guillep2k * Just make queues all implement flushable and clean up the wrapped queue flushes * cope with no timeout Co-authored-by: Lauris BH --- routers/admin/admin.go | 30 ++++++++++++++++++++++++++---- 1 file changed, 26 insertions(+), 4 deletions(-) (limited to 'routers/admin') diff --git a/routers/admin/admin.go b/routers/admin/admin.go index 71a22e1f9e..56e5d321b3 100644 --- a/routers/admin/admin.go +++ b/routers/admin/admin.go @@ -404,6 +404,28 @@ func WorkerCancel(ctx *context.Context) { }) } +// Flush flushes a queue +func Flush(ctx *context.Context) { + qid := ctx.ParamsInt64("qid") + mq := queue.GetManager().GetManagedQueue(qid) + if mq == nil { + ctx.Status(404) + return + } + timeout, err := time.ParseDuration(ctx.Query("timeout")) + if err != nil { + timeout = -1 + } + ctx.Flash.Info(ctx.Tr("admin.monitor.queue.pool.flush.added", mq.Name)) + go func() { + err := mq.Flush(timeout) + if err != nil { + log.Error("Flushing failure for %s: Error %v", mq.Name, err) + } + }() + ctx.Redirect(setting.AppSubURL + fmt.Sprintf("/admin/monitor/queue/%d", qid)) +} + // AddWorkers adds workers to a worker group func AddWorkers(ctx *context.Context) { qid := ctx.ParamsInt64("qid") @@ -424,7 +446,7 @@ func AddWorkers(ctx *context.Context) { ctx.Redirect(setting.AppSubURL + fmt.Sprintf("/admin/monitor/queue/%d", qid)) return } - if mq.Pool == nil { + if _, ok := mq.Managed.(queue.ManagedPool); !ok { ctx.Flash.Error(ctx.Tr("admin.monitor.queue.pool.none")) ctx.Redirect(setting.AppSubURL + fmt.Sprintf("/admin/monitor/queue/%d", qid)) return @@ -442,7 +464,7 @@ func SetQueueSettings(ctx *context.Context) { ctx.Status(404) return } - if mq.Pool == nil { + if _, ok := mq.Managed.(queue.ManagedPool); !ok { ctx.Flash.Error(ctx.Tr("admin.monitor.queue.pool.none")) ctx.Redirect(setting.AppSubURL + fmt.Sprintf("/admin/monitor/queue/%d", qid)) return @@ -488,10 +510,10 @@ func SetQueueSettings(ctx *context.Context) { return } } else { - timeout = mq.Pool.BoostTimeout() + timeout = mq.BoostTimeout() } - mq.SetSettings(maxNumber, number, timeout) + mq.SetPoolSettings(maxNumber, number, timeout) ctx.Flash.Success(ctx.Tr("admin.monitor.queue.settings.changed")) ctx.Redirect(setting.AppSubURL + fmt.Sprintf("/admin/monitor/queue/%d", qid)) } -- cgit v1.2.3