Create Tasks.Tags endpoint and Cron job

appengine/swarming/handlers_backend.py

 # Copyright 2014 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.
 
 """Main entry point for Swarming backend handlers."""
 
+import datetime
 import json
 import logging
 
 import webapp2
 from google.appengine.api import app_identity
 from google.appengine.api import datastore_errors
 from google.appengine.api import taskqueue
 
 from components import utils
 
 import mapreduce_jobs
 from components import decorators
+from components import datastore_utils
 from components import machine_provider
 from server import bot_management
 from server import config
 from server import lease_management
 from server import stats
+from server import task_result
 from server import task_scheduler
 
 
 class CronBotDiedHandler(webapp2.RequestHandler):
   @decorators.require_cronjob
   def get(self):
     try:
       task_scheduler.cron_handle_bot_died(self.request.host_url)
     except datastore_errors.NeedIndexError as e:
       # When a fresh new instance is deployed, it takes a few minutes for the
@@ skipping 83 matching lines @@
       for k, values in sorted(seen.iteritems())
     ]
 
     logging.info('Saw dimensions %s', dims)
     bot_management.DimensionAggregation(
         key=bot_management.DimensionAggregation.KEY,
         dimensions=dims,
         ts=now).put()
 
 
+class CronTasksTagsAggregationHandler(webapp2.RequestHandler):
+  """Aggregates all task tags from the last 48 hours."""
+
+  @decorators.require_cronjob
+  def get(self):
+    seen = {}
+    now = utils.utcnow()
+    count = 0
+    q = task_result.get_result_summaries_query(
+        now - datetime.timedelta(days=2), None, 'created_ts', 'all', None)
+    cursor = None
+    while True:
+      tasks, cursor = datastore_utils.fetch_page(q, 1000, cursor)

[M-A Ruel, 2016/08/16 15:37:21]

fetching 1000 at a time is going to be very slow. If you iterate one at a time,
the ndb.Query object will actually do batches, so I recommend just doing that.
Then no need to use a cursor.

[kjlubick, 2016/08/16 17:20:44]

I tried iterating one at a time, even with a filter and it timed out after 1
minute. 

I will have it only do the last 12 hours, until we see how it does under load.

+      count += len(tasks)
+      for t in tasks:
+        for i in t.tags:
+          k, v = i.split(':', 1)
+          if k != 'cron_invocation_id':

[M-A Ruel, 2016/08/16 15:37:21]

what's that?

[kjlubick, 2016/08/16 17:20:44]

It appeared to be something that was different for all, or at least many of the
heartbeat tasks.  I figured we need not aggregate on it, right?

+            seen.setdefault(k, set()).add(v)
+      if not cursor or len(tasks) == 0:
+        break
+
+    tags = [
+      task_result.TagValues(tag=k,values=sorted(values))

[M-A Ruel, 2016/08/16 15:37:21]

task_result.TagValues(tag=k, values=sorted(values))

[kjlubick, 2016/08/16 17:20:44]

Done.

+      for k, values in sorted(seen.iteritems())
+    ]
+
+    logging.info('From %d tasks, saw tags %s', count, tags)
+    task_result.TagAggregation(
+        key=task_result.TagAggregation.KEY,
+        tags=tags,
+        ts=now).put()
+
+
 class CronMachineProviderPubSubHandler(webapp2.RequestHandler):
   """Listens for Pub/Sub communication from Machine Provider."""
 
   @decorators.require_cronjob
   def get(self):
     if not config.settings().mp.enabled:
       logging.info('MP support is disabled')
       return
 
     taskqueue.add(
@@ skipping 55 matching lines @@
     # the app version has to be running an already compatible version already.
     ('/internal/cron/abort_bot_died', CronBotDiedHandler),
     ('/internal/cron/handle_bot_died', CronBotDiedHandler),
     ('/internal/cron/abort_expired_task_to_run',
         CronAbortExpiredShardToRunHandler),
 
     ('/internal/cron/stats/update', stats.InternalStatsUpdateHandler),
     ('/internal/cron/trigger_cleanup_data', CronTriggerCleanupDataHandler),
     ('/internal/cron/aggregate_bots_dimensions',
         CronBotsDimensionAggregationHandler),
+    ('/internal/cron/aggregate_tasks_tags',
+        CronTasksTagsAggregationHandler),
     ('/internal/cron/machine_provider', CronMachineProviderBotHandler),
     ('/internal/cron/machine_provider_cleanup',
         CronMachineProviderCleanUpHandler),
     ('/internal/cron/machine_provider_pubsub',
         CronMachineProviderPubSubHandler),
 
     # Task queues.
     ('/internal/taskqueue/cleanup_data', TaskCleanupDataHandler),
     (r'/internal/taskqueue/pubsub/<task_id:[0-9a-f]+>', TaskSendPubSubMessage),
     ('/internal/taskqueue/pubsub/machine_provider',
         TaskMachineProviderPubSubHandler),
 
     # Mapreduce related urls.
     (r'/internal/taskqueue/mapreduce/launch/<job_id:[^\/]+>',
       InternalLaunchMapReduceJobWorkerHandler),
   ]
   return [webapp2.Route(*a) for a in routes]