swarming: add active jobs pending times metric

appengine/swarming/ts_mon_metrics.py

 # Copyright 2016 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """Timeseries metrics."""
 
 from collections import defaultdict
 import itertools
 
 from google.appengine.ext import ndb
@@ skipping 28 matching lines @@
 jobs_completed = gae_ts_mon.CounterMetric(
     'jobs/completed',
     description='Number of completed jobs.')
 
 
 jobs_durations = gae_ts_mon.CumulativeDistributionMetric(
     'jobs/durations', bucketer=_bucketer,
     description='Cycle times of completed jobs, in seconds.')
 
 
+jobs_pending_durations = gae_ts_mon.NonCumulativeDistributionMetric(
+    'jobs/pending_durations', bucketer=_bucketer,
+    description='Pending times of active jobs, in seconds.')
+
+
 # Swarming-specific metric. Metric fields:
 # - project_id: e.g. 'chromium'
 # - subproject_id: e.g. 'blink'. Set to empty string if not used.
 # - spec_name: name of a job specification, e.g. '<master>:<builder>:<test>'
 #     for buildbot jobs.
 tasks_expired = gae_ts_mon.CounterMetric(
     'swarming/tasks/expired',
     description='Number of expired tasks')
 
 # Global metric. Metric fields:
@@ skipping 75 matching lines @@
     fields['result'] = 'infra-failure'
   elif task_result_summary.failure:
     fields['result'] = 'failure'
   else:
     fields['result'] = 'success'
   jobs_completed.increment(fields=fields)
   jobs_durations.add(task_result_summary.duration, fields=fields)
 
 
 @ndb.tasklet
-def _set_jobs_metrics():
+def _set_jobs_metrics(now):
   state_map = {task_result.State.RUNNING: 'running',
                task_result.State.PENDING: 'pending'}
   query_iter = task_result.get_result_summaries_query(
       None, None, 'created_ts', 'pending_running', None).iter()
   jobs_counts = defaultdict(lambda: 0)
+  jobs_pending_distributions = defaultdict(
+      lambda: gae_ts_mon.Distribution(_bucketer))
   while (yield query_iter.has_next_async()):
     summary = query_iter.next()
     status = state_map.get(summary.state, '')
     fields = extract_job_fields(summary.tags)
     target_fields = dict(TARGET_FIELDS)
     if summary.bot_id:
       target_fields['hostname'] = 'autogen:' + summary.bot_id
     if summary.bot_id and status == 'running':
       jobs_running.set(True, target_fields=target_fields, fields=fields)
     fields['status'] = status
-    jobs_counts[tuple(sorted(fields.iteritems()))] += 1
+
+    key = tuple(sorted(fields.iteritems()))
+
+    jobs_counts[key] += 1
+
+    pending_duration = summary.pending_now(now)
+    if pending_duration:
+      jobs_pending_distributions[key].add(pending_duration)

[Sergey Berezin, 2016/07/06 18:31:47]

pending_durations.total_seconds()

[Paweł Hajdan Jr., 2016/07/08 12:03:27]

Done (the test also detected this).

 
   for key, count in jobs_counts.iteritems():
     jobs_active.set(count, target_fields=TARGET_FIELDS, fields=dict(key))
 
+  for key, distribution in jobs_pending_distributions.iteritems():
+    jobs_pending_durations.set(
+        distribution, target_fields=TARGET_FIELDS, fields=dict(key))
+
 
 @ndb.tasklet
 def _set_executors_metrics(now):
   query_iter = bot_management.BotInfo.query().iter()
   while (yield query_iter.has_next_async()):
     bot_info = query_iter.next()
     status = 'ready'
     if bot_info.task_id:
       status = 'running'
     elif bot_info.quarantined:
       status = 'quarantined'
     elif bot_info.is_dead(now):
       status = 'dead'
 
     target_fields = dict(TARGET_FIELDS)
     target_fields['hostname'] = 'autogen:' + bot_info.id
 
     executors_status.set(status, target_fields=target_fields)
     executors_pool.set(
         pool_from_dimensions(bot_info.dimensions),
         target_fields=target_fields)
 
 
 @ndb.tasklet
 def _set_global_metrics_async(now):
-  yield _set_executors_metrics(now), _set_jobs_metrics()
+  yield _set_executors_metrics(now), _set_jobs_metrics(now)
 
 
 def _set_global_metrics(now=None):
   if now is None:
     now = utils.utcnow()
   _set_global_metrics_async(now).get_result()
 
 
 def initialize():
   gae_ts_mon.register_global_metrics(
       [jobs_running, executors_pool, executors_status])
   gae_ts_mon.register_global_metrics_callback('callback', _set_global_metrics)