Monday, June 14, 2021

Datadog event - function to capture notebook exception

This function can be used at any "except Exception" like below. e.g. 

try:
## any code logic

except Exception as e:
context_str = dbutils.notebook.entry_point.getDbutils().notebook().getContext().toJson()
DatabricksNBRuntimeError(context_str)

Functions are:

from datadog import initialize, api

api_key = dbutils.secrets.get(scope="datadog", key="datadog_api_key")
app_key = dbutils.secrets.get(scope="datadog", key="datadog_app_key")

options = {
"api_key": f"{api_key}",
"app_key": f"{app_key}"
}

initialize(**options)
import sys
import requests
import traceback
import json

def datadog_event(notebook_path, job_name, user):
exc_type, exc_value, exc_tb = sys.exc_info()
traceback_error=str(traceback.format_exception(exc_type, exc_value, exc_tb))
error_case_list = [find_between(traceback_error,"Exception","\n\tat"), find_between(traceback_error,"Error","\n\tat"), traceback_error]
error_message = next(error for error in error_case_list if error != '')

error_dict = {"notebook_metadata": {"job_name":job_name, "notebook_path": notebook_path,"user": user}, "error": error_message}
error_message = json.dumps(error_dict)

title = datadog_identify_string
if job_name:
title = datadog_identify_string + ': job( ' + job_name + ' )'

text = error_message

print(title, text)
api.Event.create(title=title, text=text, tags=datadog_tags, priority='Low')
def exit_notebook(job_name, user):

error_case_list = [find_between(str(e),"Exception","\n\tat"), find_between(str(e),"Error","\n\tat"), str(e)[:200]]
error_message = next(error for error in error_case_list if error != '')
output = {"status": "FAIL", "notebook_metadata": {"job_name":job_name, "user":user}, "message": error_message}

dbutils.notebook.exit(json.dumps(output))
class DatabricksNBRuntimeError(Exception):

def __init__(self, context_str, **kwargs):

self.strerror = args
self.args = args

notebook_metadata = json.loads(context_str)

job_name = None
user = None
notebook_path = notebook_metadata['extraContext']['notebook_path']
if 'jobName' in notebook_metadata['tags'].keys():
job_name = notebook_metadata['tags']['jobName']
if 'user' in notebook_metadata['tags'].keys():
user = notebook_metadata['tags']['user']

datadog_event(notebook_path, job_name, user)
exit_notebook(job_name, user)

No comments:

Post a Comment