Upload basic info about completed swarming tasks to event_mon

appengine/swarming/server/task_scheduler.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.
 
 """High level tasks execution scheduling API.
 
 This is the interface closest to the HTTP handlers.
 """
 
 import datetime
 import logging
 import math
 import random
 
 from google.appengine.ext import ndb
 
 from components import auth
 from components import datastore_utils
 from components import pubsub
 from components import utils
 
+import event_mon_metrics
 import ts_mon_metrics
 
 from server import acl
 from server import config
 from server import stats
 from server import task_pack
 from server import task_request
 from server import task_result
 from server import task_to_run
 
@@ skipping 773 matching lines @@
   assert bool(error) != bool(run_result), (error, run_result)
   if error:
     logging.error('Task %s %s', packed, error)
     return None
   # Caller must retry if PubSub enqueue fails.
   task_completed = run_result.state != task_result.State.RUNNING
   if not _maybe_pubsub_notify_now(smry, request):
     return None
   _update_stats(run_result, bot_id, request, task_completed)
   if task_completed:
+    event_mon_metrics.send_task_event(smry)

[M-A Ruel, 2016/11/28 16:56:32]

I'm concerned in adding more synchronous I/O in the hot path. I was concerned
about ts_mon_metrics too and it did cause a fair amount of failure once deployed
to prod.

I'm fine with testing on prod and looking at the failure rate.

[Paweł Hajdan Jr., 2016/11/29 14:37:33]

I'm also concerned. How about proactively requesting async interfaces for both
monitoring pipelines?

[M-A Ruel, 2016/11/29 14:48:40]

As long as it is quick to disable the functionality by adding an early return in
send_task_event(), it's not worth overdesigning it.

     ts_mon_metrics.update_jobs_completed_metrics(smry)
   return run_result.state
 
 
 def bot_kill_task(run_result_key, bot_id):
   """Terminates a task that is currently running as an internal failure.
 
   Returns:
     str if an error message.
   """
@@ skipping 172 matching lines @@
 ## Task queue tasks.
 
 
 def task_handle_pubsub_task(payload):
   """Handles task enqueued by _maybe_pubsub_notify_via_tq."""
   # Do not catch errors to trigger task queue task retry. Errors should not
   # happen in normal case.
   _pubsub_notify(
       payload['task_id'], payload['topic'],
       payload['auth_token'], payload['userdata'])