Consumer config management (#1163)

core/events/consumers.py

@@ -126,7 +126,10 @@ def run(self):
         prog="yeti-consumer", description="Consume events and logs from the event bus"
     )
     parser.add_argument(
-        "--concurrency", type=int, default=None, help="Number of consumers to start"
+        "--concurrency",
+        type=int,
+        help="Number of consumers to start",
+        default=yeti_config.get("events", "consumers_concurrency", None),
     )
     parser.add_argument(
         "type", choices=["events", "logs"], help="Type of consumer to start"

extras/docker/docker-entrypoint.sh

@@ -5,6 +5,8 @@ if [[ "$1" = 'webserver' ]]; then
     poetry run uvicorn core.web.webapp:app --reload --host 0.0.0.0
 elif [[ "$1" = 'tasks' ]]; then
     poetry run celery -A core.taskscheduler worker --loglevel=INFO --purge -B -P threads
+elif [[ "$1" = 'events-tasks' ]]; then
+    poetry run python -m core.events.consumers events
 elif [[ "$1" = 'create-user' ]]; then
     poetry run python yetictl/cli.py create-user "${@:2}"
 elif [[ "$1" = 'reset-password' ]]; then

yeti.conf.sample

@@ -79,7 +79,10 @@ enabled = True
 # tls = ok
 
 [events]
+# if max_queue_size is not defined, defaults to 30000
 max_queue_size = 30000
+# if concurrency is not defined, defaults to multiprocessing.cpu_count
+consumers_concurrency = 2
 
 [misp]