Merge develop to master for release 6.4.0

.gitignore

@@ -22,6 +22,7 @@ eventgen_wsgi.conf
 .cache
 *.xml
 !tests/large/splunk/input.xml
+!splunk_eventgen/splunk_app/default/data/ui/nav/default.xml
 dist
 _book
 *.result

LICENSE

@@ -187,7 +187,7 @@
       same "printed page" as the copyright notice for easier
       identification within third-party archives.
 
-   Copyright [yyyy] [name of copyright owner]
+   Copyright 2019 Splunk, Inc.
 
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
@@ -200,3 +200,57 @@
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
+
+=======================================================================
+Splunk Event Generator
+
+Splunk Event Generator project contains subcomponents with separate copyright
+notices and license terms. Your use of the source code for the these
+subcomponents is subject to the terms and conditions of the following
+licenses.
+
+========================================================================
+Apache License 2.0
+========================================================================
+The following components are provided under the Apache License 2.0. See project link for details.
+
+    (Apache License 2.0) boto3 (https://github.com/boto/boto3/blob/master/LICENSE)
+    (Apache License 2.0) requests (https://github.com/kennethreitz/requests/blob/master/LICENSE)
+    (Apache License 2.0) pyOpenSSL (https://github.com/pyca/pyopenssl/blob/master/LICENSE)
+    (Apache License 2.0) docker (https://github.com/docker/docker-py/blob/master/LICENSE)
+    (Apache License 2.0) requests-futures (https://github.com/ross/requests-futures/blob/master/LICENSE)
+    (Apache License 2.0) nameko (https://github.com/nameko/nameko/blob/master/LICENSE.txt)
+
+========================================================================
+MIT licenses
+========================================================================
+The following components are provided under the MIT License. See project link for details.
+
+    (MIT License) pyyaml (https://github.com/yaml/pyyaml/blob/master/LICENSE)
+    (MIT License) httplib2 (https://github.com/httplib2/httplib2/blob/master/LICENSE)
+    (MIT License) urllib3 (https://github.com/urllib3/urllib3/blob/master/LICENSE.txt)
+    (MIT License) isort (https://github.com/timothycrosley/isort/blob/develop/LICENSE)
+    (MIT License) flake8 (https://gitlab.com/pycqa/flake8/blob/master/LICENSE)
+    (MIT License) pytest (https://github.com/pytest-dev/pytest/blob/master/LICENSE)
+    (MIT License) pytest-mock (https://github.com/pytest-dev/pytest-mock/blob/master/LICENSE)
+    (MIT License) pytest-xdist (https://github.com/pytest-dev/pytest-xdist/blob/master/LICENSE)
+    (MIT License) pytest-cov (https://github.com/pytest-dev/pytest-cov/blob/master/LICENSE)
+
+========================================================================
+BSD-style licenses
+========================================================================
+
+The following components are provided under a BSD-style license. See project link for details.
+
+    (BSD 2-Clause "Simplified" License) mock (https://github.com/testing-cabal/mock/blob/master/LICENSE.txt)
+    (BSD 3-Clause) pyrabbit (https://github.com/bkjones/pyrabbit/blob/master/LICENSE)
+    (BSD 3-Clause) logutils (https://opensource.org/licenses/BSD-3-Clause)
+    (BSD 3-Clause) jinja2 (https://github.com/pallets/jinja/blob/master/LICENSE)
+    (BSD 3-Clause) ujson(https://github.com/esnme/ultrajson/blob/master/LICENSE.txt)
+
+========================================================================
+PSF licenses
+========================================================================
+
+The following components are provided under a PSF license. See project link for details.
+    (PSD License) futures (https://github.com/agronholm/pythonfutures/blob/master/LICENSE)

README.md

@@ -1,7 +1,7 @@
 # Splunk Event Generator (Eventgen)
 
 ### Status
-[![CircleCI](https://circleci.com/gh/splunk/eventgen/tree/develop.svg?style=svg)](https://circleci.com/gh/splunk/eventgen/tree/develop)
+[![CircleCI](https://circleci.com/gh/splunk/eventgen/tree/develop.svg?style=svg&circle-token=15e952a75e368102d8cebc6d9445af87e6c7d57e)](https://circleci.com/gh/splunk/eventgen/tree/develop)
 
 ### Introduction
 

dockerfiles/Dockerfile

@@ -22,13 +22,13 @@ RUN apk --no-cache upgrade && \
 	erlang-asn1 \
 	erlang-inets \
 	erlang-os-mon \
-	erlang-xmerl \ 
+	erlang-xmerl \
 	erlang-eldap \
 	erlang-syntax-tools \
 	pwgen \
 	xz \
 	curl \
-	bash && \ 
+	bash && \
 	rm -rf /var/cache/apk/* && \
 	curl -sL https://www.rabbitmq.com/releases/rabbitmq-server/v${RABBITMQ_VERSION}/rabbitmq-server-generic-unix-${RABBITMQ_VERSION}.tar.xz | tar -xJ -C /usr/local && \
 	ln -s /usr/local/rabbitmq_server-${RABBITMQ_VERSION}/sbin/rabbitmq-server /usr/sbin/rabbitmq-server && \

docs/CONFIGURE.md

@@ -459,6 +459,9 @@ specifically be supported by all plugins. Plugins that write to files like spool
     httpeventWaitResponse = <bool>
     * wait for all responses on a generator output before returning the outputter.
     * Defaults to true.
+---
+    httpeventAllowFailureCount = <int>
+    * Number of transmission failure allowed for a certain httpserver before we remove that server from the pool. For example, 100 means that we will no longer include a specific httpserver after 100 failures. Even after some failures, if we see a success for the server, we will reset the count and continue the transmission.
 
 ###### spool
     spoolDir = <spool directory>

docs/REFERENCE.md

@@ -59,6 +59,7 @@ maxQueueLength = 0
 autotimestamps = [ <jsonlist> ]
 autotimestamp = false
 outputCounter = false
+disableLoggingQueue = true
 
 
 [<sample file name>]
@@ -105,6 +106,11 @@ useOutputQueue = true | false
       for instance if you're outputting to a file or to stdout/modular input.
     * Default value depends on the output plugin being used.
 
+disableLoggingQueue = true | false
+    * Disable the logging queue for process mode
+    * In process mode, logs in each process will be collected via a logging queue
+    * Default is true which will disable the logging queue
+
 #############################
 ## OUTPUT RELATED SETTINGS ##
 #############################
@@ -161,6 +167,9 @@ httpeventWaitResponse = <bool>
     * wait for all responses on a generator output before returning the outputter.
     * Defaults to true.
 
+httpeventAllowFailureCount = <int>
+    * Number of transmission failure allowed for a certain httpserver before we remove that server from the pool. For example, 100 means that we will no longer include a specific httpserver after 100 failures. Even after some failures, if we see a success for the server, we will reset the count and continue the transmission.
+
 spoolDir = <spool directory>
     * Spool directory is the generated files destination directory.
     * Only valid in spool outputMode.

setup.cfg

@@ -19,4 +19,4 @@ each_dict_entry_on_separate_line = false
 
 [isort]
 # isort/yapf solutions to below files are not compatible
-skip = splunk_eventgen/lib/concurrent,splunk_eventgen/lib/requests_futures
+skip = splunk_eventgen/lib/concurrent,splunk_eventgen/lib/requests_futures

splunk_eventgen/__main__.py

@@ -53,6 +53,9 @@ def parse_args():
                                     help="Use multiprocesing instead of threading")
     generate_subparser.add_argument("--profiler", action="store_true", help="Turn on cProfiler")
     generate_subparser.add_argument("--log-path", type=str, default="{0}/logs".format(FILE_LOCATION))
+    generate_subparser.add_argument(
+        "--generator-queue-size", type=int, default=500, help="the max queue size for the "
+        "generator queue, timer object puts all the generator tasks into this queue, default max size is 500")
     # Build subparser
     build_subparser = subparsers.add_parser('build', help="Will build different forms of sa-eventgen")
     build_subparser.add_argument("--mode", type=str, default="splunk-app",
@@ -308,6 +311,16 @@ def build_splunk_app(dest, source=os.getcwd(), remove=True):
     directory_default_dir = os.path.join(directory, 'default', 'eventgen.conf')
     eventgen_conf = os.path.join(module_path, 'default', 'eventgen.conf')
     shutil.copyfile(eventgen_conf, directory_default_dir)
+
+    # install 3rd lib dependencies
+    install_target = os.path.join(directory, 'lib')
+    install_cmd = "pip install --requirement splunk_eventgen/lib/requirements.txt --upgrade --no-compile " + \
+                  "--no-binary :all: --target " + install_target
+    return_code = os.system(install_cmd)
+    if return_code != 0:
+        print("Failed to install dependencies via pip. Please check whether pip is installed.")
+    os.system("rm -rf " + os.path.join(install_target, "*.egg-info"))
+
     make_tarfile(target_file, directory)
     shutil.rmtree(splunk_app_samples)
     if remove:

splunk_eventgen/default/eventgen.conf

@@ -48,3 +48,4 @@ useOutputQueue = false
 autotimestamps = [["\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}", "%Y-%m-%d %H:%M:%S"], ["\\d{1,2}\\/\\w{3}\\/\\d{4}\\s\\d{2}:\\d{2}:\\d{2}:\\d{1,3}", "%d/%b/%Y %H:%M:%S:%f"], ["\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}", "%Y-%m-%dT%H:%M:%S.%f"], ["\\d{1,2}/\\w{3}/\\d{4}\\s\\d{2}:\\d{2}:\\d{2}:\\d{1,3}", "%d/%b/%Y %H:%M:%S:%f"], ["\\d{1,2}/\\d{2}/\\d{2}\\s\\d{1,2}:\\d{2}:\\d{2}", "%m/%d/%y %H:%M:%S"], ["\\d{2}-\\d{2}-\\d{4} \\d{2}:\\d{2}:\\d{2}", "%m-%d-%Y %H:%M:%S"], ["\\w{3} \\w{3} +\\d{1,2} \\d{2}:\\d{2}:\\d{2}", "%a %b %d %H:%M:%S"], ["\\w{3} \\w{3} \\d{2} \\d{4} \\d{2}:\\d{2}:\\d{2}", "%a %b %d %Y %H:%M:%S"], ["^(\\w{3}\\s+\\d{1,2}\\s\\d{2}:\\d{2}:\\d{2})", "%b %d %H:%M:%S"], ["(\\w{3}\\s+\\d{1,2}\\s\\d{1,2}:\\d{1,2}:\\d{1,2})", "%b %d %H:%M:%S"], ["(\\w{3}\\s\\d{1,2}\\s\\d{1,4}\\s\\d{1,2}:\\d{1,2}:\\d{1,2})", "%b %d %Y %H:%M:%S"], ["\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\.\\d{3}", "%Y-%m-%d %H:%M:%S.%f"], ["\\,\\d{2}\\/\\d{2}\\/\\d{2,4}\\s+\\d{2}:\\d{2}:\\d{2}\\s+[AaPp][Mm]\\,", ",%m/%d/%Y %I:%M:%S %p,"], ["^\\w{3}\\s+\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}", "%b %d %H:%M:%S"], ["\\d{2}/\\d{2}/\\d{4} \\d{2}:\\d{2}:\\d{2}", "%m/%d/%Y %H:%M:%S"], ["^\\d{2}\\/\\d{2}\\/\\d{2,4}\\s+\\d{2}:\\d{2}:\\d{2}\\s+[AaPp][Mm]", "%m/%d/%Y %I:%M:%S %p"], ["\\d{2}\\/\\d{2}\\/\\d{4}\\s\\d{2}:\\d{2}:\\d{2}", "%m-%d-%Y %H:%M:%S"], ["\\\"timestamp\\\":\\s\\\"(\\d+)", "%s"], ["\\d{2}\\/\\w+\\/\\d{4}\\s\\d{2}:\\d{2}:\\d{2}:\\d{3}", "%d-%b-%Y %H:%M:%S:%f"], ["\\\"created\\\":\\s(\\d+)", "%s"], ["\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}", "%Y-%m-%dT%H:%M:%S"], ["\\d{1,2}/\\w{3}/\\d{4}:\\d{2}:\\d{2}:\\d{2}:\\d{1,3}", "%d/%b/%Y:%H:%M:%S:%f"], ["\\d{1,2}/\\w{3}/\\d{4}:\\d{2}:\\d{2}:\\d{2}", "%d/%b/%Y:%H:%M:%S"]]
 autotimestamp = false
 httpeventWaitResponse = true
+disableLoggingQueue = true

splunk_eventgen/eventgen_core.py

@@ -75,9 +75,15 @@ def __init__(self, args=None):
         # attach to the logging queue
         self.logger.info("Logging Setup Complete.")
 
+        self._generator_queue_size = getattr(self.args, 'generator_queue_size', 500)
+        if self._generator_queue_size < 0:
+            self._generator_queue_size = 0
+        self.logger.info("set generator queue size to %d", self._generator_queue_size)
+
         if self.args and 'configfile' in self.args and self.args.configfile:
             self._load_config(self.args.configfile, args=args)
 
+
     def _load_config(self, configfile, **kwargs):
         '''
         This method will use a configfile and set self.confg as a processeded config object,
@@ -115,8 +121,13 @@ def _load_config(self, configfile, **kwargs):
         self.config = Config(configfile, **new_args)
         self.config.parse()
         self._reload_plugins()
+        if "args" in kwargs and getattr(kwargs["args"], "generators"):
+            generator_worker_count = kwargs["args"].generators
+        else:
+            generator_worker_count = self.config.generatorWorkers
+
         # TODO: Probably should destroy pools better so processes are cleaned.
-        self._setup_pools()
+        self._setup_pools(generator_worker_count)
 
     def _reload_plugins(self):
         # Initialize plugins
@@ -133,7 +144,7 @@ def _reload_plugins(self):
                 os.path.join(file_path, 'lib', 'plugins', 'rater'), self.config.plugins, 'rater')
             self.config._complexSettings['rater'] = plugins
         except Exception as e:
-            self.logger.exception(e)
+            self.logger.exception(str(e))
 
     def _load_custom_plugins(self, PluginNotLoadedException):
         plugintype = PluginNotLoadedException.type
@@ -147,7 +158,7 @@ def _load_custom_plugins(self, PluginNotLoadedException):
         # APPPERF-263: be greedy when scanning plugin dir (eat all the pys)
         self._initializePlugins(plugindir, pluginsdict, plugintype)
 
-    def _setup_pools(self):
+    def _setup_pools(self, generator_worker_count):
         '''
         This method is an internal method called on init to generate pools needed for processing.
 
@@ -157,7 +168,7 @@ def _setup_pools(self):
         self._create_generator_pool()
         self._create_timer_threadpool()
         self._create_output_threadpool()
-        self._create_generator_workers()
+        self._create_generator_workers(generator_worker_count)
 
     def _create_timer_threadpool(self, threadcount=100):
         '''
@@ -212,15 +223,19 @@ def _create_generator_pool(self, workercount=20):
         if self.args.multiprocess:
             import multiprocessing
             self.manager = multiprocessing.Manager()
-            self.loggingQueue = self.manager.Queue()
-            self.logging_pool = Thread(target=self.logger_thread, args=(self.loggingQueue, ), name="LoggerThread")
-            self.logging_pool.start()
+            if self.config.disableLoggingQueue:
+                self.loggingQueue = None
+            else:
+                # TODO crash caused by logging Thread https://github.com/splunk/eventgen/issues/217
+                self.loggingQueue = self.manager.Queue()
+                self.logging_pool = Thread(target=self.logger_thread, args=(self.loggingQueue, ), name="LoggerThread")
+                self.logging_pool.start()
             # since we're now in multiprocess, we need to use better queues.
-            self.workerQueue = multiprocessing.JoinableQueue(maxsize=500)
+            self.workerQueue = multiprocessing.JoinableQueue(maxsize=self._generator_queue_size)
             self.genconfig = self.manager.dict()
             self.genconfig["stopping"] = False
         else:
-            self.workerQueue = Queue(maxsize=500)
+            self.workerQueue = Queue(maxsize=self._generator_queue_size)
             worker_threads = workercount
             if hasattr(self.config, 'outputCounter') and self.config.outputCounter:
                 self.output_counters = []
@@ -354,13 +369,13 @@ def _worker_do_work(self, work_queue, logging_queue):
                 startTime = time.time()
                 item.run()
                 totalTime = time.time() - startTime
-                if totalTime > self.config.interval:
+                if totalTime > self.config.interval and self.config.end != 1:
                     self.logger.warning("work took longer than current interval, queue/threading throughput limitation")
                 work_queue.task_done()
             except Empty:
                 pass
             except Exception as e:
-                self.logger.exception(e)
+                self.logger.exception(str(e))
                 raise e
 
     def _generator_do_work(self, work_queue, logging_queue, output_counter=None):
@@ -370,23 +385,27 @@ def _generator_do_work(self, work_queue, logging_queue, output_counter=None):
                 startTime = time.time()
                 item.run(output_counter=output_counter)
                 totalTime = time.time() - startTime
-                if totalTime > self.config.interval:
+                if totalTime > self.config.interval and item._sample.end != 1:
                     self.logger.warning("work took longer than current interval, queue/threading throughput limitation")
                 work_queue.task_done()
             except Empty:
                 pass
             except Exception as e:
-                self.logger.exception(e)
+                self.logger.exception(str(e))
                 raise e
 
     @staticmethod
     def _proc_worker_do_work(work_queue, logging_queue, config):
         genconfig = config
         stopping = genconfig['stopping']
-        qh = logutils.queue.QueueHandler(logging_queue)
         root = logging.getLogger()
         root.setLevel(logging.DEBUG)
-        root.addHandler(qh)
+        if logging_queue is not None:
+            # TODO https://github.com/splunk/eventgen/issues/217
+            qh = logutils.queue.QueueHandler(logging_queue)
+            root.addHandler(qh)
+        else:
+            root.addHandler(logging.StreamHandler())
         while not stopping:
             try:
                 root.info("Checking for work")
@@ -418,7 +437,7 @@ def logger_thread(self, loggingQueue):
             except Empty:
                 pass
             except Exception as e:
-                self.logger.exception(e)
+                self.logger.exception(str(e))
                 raise e
 
     def _initializePlugins(self, dirname, plugins, plugintype, name=None):
@@ -493,7 +512,7 @@ def _initializePlugins(self, dirname, plugins, plugintype, name=None):
                         self.logger.warn("Could not load plugin: %s, skipping" % mod_name.name)
                         self.logger.exception(ie)
                     except Exception as e:
-                        self.logger.exception(e)
+                        self.logger.exception(str(e))
                         raise e
         return ret
 
@@ -521,21 +540,6 @@ def start(self, join_after_start=True):
         if join_after_start:
             self.logger.info("All timers started, joining queue until it's empty.")
             self.join_process()
-        # Only need to start timers once
-        # Every 5 seconds, get values and output basic statistics about our operations
-        # TODO: Figure out how to do this better...
-        # generatorsPerSec = (generatorDecrements - generatorQueueCounter) / 5
-        # outputtersPerSec = (outputDecrements - outputQueueCounter) / 5
-        # outputQueueCounter = outputDecrements
-        # generatorQueueCounter = generatorDecrements
-        # self.logger.info('OutputQueueDepth=%d GeneratorQueueDepth=%d GeneratorsPerSec=%d OutputtersPerSec=%d' %
-        #                  (self.config.outputQueueSize.value(), self.config.generatorQueueSize.value(),
-        #                   generatorsPerSec, outputtersPerSec))
-        # kiloBytesPerSec = self.config.bytesSent.valueAndClear() / 5 / 1024
-        # gbPerDay = (kiloBytesPerSec / 1024 / 1024) * 60 * 60 * 24
-        # eventsPerSec = self.config.eventsSent.valueAndClear() / 5
-        # self.logger.info('GlobalEventsPerSec=%s KilobytesPerSec=%1f GigabytesPerDay=%1f' %
-        #                  (eventsPerSec, kiloBytesPerSec, gbPerDay))
 
     def join_process(self):
         '''
@@ -550,7 +554,7 @@ def join_process(self):
             self.logger.info("All timers have finished, signalling workers to exit.")
             self.stop()
         except Exception as e:
-            self.logger.exception(e)
+            self.logger.exception(str(e))
             raise e
 
     def stop(self):
@@ -577,7 +581,7 @@ def stop(self):
                     count += 1
         self.logger.info("All generators working/exited, joining output queue until it's empty.")
         self.outputQueue.join()
-        self.logger.info("All items fully processed, stopping.")
+        self.logger.info("All items fully processed. Cleaning up internal processes.")
         self.started = False
         self.stopping = False