Parse json events into data dog events
Testing
ubuntu@ip-10-251-55-110:~/dd_udpstream$ export PYTHONPATH=$PYTHONPATH:/opt/datadog-agent/agent/
ubuntu@ip-10-251-55-110:~/dd_udpstream$ python dd_udpstream_test.py
# Use this custom json log parser to send events to datadog | |
from datetime import datetime | |
import json | |
#from dogstream import common | |
DEFAULT_ALERT = 'normal' | |
def parse_json(log, line): | |
event = {} | |
## XXX - add an exception handler and report the error | |
event_type, msg = log.strip().split('\t') | |
json_event = json.loads(msg) | |
event.update(json_event) | |
event['event_type'] = event_type | |
if not event.has_key('palert_ype'): | |
event['alert_type'] = DEFAULT_ALERT | |
return event | |
import unittest | |
import dd_udpstream | |
class TestStream(unittest.TestCase): | |
""" Parsing tests """ | |
def test_basic(self): | |
log = "category\t{\"wat\": \"this\"} " | |
res = dd_udpstream.parse_json(log, 1) | |
self.assertEqual('this', res.get('wat')) | |
self.assertEqual('category', res.get('event_type')) | |
self.assertEqual('normal', res.get('alert_type')) | |
if __name__ == '__main__': | |
unittest.main() |
from udplog import udplog | |
logger = udplog.UDPLogger() | |
logger.log('test', {u'message': u'test'}) |