blob: b3a2ffadebaf6b733f2010fca459b9000fb3773f [file] [log] [blame]
# Copyright (c) 2012 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Breakpad for Python.
Sends a notification when a process stops on an exception.
It is only enabled when all these conditions are met:
1. hostname finishes with '.google.com' or 'chromium.org'
2. main module name doesn't contain the word 'test'
3. no NO_BREAKPAD environment variable is defined
"""
import atexit
import getpass
import os
import socket
import sys
import time
import traceback
import urllib
import urllib2
# Configure these values.
DEFAULT_URL = 'https://chromium-status.appspot.com'
# Global variable to prevent double registration.
_REGISTERED = False
_TIME_STARTED = time.time()
_HOST_NAME = socket.getfqdn()
# Skip unit tests and we don't want anything from non-googler.
IS_ENABLED = (
not 'test' in getattr(sys.modules['__main__'], '__file__', '') and
not 'NO_BREAKPAD' in os.environ and
_HOST_NAME.endswith(('.google.com', '.chromium.org')))
# TODO: COBALT: Disable breakpad to avoid spamming the Chromium team.
# Eventually this should be re-enabled and the exceptions should be forwarded to
# our own server.
IS_ENABLED = False
def post(url, params):
"""HTTP POST with timeout when it's supported."""
if not IS_ENABLED:
# Make sure to not send anything for non googler.
return
kwargs = {}
if (sys.version_info[0] * 10 + sys.version_info[1]) >= 26:
kwargs['timeout'] = 4
try:
request = urllib2.urlopen(url, urllib.urlencode(params), **kwargs)
out = request.read()
request.close()
return out
except IOError:
return 'There was a failure while trying to send the stack trace. Too bad.'
def FormatException(e):
"""Returns a human readable form of an exception.
Adds the maximum number of interesting information in the safest way."""
try:
out = repr(e)
except Exception:
out = ''
try:
out = str(e)
if isinstance(e, Exception):
# urllib exceptions, usually the HTTP headers.
if hasattr(e, 'headers'):
out += '\nHeaders: %s' % e.headers
if hasattr(e, 'url'):
out += '\nUrl: %s' % e.url
if hasattr(e, 'msg'):
out += '\nMsg: %s' % e.msg
# The web page in some urllib exceptions.
if hasattr(e, 'read') and callable(e.read):
out += '\nread(): %s' % e.read()
if hasattr(e, 'info') and callable(e.info):
out += '\ninfo(): %s' % e.info()
except Exception:
pass
return out
def SendStack(last_tb, stack, url=None, maxlen=50, verbose=True):
"""Sends the stack trace to the breakpad server."""
if not IS_ENABLED:
return
def p(o):
if verbose:
print(o)
p('Sending crash report ...')
params = {
'args': sys.argv,
'cwd': os.getcwd(),
'exception': FormatException(last_tb),
'host': _HOST_NAME,
'stack': stack[0:4096],
'user': getpass.getuser(),
'version': sys.version,
}
p('\n'.join(' %s: %s' % (k, params[k][0:maxlen]) for k in sorted(params)))
p(post(url or DEFAULT_URL + '/breakpad', params))
def SendProfiling(duration, url=None):
params = {
'argv': ' '.join(sys.argv),
# Strip the hostname.
'domain': _HOST_NAME.split('.', 1)[-1],
'duration': duration,
'platform': sys.platform,
}
post(url or DEFAULT_URL + '/profiling', params)
def CheckForException():
"""Runs at exit. Look if there was an exception active."""
last_value = getattr(sys, 'last_value', None)
if last_value:
if not isinstance(last_value, KeyboardInterrupt):
last_tb = getattr(sys, 'last_traceback', None)
if last_tb:
SendStack(last_value, ''.join(traceback.format_tb(last_tb)))
else:
duration = time.time() - _TIME_STARTED
if duration > 90:
SendProfiling(duration)
def Register():
"""Registers the callback at exit. Calling it multiple times is no-op."""
global _REGISTERED
if _REGISTERED:
return
_REGISTERED = True
atexit.register(CheckForException)
if IS_ENABLED:
Register()
# Uncomment this line if you want to test it out.
#Register()