Add prometheus metrics for the jobs, update tests

Signed-off-by: Valery Piashchynski <[email protected]>

1 files changed, 6 insertions, 1 deletions

diff --git a/plugins/jobs/drivers/amqp/consumer.go b/plugins/jobs/drivers/amqp/consumer.go
index f3b40ae3..95df02ec 100644
--- a/plugins/jobs/drivers/amqp/consumer.go
+++ b/plugins/jobs/drivers/amqp/consumer.go
@@ -307,7 +307,8 @@ func (j *JobConsumer) State(ctx context.Context) (*jobState.State, error) {
 			Driver:   pipe.Driver(),
 			Queue:    q.Name,
 			Active:   int64(q.Messages),
-			Delayed:  *j.delayed,
+			Delayed:  atomic.LoadInt64(j.delayed),
+			Ready:    ready(atomic.LoadUint32(&j.listeners)),
 		}, nil
 
 	case <-ctx.Done():
@@ -501,3 +502,7 @@ func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
 		return errors.E(op, errors.TimeOut, ctx.Err())
 	}
 }
+
+func ready(r uint32) bool {
+	return r > 0
+}