Fetch events from k8s API, small Kubeutil refactory, tests

resolve host_ip

minor improvements, fixed log prints

post events to DD

fixed timestamp evaluation

fix timestamp evaluation

again on ts logic

do not overwrite last timestamp with 0

skip events from other nodes

default value if port is missing from instance

update skip condition

debug logs++

use node ip as host

moved timestamp register to kubeutil

fixed tests

added kubeutil tests

added tests for events

checks.d/kubernetes.py

@@ -10,6 +10,7 @@
 from fnmatch import fnmatch
 import numbers
 import re
+import time
 
 # 3rd party
 import requests
@@ -18,7 +19,6 @@
 # project
 from checks import AgentCheck
 from config import _is_affirmative
-from utils.http import retrieve_json
 from utils.kubeutil import KubeUtil
 
 NAMESPACE = "kubernetes"
@@ -46,6 +46,8 @@
     RATE: {True: HISTORATE, False: RATE}
 }
 
+EVENT_TYPE = 'kubernetes'
+
 
 class Kubernetes(AgentCheck):
     """ Collect metrics and events from kubelet """
@@ -239,11 +241,8 @@ def _update_container_metrics(self, instance, subcontainer, kube_labels):
                               sum(float(net[x]) for x in NET_ERRORS),
                               tags)
 
-    def _retrieve_metrics(self, url):
-        return retrieve_json(url)
-
     def _update_metrics(self, instance, pods_list):
-        metrics = self._retrieve_metrics(self.kubeutil.metrics_url)
+        metrics = self.kubeutil.retrieve_metrics()
 
         excluded_labels = instance.get('excluded_labels')
         kube_labels = self.kubeutil.extract_kube_labels(pods_list, excluded_keys=excluded_labels)
@@ -289,19 +288,66 @@ def _process_events(self, instance, pods_list):
         """
         Retrieve a list of events from kubelet that is relevant for the host the
         agent is running on.
+
+        At the moment there is no support to select events based on a timestamp query, so we
+        go through the whole list every time. This should be fine for now as events
+        have a TTL of one hour[1] but logic needs to improve as soon as they provide
+        query capabilities or at least pagination, see [2][3].
+
+        [1] https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-apiserver/app/options/options.go#L50
+        [2] https://github.com/kubernetes/kubernetes/issues/4432
+        [3] https://github.com/kubernetes/kubernetes/issues/1362
         """
-        pods = self.kubeutil.filter_pods_list(pods_list, self.kubeutil.host)
-        pod_uids = self.kubeutil.extract_uids(pods)
+        node_ip, node_name = self.kubeutil.get_node_info()
+        self.log.debug('Processing events on {} [{}]'.format(node_name, node_ip))
+        pods = self.kubeutil.filter_pods_list(pods_list, node_ip)
+        pod_uids = self.kubeutil.extract_meta(pods, 'uid')
 
         k8s_namespace = instance.get('namespace', 'default')
         events_endpoint = '{}/namespaces/{}/events'.format(self.kubeutil.kubernetes_api_url, k8s_namespace)
         self.log.debug('Kubernetes API endpoint to query events: %s' % events_endpoint)
 
         events = self.kubeutil.retrieve_json_auth(events_endpoint, self.kubeutil.get_auth_token())
         event_items = events.get('items') or []
-        self.log.debug('Found {} events, filtering out unwanted ones...'.format(len(event_items)))
+        last_read = self.kubeutil.last_event_collection_ts[k8s_namespace]
+        most_recent_read = 0
+
+        self.log.debug('Found {} events, filtering out using timestamp: {}'.format(len(event_items), last_read))
 
         for event in event_items:
-            involved_obj = event.get('involvedObject')
-            if involved_obj and involved_obj.get('uid') in pod_uids:
-                self.log.error('Should post this event:', event.get('reason'), event.get('message'))
+            # skip if the event is from another node
+            involved_obj = event.get('involvedObject', {})
+            if involved_obj.get('uid') not in pod_uids:
+                self.log.debug('Pod {} is not running on this node, skip...'.format(involved_obj.get('uid')))
+                continue
+
+            # skip if the event is too old
+            event_ts = int(time.mktime(time.strptime(event.get('lastTimestamp'), '%Y-%m-%dT%H:%M:%SZ')))
+            if event_ts <= last_read:
+                continue
+
+            # compute the most recently seen event, without relying on items order
+            if event_ts > most_recent_read:
+                most_recent_read = event_ts
+
+            title = '{} {} on {}'.format(involved_obj.get('name'), event.get('reason'), node_name)
+            message = event.get('message')
+            source = event.get('source')
+            if source:
+                message += '\nSource: {} {}\n'.format(source.get('component', ''), source.get('host', ''))
+            msg_body = "%%%\n{}\n```\n{}\n```\n%%%".format(title, message)
+            dd_event = {
+                'timestamp': event_ts,
+                'host': node_ip,
+                'event_type': EVENT_TYPE,
+                'msg_title': title,
+                'msg_text': msg_body,
+                'source_type_name': EVENT_TYPE,
+                'event_object': 'kubernetes:{}'.format(involved_obj.get('name')),
+            }
+            self.log.debug('Posting event: {}'.format(dd_event))
+            self.event(dd_event)
+
+        if most_recent_read > 0:
+            self.kubeutil.last_event_collection_ts[k8s_namespace] = most_recent_read
+            self.log.debug('_last_event_collection_ts is now {}'.format(most_recent_read))