index traps data into Elasticsearch

bin/trapperkeeper

@@ -15,7 +15,7 @@ from trapperkeeper.config import Config
 from trapperkeeper.models import get_db_engine, Session
 from trapperkeeper.utils import get_template_env, get_loglevel, CachingResolver
 from trapperkeeper import __version__
-
+from trapperkeeper.es_client import get_elasticsearch_client
 
 def stats_server(port):
     class Stats(tornado.web.RequestHandler):
@@ -49,6 +49,11 @@ def main():
     config = Config.from_file(args.config)
 
     db_engine = get_db_engine(config["database"])
+    es_host = config.get("elasticsearch_host")
+    if es_host:
+        es_client = get_elasticsearch_client(es_host)
+    else:
+        es_client = None
     community = config["community"]
     if not community:
         community = None
@@ -62,7 +67,7 @@ def main():
     template_env = get_template_env(
         hostname_or_ip=resolver.hostname_or_ip
     )
-    cb = TrapperCallback(conn, template_env, config, resolver, community)
+    cb = TrapperCallback(conn, template_env, config, resolver, community, es_client)
 
     logging.basicConfig(
         level=get_loglevel(args),

conf/trapperkeeper.yaml

@@ -121,3 +121,15 @@ config:
     #
     # Type: bool
     ipv6: false
+
+    # Elastic Search Host
+    #
+    # Type: string
+    elasticsearch_host: "http://localhost:9200"
+
+    # Elastic Search Index
+    # Since starting from ES ver 6.0, there is no type,
+    # index name will be used as type name for ES ver < 6.0
+    #
+    # Type: string
+    elasticsearch_index: "trap"

requirements.txt

@@ -12,3 +12,4 @@ pysnmp==4.2.5
 pytz==2014.1
 tornado==3.2
 wsgiref==0.1.2
+elasticsearch==6.1.1

trapperkeeper/callbacks.py

@@ -15,6 +15,7 @@
 from trapperkeeper.constants import SNMP_VERSIONS
 from trapperkeeper.models import Notification
 from trapperkeeper.utils import parse_time_string, send_trap_email
+from trapperkeeper.es_client import index_trap_to_elasticsearch
 
 
 try:
@@ -26,13 +27,14 @@ def dde_run(notification):
 
 
 class TrapperCallback(object):
-    def __init__(self, conn, template_env, config, resolver, community):
+    def __init__(self, conn, template_env, config, resolver, community, es_client):
         self.conn = conn
         self.template_env = template_env
         self.config = config
         self.hostname = socket.gethostname()
         self.resolver = resolver
         self.community = community
+        self.es_client = es_client
 
     def __call__(self, *args, **kwargs):
         try:
@@ -164,3 +166,9 @@ def _call(self, transport_dispatcher, transport_domain, transport_address, whole
             logging.debug(err)
 
         self._send_mail(handler, trap, duplicate)
+
+        if self.es_client is not None:
+            index_trap_to_elasticsearch(self.es_client, self.config.get("elasticsearch_index"), trap.es_doc)
+
+
+

trapperkeeper/es_client.py

@@ -0,0 +1,9 @@
+from elasticsearch import Elasticsearch
+
+
+def get_elasticsearch_client(host):
+    return Elasticsearch(host)
+
+
+def index_trap_to_elasticsearch(es, index_name, doc):
+    return es.index(index=index_name, doc_type=index_name, body=doc)

trapperkeeper/models.py

@@ -76,6 +76,24 @@ def pprint(self):
         for varbind in self.varbinds:
             varbind.pprint()
 
+    # for ElasticSearch index
+    @property
+    def es_doc(self):
+        doc = {
+            "notification_id": self.id,
+            "host": self.host,
+            "oid": self.oid,
+            "mib_name": ObjectId(self.oid).name,
+            "severity": self.severity,
+            "@timestamp": utcnow()
+        }
+
+        for varbind in self.varbinds:
+            doc.update(varbind.es_doc)
+
+        return doc
+
+
     @staticmethod
     def _from_pdu_v1(host, proto_module, version, pdu):
         trapoid = str(proto_module.apiTrapPDU.getEnterprise(pdu))
@@ -177,6 +195,17 @@ def to_dict(self, pretty=False):
     def pprint(self):
         print "\t", self.oid, "(%s)" % self.value_type, "=", self.value
 
+    @property
+    def es_doc(self):
+        d = self.to_dict(pretty=True)
+        doc = dict()
+        doc.update({
+            d['name']: d['pretty_value'],
+            d['oid']: d['value'],
+            "%s:type" % (d['name']): d['value_type']
+        })
+        return doc
+
     def __repr__(self):
         return "Varbind(oid=%s, value_type=%s, value=%s)" % (
             repr(self.oid), repr(self.value_type), repr(self.value)

trapperkeeper/utils.py

@@ -13,6 +13,8 @@
 from email.mime.text import MIMEText
 
 
+
+
 _TIME_STRING_RE = re.compile(
     r"(?:(?P<days>\d+)d)?"
     r"(?:(?P<hours>\d+)h)?"
@@ -128,6 +130,7 @@ def send_trap_email(recipients, sender, subject, template_env, context):
     smtp.quit()
 
 
+
 def get_loglevel(args):
     verbose = args.verbose * 10
     quiet = args.quiet * 10