diff --git a/packages/logstash/_dev/deploy/docker/config/logstash.yml b/packages/logstash/_dev/deploy/docker/config/logstash.yml
new file mode 100644
index 00000000000..90c16104430
--- /dev/null
+++ b/packages/logstash/_dev/deploy/docker/config/logstash.yml
@@ -0,0 +1,2 @@
+http.host: "0.0.0.0"
+config.reload.automatic: true
diff --git a/packages/logstash/_dev/deploy/docker/config/pipelines.yml b/packages/logstash/_dev/deploy/docker/config/pipelines.yml
new file mode 100644
index 00000000000..cfb1b51d35d
--- /dev/null
+++ b/packages/logstash/_dev/deploy/docker/config/pipelines.yml
@@ -0,0 +1,7 @@
+- pipeline.id: pipeline-with-persisted-queue
+  path.config: "/usr/share/logstash/pipeline/persisted-queue.conf"
+  queue.type: persisted
+- pipeline.id: pipeline-with-memory-queue
+  path.config: "/usr/share/logstash/pipeline/memory-queue.conf"
+- pipeline.id: standalone-pipeline
+  path.config: "/usr/share/logstash/pipeline/standalone-pipeline.conf"
diff --git a/packages/logstash/_dev/deploy/docker/docker-compose.yml b/packages/logstash/_dev/deploy/docker/docker-compose.yml
new file mode 100644
index 00000000000..2ba386e1d29
--- /dev/null
+++ b/packages/logstash/_dev/deploy/docker/docker-compose.yml
@@ -0,0 +1,9 @@
+version: '2.3'
+services:
+  logstash:
+    image: "docker.elastic.co/logstash/logstash:8.5.0-SNAPSHOT"
+    volumes:
+      - "./pipeline:/usr/share/logstash/pipeline"
+      - "./config:/usr/share/logstash/config"
+    ports:
+      - "127.0.0.1:9600:9600"
diff --git a/packages/logstash/_dev/deploy/docker/pipeline/memory-queue.conf b/packages/logstash/_dev/deploy/docker/pipeline/memory-queue.conf
new file mode 100644
index 00000000000..0e3ed11ebf4
--- /dev/null
+++ b/packages/logstash/_dev/deploy/docker/pipeline/memory-queue.conf
@@ -0,0 +1,15 @@
+input {
+  java_generator {
+    eps => 10
+  }
+}
+
+output {
+  elasticsearch {
+    hosts => ["https://elasticsearch:9200"]
+    user => "elastic"
+    password => "changeme"
+    index => "logstash-memory-queue"
+    ssl_certificate_verification => "false"
+  }
+}
diff --git a/packages/logstash/_dev/deploy/docker/pipeline/persisted-queue.conf b/packages/logstash/_dev/deploy/docker/pipeline/persisted-queue.conf
new file mode 100644
index 00000000000..38dfb247cfa
--- /dev/null
+++ b/packages/logstash/_dev/deploy/docker/pipeline/persisted-queue.conf
@@ -0,0 +1,15 @@
+input {
+  java_generator {
+    eps => 50
+  }
+}
+
+output {
+  elasticsearch {
+    hosts => ["https://elasticsearch:9200"]
+    user => "elastic"
+    password => "changeme"
+    index => "logstash-persisted-queue"
+    ssl_certificate_verification => "false"
+  }
+}
diff --git a/packages/logstash/_dev/deploy/docker/pipeline/standalone-pipeline.conf b/packages/logstash/_dev/deploy/docker/pipeline/standalone-pipeline.conf
new file mode 100644
index 00000000000..c0bfd17db2d
--- /dev/null
+++ b/packages/logstash/_dev/deploy/docker/pipeline/standalone-pipeline.conf
@@ -0,0 +1,9 @@
+input {
+  java_generator {
+    eps => 5
+  }
+}
+
+output {
+  stdout { }
+}
diff --git a/packages/logstash/data_stream/node/fields/ecs.yml b/packages/logstash/data_stream/node/fields/ecs.yml
index 0fb98cc9575..cb8e74422f1 100644
--- a/packages/logstash/data_stream/node/fields/ecs.yml
+++ b/packages/logstash/data_stream/node/fields/ecs.yml
@@ -15,3 +15,11 @@
   external: ecs
 - name: ecs.version
   external: ecs
+- name: event.dataset
+  external: ecs
+- name: event.duration
+  external: ecs
+- name: event.module
+  external: ecs
+- name: host.name
+  external: ecs
diff --git a/packages/logstash/data_stream/node/fields/fields.yml b/packages/logstash/data_stream/node/fields/fields.yml
index baf21e108e8..4e71f8e6ce1 100644
--- a/packages/logstash/data_stream/node/fields/fields.yml
+++ b/packages/logstash/data_stream/node/fields/fields.yml
@@ -1,45 +1,49 @@
-- name: logstash.node
+- name: logstash
   type: group
-  release: ga
   fields:
-    - name: host
+    - name: elasticsearch.cluster.id
       type: keyword
-    - name: version
-      type: keyword
-    - name: id
-      type: keyword
-    - name: state.pipeline
+    - name: node
       type: group
       fields:
-        - name: id
+        - name: jvm
+          type: group
+          fields:
+            - name: version
+              type: keyword
+              ignore_above: 1024
+        - name: host
           type: keyword
-        - name: hash
+        - name: version
           type: keyword
-        - name: ephemeral_id
+        - name: id
           type: keyword
-        - name: batch_size
-          type: long
-        - name: workers
-          type: long
-        - name: representation
+        - name: state.pipeline
           type: group
           fields:
-            - name: hash
+            - name: id
               type: keyword
-            - name: type
+            - name: hash
               type: keyword
-            - name: version
+            - name: ephemeral_id
               type: keyword
-            - name: graph
+            - name: batch_size
+              type: long
+            - name: workers
+              type: long
+            - name: representation
               type: group
               fields:
-                - name: edges
-                  type: object
-                - name: vertices
-                  type: object
-    - name: jvm
-      type: group
-      fields:
-        - name: version
-          type: keyword
-          description: JVM version
+                - name: hash
+                  type: keyword
+                - name: type
+                  type: keyword
+                - name: version
+                  type: keyword
+                - name: graph
+                  type: group
+                  fields:
+                    - name: edges
+                      type: object
+                    - name: vertices
+                      type: object
diff --git a/packages/logstash/data_stream/node/fields/package-fields.yml b/packages/logstash/data_stream/node/fields/package-fields.yml
index fdb0b7a24d4..f7af108fdca 100644
--- a/packages/logstash/data_stream/node/fields/package-fields.yml
+++ b/packages/logstash/data_stream/node/fields/package-fields.yml
@@ -1,6 +1,15 @@
+- name: cluster_uuid
+  type: alias
+  path: logstash.elasticsearch.cluster.id
+- name: timestamp
+  type: alias
+  path: '@timestamp'
 - name: logstash_stats
   type: group
   fields:
+    - name: timestamp
+      type: alias
+      path: '@timestamp'
     - name: jvm
       type: group
       fields:
@@ -37,8 +46,6 @@
         - name: version
           type: alias
           path: logstash.node.stats.logstash.version
-    - name: pipelines
-      type: nested
     - name: os
       type: group
       fields:
@@ -90,5 +97,3 @@
     - name: pipeline.hash
       type: alias
       path: logstash.node.state.pipeline.hash
-- name: logstash
-  type: group
diff --git a/packages/logstash/data_stream/node/manifest.yml b/packages/logstash/data_stream/node/manifest.yml
index df8425c4da6..ffd5b60089f 100644
--- a/packages/logstash/data_stream/node/manifest.yml
+++ b/packages/logstash/data_stream/node/manifest.yml
@@ -1,6 +1,10 @@
 type: metrics
 title: Logstash node metrics
 release: experimental
+elasticsearch:
+  index_template:
+    mappings:
+      dynamic: false
 streams:
   - input: logstash/metrics
     title: Logstash node metrics
diff --git a/packages/logstash/data_stream/node_stats/fields/ecs.yml b/packages/logstash/data_stream/node_stats/fields/ecs.yml
index e441309a0ed..cb8e74422f1 100644
--- a/packages/logstash/data_stream/node_stats/fields/ecs.yml
+++ b/packages/logstash/data_stream/node_stats/fields/ecs.yml
@@ -1,8 +1,8 @@
 - name: '@timestamp'
   external: ecs
-- name: host.hostname
+- name: service.id
   external: ecs
-- name: process.pid
+- name: service.type
   external: ecs
 - name: service.version
   external: ecs
@@ -11,9 +11,15 @@
   description: Address where data about this service was collected from.
 - name: service.name
   external: ecs
-- name: service.id
-  external: ecs
-- name: service.type
+- name: process.pid
   external: ecs
 - name: ecs.version
   external: ecs
+- name: event.dataset
+  external: ecs
+- name: event.duration
+  external: ecs
+- name: event.module
+  external: ecs
+- name: host.name
+  external: ecs
diff --git a/packages/logstash/data_stream/node_stats/fields/fields.yml b/packages/logstash/data_stream/node_stats/fields/fields.yml
index b52ce13d236..11430915fac 100644
--- a/packages/logstash/data_stream/node_stats/fields/fields.yml
+++ b/packages/logstash/data_stream/node_stats/fields/fields.yml
@@ -1,101 +1,138 @@
-- name: logstash.node
+- name: logstash
   type: group
-  release: ga
   fields:
-    - name: state.pipeline
+    - name: elasticsearch.cluster.id
+      type: keyword
+    - name: node
       type: group
       fields:
-        - name: id
-          type: keyword
-        - name: hash
-          type: keyword
-    - name: jvm
-      type: group
-      fields:
-        - name: version
-          type: keyword
-          description: JVM version
-    - name: stats
-      type: group
-      fields:
-        - name: jvm
+        - name: state.pipeline
           type: group
           fields:
-            - name: uptime_in_millis
-              type: long
-            - name: mem
-              type: group
-              fields:
-                - name: heap_used_in_bytes
-                  type: long
-                - name: heap_max_in_bytes
-                  type: long
-        - name: events
-          type: group
-          fields:
-            - name: in
-              type: long
-              description: |
-                Incoming events counter.
-            - name: out
-              type: long
-              description: |
-                Outgoing events counter.
-            - name: filtered
-              type: long
-              description: |
-                Filtered events counter.
-            - name: duration_in_millis
-              type: long
-        - name: logstash
-          type: group
-          fields:
-            - name: uuid
+            - name: id
               type: keyword
-            - name: version
+            - name: hash
               type: keyword
-        - name: os
+        - name: stats
           type: group
           fields:
-            - name: cpu
+            - name: timestamp
+              type: date
+            - name: jvm
               type: group
               fields:
-                - name: load_average
+                - name: uptime_in_millis
+                  type: long
+                - name: mem
                   type: group
                   fields:
-                    - name: 15m
-                      type: long
-                    - name: 1m
+                    - name: heap_used_in_bytes
                       type: long
-                    - name: 5m
+                    - name: heap_max_in_bytes
                       type: long
-            - name: cgroup
+            - name: events
               type: group
               fields:
-                - name: cpuacct.usage_nanos
+                - name: in
+                  type: long
+                  description: |
+                    Incoming events counter.
+                - name: out
                   type: long
+                  description: |
+                    Outgoing events counter.
+                - name: filtered
+                  type: long
+                  description: |
+                    Filtered events counter.
+                - name: duration_in_millis
+                  type: long
+            - name: logstash
+              type: group
+              fields:
+                - name: uuid
+                  type: keyword
+                - name: version
+                  type: keyword
+            - name: os
+              type: group
+              fields:
                 - name: cpu
                   type: group
                   fields:
-                    - name: stat
+                    - name: load_average
                       type: group
                       fields:
-                        - name: number_of_elapsed_periods
+                        - name: 15m
                           type: long
-                        - name: time_throttled_nanos
+                        - name: 1m
                           type: long
-                        - name: number_of_times_throttled
+                        - name: 5m
                           type: long
-        - name: process.cpu.percent
-          type: double
-        - name: pipelines
-          type: nested
-          fields:
-            - name: id
+                - name: cgroup
+                  type: group
+                  fields:
+                    - name: cpuacct.usage_nanos
+                      type: long
+                    - name: cpu
+                      type: group
+                      fields:
+                        - name: stat
+                          type: group
+                          fields:
+                            - name: number_of_elapsed_periods
+                              type: long
+                            - name: time_throttled_nanos
+                              type: long
+                            - name: number_of_times_throttled
+                              type: long
+            - name: process.cpu.percent
+              type: double
+            - name: queue.events_count
+              type: long
+            - name: pipelines
+              type: nested
+              fields:
+                - name: vertices
+                  type: nested
+                  fields:
+                    - name: id
+                      type: keyword
+                      description: id
+                    - name: long_counters.name
+                      type: keyword
+                    - name: long_counters
+                      type: nested
+                      fields:
+                        - name: value
+                          type: long
+                    - name: duration_in_millis
+                      type: long
+                    - name: events_in
+                      type: long
+                    - name: pipeline_ephemeral_id
+                      type: keyword
+                      description: pipeline_ephemeral_id
+                    - name: events_out
+                      type: long
+                      description: events_out
+                    - name: queue_push_duration_in_millis
+                      type: long
+                      description: queue_push_duration_in_millis
+            - name: pipelines.ephemeral_id
               type: keyword
-            - name: hash
+            - name: pipelines.id
+              type: keyword
+            - name: pipelines.hash
               type: keyword
-            - name: queue
+            - name: pipelines.reloads
+              type: group
+              fields:
+                - name: failures
+                  type: long
+                - name: successes
+                  type: long
+            - name: pipelines.queue
               type: group
               fields:
                 - name: events_count
@@ -106,31 +143,40 @@
                   type: long
                 - name: max_queue_size_in_bytes
                   type: long
-            - name: events
+            - name: pipelines.events
               type: group
               fields:
+                - name: in
+                  type: long
                 - name: out
                   type: long
-                - name: duration_in_millis
+                - name: filtered
                   type: long
-            - name: vertices
-              type: group
-              fields:
                 - name: duration_in_millis
                   type: long
-                - name: events_in
-                  type: long
-                - name: pipeline_ephemeral_id
-                  type: keyword
-                  description: pipeline_ephemeral_id
-                - name: events_out
-                  type: long
-                  description: events_out
-                - name: id
-                  type: keyword
-                  description: id
                 - name: queue_push_duration_in_millis
-                  type: float
-                  description: queue_push_duration_in_millis
-        - name: queue.events_count
-          type: long
+                  type: long
+            - name: pipelines.vertices
+              type: nested
+            - name: pipelines.vertices.id
+              type: keyword
+              description: id
+            - name: pipelines.vertices.long_counters
+              type: nested
+            - name: pipelines.vertices.long_counters.name
+              type: keyword
+            - name: pipelines.vertices.long_counters.value
+              type: long
+            - name: pipelines.vertices.duration_in_millis
+              type: long
+            - name: pipelines.vertices.events_in
+              type: long
+            - name: pipelines.vertices.pipeline_ephemeral_id
+              type: keyword
+              description: pipeline_ephemeral_id
+            - name: pipelines.vertices.events_out
+              type: long
+              description: events_out
+            - name: pipelines.vertices.queue_push_duration_in_millis
+              type: long
+              description: queue_push_duration_in_millis
diff --git a/packages/logstash/data_stream/node_stats/fields/package-fields.yml b/packages/logstash/data_stream/node_stats/fields/package-fields.yml
index fdb0b7a24d4..92ec0ae3e20 100644
--- a/packages/logstash/data_stream/node_stats/fields/package-fields.yml
+++ b/packages/logstash/data_stream/node_stats/fields/package-fields.yml
@@ -1,6 +1,17 @@
+- name: cluster_uuid
+  type: alias
+  path: logstash.elasticsearch.cluster.id
+- name: timestamp
+  type: alias
+  path: '@timestamp'
 - name: logstash_stats
   type: group
   fields:
+    - name: pipelines
+      type: nested
+    - name: timestamp
+      type: alias
+      path: '@timestamp'
     - name: jvm
       type: group
       fields:
@@ -37,8 +48,6 @@
         - name: version
           type: alias
           path: logstash.node.stats.logstash.version
-    - name: pipelines
-      type: nested
     - name: os
       type: group
       fields:
@@ -90,5 +99,3 @@
     - name: pipeline.hash
       type: alias
       path: logstash.node.state.pipeline.hash
-- name: logstash
-  type: group
diff --git a/packages/logstash/data_stream/node_stats/manifest.yml b/packages/logstash/data_stream/node_stats/manifest.yml
index 6c4a1e1768b..5d24f75f9db 100644
--- a/packages/logstash/data_stream/node_stats/manifest.yml
+++ b/packages/logstash/data_stream/node_stats/manifest.yml
@@ -1,6 +1,10 @@
 type: metrics
 title: Logstash node_stats metrics
 release: experimental
+elasticsearch:
+  index_template:
+    mappings:
+      dynamic: false
 streams:
   - input: logstash/metrics
     title: Logstash node stats