events api and feed sync event examples

Signed-off-by: Swathi Gangisetty <swathi@anchore.com>

.gitignore

@@ -102,3 +102,4 @@ ENV/
 
 # IDEs
 .idea/
+*.iml

anchore_engine/clients/catalog.py

@@ -7,6 +7,7 @@
 import anchore_engine.services.common
 import anchore_engine.clients.common
 from anchore_engine.subsys import logger
+from anchore_engine.subsys.events import Event
 
 localconfig = None
 headers = {'Content-Type': 'application/json'}
@@ -754,7 +755,10 @@ def delete_registry(userId, registry=None):
 
     return(ret)
 
-def add_event(userId, hostId, service_name, level, message, detail=None):
+def add_event(userId, event):
+    if not isinstance(event, Event):
+        raise TypeError('Invalid event definition')
+
     global localconfig, headers
     if localconfig == None:
         localconfig = anchore_engine.configuration.localconfig.get_config()
@@ -769,20 +773,12 @@ def add_event(userId, hostId, service_name, level, message, detail=None):
 
     base_url = anchore_engine.clients.common.get_service_endpoint(userId, 'catalog')
     url = base_url + "/events"
-
-    payload = {
-        'hostId':hostId,
-        'service_name':service_name,
-        'level':level,
-        'message':message,
-        'detail':detail
-    }
 
-    ret = http.anchy_post(url, data=json.dumps(payload), auth=auth, headers=headers, verify=localconfig['internal_ssl_verify'])
+    ret = http.anchy_post(url, data=event.to_json(), auth=auth, headers=headers, verify=localconfig['internal_ssl_verify'])
 
     return(ret)
 
-def get_event(userId, hostId=None, level=None, message=None):
+def get_events(userId, source_servicename=None, source_hostid=None, resource_type=None, level=None, since=None, before=None, next=None):
     global localconfig, headers
     if localconfig == None:
         localconfig = anchore_engine.configuration.localconfig.get_config()
@@ -797,14 +793,105 @@ def get_event(userId, hostId=None, level=None, message=None):
 
     base_url = anchore_engine.clients.common.get_service_endpoint(userId, 'catalog')
     url = base_url + "/events"
-
-    payload = {
-        'hostId':hostId,
-        'level':level,
-        'message':message
-    }
 
-    ret = http.anchy_get(url, data=json.dumps(payload), auth=auth, headers=headers, verify=localconfig['internal_ssl_verify'])
+    path_params = []
+
+    if source_servicename:
+        path_params.append('source_servicename={}'.format(source_servicename))
+
+    if source_hostid:
+        path_params.append('source_hostid={}'.format(source_hostid))
+
+    if resource_type:
+        path_params.append('resource_type={}'.format(resource_type))
+
+    if level:
+        path_params.append('level={}'.format(level))
+
+    if since:
+        path_params.append('since={}'.format(since))
+
+    if before:
+        path_params.append('before={}'.format(before))
+
+    if next:
+        path_params.append('next={}'.format(since))
+
+    if path_params:
+        url = url + '?' + '&'.join(path_params)
+
+    ret = http.anchy_get(url, auth=auth, headers=headers, verify=localconfig['internal_ssl_verify'])
+
+    return(ret)
+
+def delete_events(userId, since=None, before=None):
+    global localconfig, headers
+    if localconfig == None:
+        localconfig = anchore_engine.configuration.localconfig.get_config()
+
+    ret = False
+
+    if type(userId) == tuple:
+        userId, pw = userId
+    else:
+        pw = ""
+    auth = (userId, pw)
+
+    base_url = anchore_engine.clients.common.get_service_endpoint(userId, 'catalog')
+    url = base_url + "/events"
+
+    path_params = []
+
+    if since:
+        path_params.append('since={}'.format(since))
+
+    if before:
+        path_params.append('before={}'.format(before))
+
+    if path_params:
+        url = url + '?' + '&'.join(path_params)
+
+    ret = http.anchy_delete(url, auth=auth, headers=headers, verify=localconfig['internal_ssl_verify'])
+
+    return(ret)
+
+def get_event(userId, eventId):
+    global localconfig, headers
+    if localconfig == None:
+        localconfig = anchore_engine.configuration.localconfig.get_config()
+
+    ret = False
+
+    if type(userId) == tuple:
+        userId, pw = userId
+    else:
+        pw = ""
+    auth = (userId, pw)
+
+    base_url = anchore_engine.clients.common.get_service_endpoint(userId, 'catalog')
+    url = base_url + "/events/" + eventId
+
+    ret = http.anchy_get(url, auth=auth, headers=headers, verify=localconfig['internal_ssl_verify'])
+
+    return(ret)
+
+def delete_event(userId, eventId):
+    global localconfig, headers
+    if localconfig == None:
+        localconfig = anchore_engine.configuration.localconfig.get_config()
+
+    ret = False
+
+    if type(userId) == tuple:
+        userId, pw = userId
+    else:
+        pw = ""
+    auth = (userId, pw)
+
+    base_url = anchore_engine.clients.common.get_service_endpoint(userId, 'catalog')
+    url = base_url + "/events/" + eventId
+
+    ret = http.anchy_delete(url, auth=auth, headers=headers, verify=localconfig['internal_ssl_verify'])
 
     return(ret)
 

anchore_engine/db/__init__.py

@@ -6,6 +6,7 @@
 from .entities.catalog import CatalogImage
 from .entities.catalog import CatalogImageDocker
 #from .entities.catalog import CatalogRepoTag
+from .entities.catalog import Event
 from .entities.catalog import EventLog
 from .entities.catalog import PolicyBundle
 from .entities.catalog import PolicyEval

anchore_engine/db/db_events.py

@@ -0,0 +1,146 @@
+import datetime
+
+from dateutil import parser as dateparser
+
+from anchore_engine import db
+from anchore_engine.db import Event
+from anchore_engine.subsys import logger
+
+
+def get_byfilter(userId, session=None, since=None, before=None, next=None, limit=100, **dbfilter):
+    if not session:
+        session = db.Session
+
+    ret = {'results': [], 'next': None}
+
+    query = session.query(Event).filter(Event.resource_user_id == userId)
+
+    if dbfilter:
+        query = query.filter_by(**dbfilter)
+
+    if before:
+        query = query.filter(Event.timestamp < before)
+
+    if since:
+        query = query.filter(Event.timestamp > since)
+
+    if next:
+        query = query.filter(Event.timestamp < next)
+
+    query = query.order_by(Event.timestamp.desc()).limit(limit + 1)
+
+    for db_event in query:
+        if len(ret) < limit:
+            ret['results'].append(_db_to_dict(db_event))
+        else:
+            ret['next'] = db_event.timestamp.isoformat()
+
+    return ret
+
+
+def get_byevent_id(userId, eventId, session=None):
+    if not session:
+        session = db.Session
+
+    db_event = session.query(Event).filter(Event.resource_user_id == userId, Event.generated_uuid == eventId).one_or_none()
+
+    return _db_to_dict(db_event) if db_event else None
+
+
+def add(msg, session=None):
+    if not session:
+        session = db.Session
+
+    db_event = _dict_to_db(msg)
+
+    session.add(db_event)
+    session.flush()
+    res = db_event.to_detached()
+
+    return _db_to_dict(res)
+
+
+def delete_byfilter(userId, session=None, since=None, before=None):
+    if not session:
+        session = db.Session
+
+    ret = {'deleted': []}
+
+    query = session.query(Event).filter(Event.resource_user_id == userId)
+
+    if before:
+        query = query.filter(Event.timestamp < before)
+
+    if since:
+        query = query.filter(Event.timestamp > since)
+
+    for db_event in query:
+        ret['deleted'].append(db_event.generated_uuid)
+        session.delete(db_event)
+
+    return ret
+
+
+def delete_byevent_id(userId, eventId, session=None):
+    if not session:
+        session = db.Session
+
+    ret = None
+
+    db_event = session.query(Event).filter(Event.resource_user_id == userId, Event.generated_uuid == eventId).one_or_none()
+    if db_event:
+        ret = db_event.generated_uuid
+        session.delete(db_event)
+
+    return ret
+
+
+def _db_to_dict(db_event):
+    msg = {'event': {}, 'generated_uuid': None, 'created_at': None}
+
+    for key, value in vars(db_event).iteritems():
+        if key.startswith('_'):
+            continue
+
+        if key in ['generated_uuid', 'created_at']:
+            msg[key] = value if type(value) != datetime.datetime else _format_timestamp(value)
+        elif value:
+            if key.startswith('resource') or key.startswith('source'):
+                key1, key2 = key.split('_', 1)
+                if key1 not in msg['event']:
+                    msg['event'][key1] = {}
+                msg['event'][key1][key2] = value if type(value) != datetime.datetime else _format_timestamp(value)
+            else:
+                msg['event'][key] = value if type(value) != datetime.datetime else _format_timestamp(value)
+
+    return msg
+
+
+def _dict_to_db(msg):
+    db_event = Event()
+
+    event_msg = {}
+    event_msg.update(msg)
+
+    if event_msg.get('source', None):
+        db_event.source_servicename = event_msg['source'].get('servicename', None)
+        db_event.source_hostid = event_msg['source'].get('hostid', None)
+        db_event.source_base_url = event_msg['source'].get('base_url', None)
+        db_event.source_request_id = event_msg['source'].get('request_id', None)
+
+    if event_msg.get('resource', None):
+        db_event.resource_user_id = event_msg['resource'].get('user_id', None)
+        db_event.resource_id = event_msg['resource'].get('id', None)
+        db_event.resource_type = event_msg['resource'].get('type', None)
+
+    db_event.type = event_msg['type']
+    db_event.level = event_msg['level']
+    db_event.message = event_msg['message']
+    db_event.details = event_msg.get('details', {})
+    db_event.timestamp = dateparser.parse(event_msg['timestamp'])
+
+    return db_event
+
+
+def _format_timestamp(ts):
+    return ts.isoformat() + 'Z'

anchore_engine/db/entities/catalog.py

@@ -4,10 +4,10 @@
 """
 import datetime
 
-from sqlalchemy import Column, Integer, String, Boolean, BigInteger, DateTime, LargeBinary
+from sqlalchemy import Column, Integer, String, Boolean, BigInteger, DateTime, LargeBinary, Index
 from sqlalchemy import inspect
 
-from .common import Base, anchore_now, UtilMixin
+from .common import Base, anchore_now, anchore_uuid, UtilMixin, StringJSON, anchore_now_datetime
 
 
 class Anchore(Base, UtilMixin):
@@ -113,6 +113,35 @@ def __repr__(self):
         return "hostId='%s' message='%s' level='%s'" % (self.hostId, self.message, self.level)
 
 
+class Event(Base, UtilMixin):
+    __tablename__ = 'events'
+
+    generated_uuid = Column(String, primary_key=True, default=anchore_uuid)
+    created_at = Column(DateTime, default=anchore_now_datetime)
+    resource_user_id = Column(String, nullable=True)
+    resource_id = Column(String, nullable=True)
+    resource_type = Column(String, nullable=True)
+    source_servicename = Column(String, nullable=True)
+    source_base_url = Column(String, nullable=True)
+    source_hostid = Column(String, nullable=True)
+    source_request_id = Column(String, nullable=True)
+    type = Column(String)
+    level = Column(String)
+    message = Column(String)
+    details = Column(StringJSON)
+    timestamp = Column(DateTime)
+
+    __table_args__ = (Index('ix_timestamp', timestamp.desc()),
+                      Index('ix_resource_user_id', resource_user_id),
+                      Index('ix_resource_type', resource_type),
+                      Index('ix_source_servicename', source_servicename),
+                      Index('ix_source_hostid', source_hostid),
+                      Index('ix_level', level))
+
+    def __repr__(self):
+        return "generated_uuid='%s' level='%s' message='%s'" % (self.generated_uuid, self.level, self.message)
+
+
 class QueueItem(Base, UtilMixin):
     """
     Queue data used by notification system for queueing up notifications for delivery.