cleanup memory

splunk_eventgen/eventgen_core.py

@@ -263,6 +263,9 @@ def _worker_do_work(self, work_queue, logging_queue):
                 work_queue.task_done()
             except Empty:
                 pass
+            except EOFError as ef:
+                self.logger.exception(str(ef))
+                continue
             except Exception as e:
                 self.logger.exception(str(e))
                 raise e
@@ -279,6 +282,9 @@ def _generator_do_work(self, work_queue, logging_queue, output_counter=None):
                 work_queue.task_done()
             except Empty:
                 pass
+            except EOFError as ef:
+                self.logger.exception(str(ef))
+                continue
             except Exception as e:
                 if self.force_stop:
                     break

splunk_eventgen/lib/eventgenoutput.py

@@ -71,21 +71,6 @@ def flush(self, endOfInterval=False):
         Flushes output buffer, unless endOfInterval called, and then only flush if we've been called
         more than maxIntervalsBeforeFlush tunable.
         """
-        # TODO: Fix interval flushing somehow with a queue, not sure I even want to support this feature anymore.
-        '''if endOfInterval:
-            logger.debugv("Sample calling flush, checking increment against maxIntervalsBeforeFlush")
-            c.intervalsSinceFlush[self._sample.name].increment()
-            if c.intervalsSinceFlush[self._sample.name].value() >= self._sample.maxIntervalsBeforeFlush:
-                logger.debugv("Exceeded maxIntervalsBeforeFlush, flushing")
-                flushing = True
-                c.intervalsSinceFlush[self._sample.name].clear()
-            else:
-                logger.debugv("Not enough events to flush, passing flush routine.")
-        else:
-            logger.debugv("maxQueueLength exceeded, flushing")
-            flushing = True'''
-
-        # TODO: This is set this way just for the time being while I decide if we want this feature.
         flushing = True
         if flushing:
             q = self._queue
@@ -113,3 +98,5 @@ def flush(self, endOfInterval=False):
                             self._sample.name, 'events': len(tmp), 'bytes': sum(tmp)})
                     tmp = None
                 outputer.run()
+            q = None
+

splunk_eventgen/lib/generatorplugin.py

@@ -37,6 +37,8 @@ def build_events(self, eventsDict, startTime, earliest, latest, ignore_tokens=Fa
         """Ready events for output by replacing tokens and updating the output queue"""
         # Replace tokens first so that perDayVolume evaluates the correct event length
         send_objects = self.replace_tokens(eventsDict, earliest, latest, ignore_tokens=ignore_tokens)
+        # after replace_tokens() is called, we don't need eventsDict
+        del eventsDict
         try:
             self._out.bulksend(send_objects)
             self._sample.timestamp = None

splunk_eventgen/lib/outputplugin.py

@@ -37,11 +37,6 @@ def run(self):
         if self.output_counter is not None:
             self.output_counter.collect(len(self.events), sum([len(e['_raw']) for e in self.events]))
         self.events = None
-        self._output_end()
-
-    def _output_end(self):
-        pass
-
 
 def load():
     return OutputPlugin

splunk_eventgen/lib/plugins/output/httpevent.py

@@ -77,6 +77,7 @@ def flush(self, q):
                     payload.append(payloadFragment)
                 logger.debug("Finished processing events, sending all to splunk")
                 self._sendHTTPEvents(payload)
+                payload = []
                 if self.config.httpeventWaitResponse:
                     for session in self.active_sessions:
                         response = session.result()

splunk_eventgen/lib/plugins/output/httpevent_core.py

@@ -207,7 +207,6 @@ def _transmitEvents(self, payloadstring):
             headers['content-type'] = 'application/json'
             try:
                 payloadsize = len(payloadstring)
-                # response = requests.post(url, data=payloadstring, headers=headers, verify=False)
                 self.active_sessions.append(
                     self.session.post(url=url, data=payloadstring, headers=headers, verify=False))
             except Exception as e:

splunk_eventgen/lib/plugins/output/stdout.py

@@ -11,7 +11,7 @@ def __init__(self, sample, output_counter=None):
 
     def flush(self, q):
         for x in q:
-            print(x['_raw'].rstrip())
+            print(x.get('_raw', '').rstrip())
 
 
 def load():