Merge branch 'master' into devel

hp-labrea/labrea-idea.py

@@ -34,6 +34,9 @@ class WindowContextMgr(object):
         self.ideagen = ideagen
         self.first_update_queue = OrderedDict()
         self.last_update_queue = OrderedDict()
+        # Hammer to mitigate too big events
+        self.max_count = 2000
+        self.max_src_ports = 1024
 
     def expire_queue(self, queue, window):
         aggr_events = []
@@ -68,7 +71,14 @@ class WindowContextMgr(object):
                     self.first_update_queue[ctx] = self.update_timestamp
                     self.last_update_queue[ctx] = self.update_timestamp
                 else:
-                    self.ctx_append(self.contexts[ctx], event)
+                    if not self.ctx_append(self.contexts[ctx], event):
+                        closed = self.ctx_close(self.contexts[ctx])
+                        if closed is not None:
+                            aggr_events.append(closed)
+                        del self.contexts[ctx]
+                        del self.first_update_queue[ctx]
+                        del self.last_update_queue[ctx]
+                    else:
                         del self.last_update_queue[ctx]
                         self.last_update_queue[ctx] = self.update_timestamp
 
@@ -107,6 +117,7 @@ class PingContextMgr(WindowContextMgr):
         ctx["tgt_ips"].add(event.tgt_ip)
         ctx["count"] += 1
         ctx["last_update"] = self.update_timestamp
+        return ctx["count"] < self.max_count
 
     def ctx_close(self, ctx):
         return self.ideagen.gen_idea(
@@ -143,11 +154,13 @@ class ConnectContextMgr(WindowContextMgr):
         ctx["src_ports"].add(event.src_port)
         ctx["count"] += 1
         ctx["last_update"] = self.update_timestamp
+        return ctx["count"] < self.max_count
 
     def ctx_close(self, ctx):
+        src_ports = ctx["src_ports"] if len(ctx["src_ports"]) <= self.max_src_ports else None
         return self.ideagen.gen_idea(
             src=ctx["src_ip"],
-            src_ports=ctx["src_ports"],
+            src_ports=src_ports,
             targets=ctx["tgt_ips_ports"].items(),
             detect_time=self.update_timestamp,
             event_time=ctx["first_update"],
@@ -419,10 +432,6 @@ def daemonize(
                 os.close(fd)
             except Exception:
                 pass
-    # Redirect stdin, stdout, stderr to /dev/null
-    devnull = os.open(os.devnull, os.O_RDWR)
-    for fd in range(3):
-        os.dup2(devnull, fd)
     # PID file
     if pidfile is not None:
         pidd = os.open(pidfile, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_TRUNC)
@@ -436,6 +445,10 @@ def daemonize(
                 os.unlink(pidfile)
             except Exception:
                 pass
+    # Redirect stdin, stdout, stderr to /dev/null
+    devnull = os.open(os.devnull, os.O_RDWR)
+    for fd in range(3):
+        os.dup2(devnull, fd)
 
 
 def save_events(aggr, filer):