Allow unsafe shutdown in Logstash

docs/logstash-pipelines.md

@@ -203,6 +203,10 @@ Here the `default` output only receives the events that haven't already been sen
 
 Every Output can have a `congestion:` option with a numerical value. If the Redis key already holds more items than the value says, the output will stop.
 
+### Unsafe shutdown ###
+
+If you need unsafe Logstash shutdowns, e.g. for testing, you can set `logstash_pipeline_unsafe_shutdown` to `true`. If you want better controll over which pipeline is allowed to shutdown unsafely, there are `ansible_input_unsafe_shutdown`and `ansible_forwarder_unsafe_shutdown` for default pipelines. And every pipeline has it's own `unsafe_shutdown` setting. All three default to the value of `logstash_pipeline_unsafe_shutdown` which by itself defaults to `false`. 
+
 ## Caveats ##
 
 There are still some minor issues you need to keep in mind:

molecule/elasticstack_default/converge.yml

@@ -15,6 +15,7 @@
     elasticstack_release: "{{ lookup('env', 'ELASTIC_RELEASE') | int}}"
     elasticsearch_heap: "1"
     elasticstack_full_stack: true
+    logstash_pipeline_unsafe_shutdown: true
     beats_filebeat_syslog_udp: true
     beats_filebeat_syslog_tcp: true
     beats_filebeat_modules:

molecule/logstash_full_stack-oss/converge.yml

@@ -17,6 +17,7 @@
     elasticstack_full_stack: true
     elasticstack_variant: oss
     logstash_security: false
+    logstash_pipeline_unsafe_shutdown: true
     elasticstack_security: false
     beats_filebeat_syslog_udp: true
     beats_filebeat_syslog_tcp: true

molecule/logstash_pipelines/converge.yml

@@ -29,6 +29,7 @@
             key: forwarder
     logstash_pipeline_identifier_field_name: "[mytest][pipelines]"
     logstash_pipeline_identifier_defaults: true
+    logstash_pipeline_unsafe_shutdown: true
     elasticstack_release: "{{ lookup('env', 'ELASTIC_RELEASE') | int}}"
     elasticstack_full_stack: false
   tasks:

molecule/logstash_specific_version/converge.yml

@@ -12,6 +12,7 @@
     logstash_logging_console: false
     logstash_logging_slow_file: false
     logstash_pipeline_identifier: false
+    logstash_pipeline_unsafe_shutdown: true
     elasticstack_release: "{{ lookup('env', 'ELASTIC_RELEASE') | int}}"
     elasticstack_full_stack: false
   tasks:

roles/logstash/defaults/main.yml

@@ -17,6 +17,7 @@ logstash_config_path_logs: /var/log/logstash
 # pipeline configuration #
 logstash_manage_pipelines: true
 logstash_queue_type: persisted
+logstash_pipeline_unsafe_shutdown: false
 
 # this will deactivate all pipeline management
 logstash_no_pipelines: false
@@ -38,10 +39,10 @@ logstash_pipelines:
 logstash_elasticsearch_output: true
 logstash_validate_after_inactivity: '300'
 logstash_beats_input: true
-ansible_input_queue_type: memory
-ansible_input_queue_max_bytes: 1gb
-ansible_forwarder_queue_type: memory
-ansible_forwarder_queue_max_bytes: 1gb
+logstash_input_queue_type: memory
+logstash_input_queue_max_bytes: 1gb
+logstash_forwarder_queue_type: memory
+logstash_forwarder_queue_max_bytes: 1gb
 logstash_sniffing: false
 
 # logstash security

roles/logstash/templates/pipelines.yml.j2

@@ -19,21 +19,29 @@
 # Default beat input #
 # Autoconfigured Redis outputs: input
 
-- pipeline.id: ansible-input
-  path.config: "/etc/logstash/conf.d/ansible-input/*.conf"
-  queue.type: {{ ansible_input_queue_type }}
-  queue.max_bytes: {{ ansible_input_queue_max_bytes }}
+- pipeline
+    id: ansible-input
+    unsafe_shutdown: {{ ansible_input_unsafe_shutdown | default({{ logstash_pipeline_unsafe_shutdown }}) }}
+  path
+    config: "/etc/logstash/conf.d/ansible-input/*.conf"
+  queue
+    type: {{ logstash_input_queue_type }}
+    max_bytes: {{ logstash_input_queue_max_bytes }}
 
 {% endif %}
 {% if logstash_elasticsearch_output | bool %}
 
 # Default elasticsearch output #
 # Autoconfigured Redis input: forwarder
 
-- pipeline.id: ansible-forwarder
-  path.config: "/etc/logstash/conf.d/ansible-forwarder/*.conf"
-  queue.type: {{ ansible_forwarder_queue_type }}
-  queue.max_bytes: {{ ansible_forwarder_queue_max_bytes }}
+- pipeline
+    id: ansible-forwarder
+    unsafe_shutdown: {{ ansible_forwarder_unsafe_shutdown | default({{ logstash_pipeline_unsafe_shutdown }}) }}
+  path
+    config: "/etc/logstash/conf.d/ansible-forwarder/*.conf"
+  queue
+    type: {{ logstash_forwarder_queue_type }}
+    max_bytes: {{ logstash_forwarder_queue_max_bytes }}
 
 {% endif %}
 {% if logstash_pipelines is defined %}
@@ -64,10 +72,14 @@
 
 {% endif %}
 
-- pipeline.id: {{ item.name }}
-  path.config: "/etc/logstash/conf.d/{{ item.name }}/*.conf"
-  queue.type: {{ item.queue_type | default('memory') }}
-  queue.max_bytes: {{ item.queue_max_bytes | default('1gb') }}
+- pipeline
+    id: {{ item.name }}
+    unsafe_shutdown: {{ item.unsafe_shutdown | default({{ logstash_pipeline_unsafe_shutdown }}) }}
+  path
+    config: "/etc/logstash/conf.d/{{ item.name }}/*.conf"
+  queue
+    type: {{ item.queue_type | default('memory') }}
+    max_bytes: {{ item.queue_max_bytes | default('1gb') }}
 
 {% endfor %}
 {% endif %}