Further limit TaskTags aggregation cron job to avoid OOM

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
 
@@ skipping 126 matching lines @@
     count = 0
     q = task_result.get_result_summaries_query(
         now - datetime.timedelta(hours=12), None, 'created_ts', 'all', None)
     cursor = None
     while True:
       tasks, cursor = datastore_utils.fetch_page(q, 1000, cursor)
       count += len(tasks)
       for t in tasks:
         for i in t.tags:
           k, v = i.split(':', 1)
-          if k != 'cron_invocation_id':
-            seen.setdefault(k, set()).add(v)
+          s = seen.setdefault(k, set())
+          if s is not None:
+            s.add(v)
+            # 128 is arbitrary large number to avoid OOM
+            if len(s) >= 128:
+              logging.info('Stripping tag %s because there are too many', k)
+              seen[k] = None
       if not cursor or len(tasks) == 0:
         break
 
     tags = [
       task_result.TagValues(tag=k, values=sorted(values))
-      for k, values in sorted(seen.iteritems())
+      for k, values in sorted(seen.iteritems()) if values is not None
     ]
 
     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):
@@ skipping 83 matching lines @@
     ('/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]