diff --git a/data/common.yaml b/data/common.yaml
index b104df1..826298c 100644
--- a/data/common.yaml
+++ b/data/common.yaml
@@ -1,235 +1,248 @@
 ---
 
 fluentbit::manage_package_repo: true
-fluentbit::repo_key_fingerprint: F209D8762A60CD49E680633B4FF8368B6EA0722A
+fluentbit::repo_key_fingerprint: C3C0A28534B9293EAF51FABD9F9DDC083888C1CD
 fluentbit::repo_key_location: 'https://packages.fluentbit.io/fluentbit.key'
 fluentbit::package_ensure: present
-fluentbit::package_name: 'td-agent-bit'
+fluentbit::package_name: 'fluent-bit'
 
 fluentbit::manage_service: true
 fluentbit::service_enable: true
 fluentbit::service_has_status: true
 fluentbit::service_has_restart: true
+fluentbit::service_restart_command: 'systemctl reload-or-restart fluent-bit'
 fluentbit::service_ensure: 'running'
-fluentbit::service_name: 'td-agent-bit'
+fluentbit::service_name: 'fluent-bit'
 fluentbit::manage_config_dir: true
+fluentbit::manage_data_dir: true
+fluentbit::manage_storage_dir: true
 
-fluentbit::config_file: '/etc/td-agent-bit/td-agent-bit.conf'
+fluentbit::binary_file: '/opt/fluent-bit/bin/fluent-bit'
+fluentbit::data_dir: '/opt/fluent-bit/db'
+fluentbit::config_file: '/etc/fluent-bit/fluent-bit.conf'
 fluentbit::config_file_mode: '0644'
+fluentbit::config_folder_mode: '0750'
 fluentbit::flush: 5
+fluentbit::grace: 5
 fluentbit::daemon: false
-fluentbit::log_file: ~
+fluentbit::dns_mode: UDP
 fluentbit::log_level: info
 fluentbit::http_server: false
 fluentbit::http_listen: '0.0.0.0'
 fluentbit::http_port: 2020
 fluentbit::coro_stack_size: 24576
+fluentbit::scheduler_cap: 2000
+fluentbit::scheduler_base: 5
+fluentbit::json_convert_nan_to_null: false
 
 fluentbit::variables: {}
 
-fluentbit::storage_path: ~
-fluentbit::storage_sync: ~
+fluentbit::storage_path: '/opt/fluent-bit/storage'
+fluentbit::storage_sync: normal
 fluentbit::storage_checksum: false
-fluentbit::storage_backlog_mem_limit: ~
+fluentbit::storage_max_chunks_up: 128
+fluentbit::storage_backlog_mem_limit: 5M
+fluentbit::storage_metrics: true
+fluentbit::storage_delete_irrecoverable_chunks: false
 
-fluentbit::input_plugins: []
-fluentbit::output_plugins: []
-fluentbit::filter_plugins: []
+fluentbit::input_plugins: {}
+fluentbit::output_plugins: {}
+fluentbit::filter_plugins: {}
 
 fluentbit::manage_plugins_file: true
-fluentbit::plugins_file: '/etc/td-agent-bit/plugins.conf'
+fluentbit::plugins_file: '/etc/fluent-bit/plugins.conf'
 fluentbit::plugins: []
 
 fluentbit::manage_streams_file: true
-fluentbit::streams_file: '/etc/td-agent-bit/streams.conf'
-fluentbit::streams: []
+fluentbit::streams_file: '/etc/fluent-bit/streams.conf'
+fluentbit::streams: {}
 
 fluentbit::manage_parsers_file: true
-fluentbit::parsers_file: '/etc/td-agent-bit/parsers.conf'
+fluentbit::parsers_file: '/etc/fluent-bit/parsers.conf'
 fluentbit::parsers:
 
-  - name:   'apache'
+  'apache':
     format: 'regex'
     regex:  '^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$'
     time_key: 'time'
     time_format: '%d/%b/%Y:%H:%M:%S %z'
 
-  - name:   'apache2'
+  'apache2':
     format: 'regex'
     regex:  '^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>.*)")?$'
     time_key: 'time'
     time_format: '%d/%b/%Y:%H:%M:%S %z'
 
-  - name:   'apache_error'
+  'apache_error':
     format: 'regex'
     regex:  '^\[[^ ]* (?<time>[^\]]*)\] \[(?<level>[^\]]*)\](?: \[pid (?<pid>[^\]]*)\])?( \[client (?<client>[^\]]*)\])? (?<message>.*)$'
 
-  - name:   'nginx'
+  'nginx':
     format: 'regex'
     regex: '^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")'
     time_key: 'time'
     time_format: '%d/%b/%Y:%H:%M:%S %z'
 
-  - name:        'k8s-nginx-ingress'
+  'k8s-nginx-ingress':
     format:      'regex'
     regex:       '^(?<host>[^ ]*) - \[(?<real_ip>[^ ]*)\] - (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*) "(?<referer>[^\"]*)" "(?<agent>[^\"]*)" (?<request_length>[^ ]*) (?<request_time>[^ ]*) \[(?<proxy_upstream_name>[^ ]*)\] (?<upstream_addr>[^ ]*) (?<upstream_response_length>[^ ]*) (?<upstream_response_time>[^ ]*) (?<upstream_status>[^ ]*) (?<last>[^$]*)'
     time_key:    'time'
     time_format: '%d/%b/%Y:%H:%M:%S %z'
 
-  - name:   'json'
+  'json':
     format: 'json'
     time_key: 'time'
     time_format: '%d/%b/%Y:%H:%M:%S %z'
 
-  - name:         'docker'
+  'docker':
     format:       'json'
     time_key:     'time'
     time_format:  '%Y-%m-%dT%H:%M:%S.%L'
     time_keep:    true
 
-  - name:        'docker-daemon'
+  'docker-daemon':
     format:      'regex'
     regex:       'time="(?<time>[^ ]*)" level=(?<level>[^ ]*) msg="(?<msg>[^ ].*)"'
     time_key:    'time'
     time_format: '%Y-%m-%dT%H:%M:%S.%L'
     time_keep:   true
 
-  - name:        'syslog-rfc5424'
+  'syslog-rfc5424':
     format:      'regex'
     regex:       '^\<(?<pri>[0-9]{1,5})\>1 (?<time>[^ ]+) (?<host>[^ ]+) (?<ident>[^ ]+) (?<pid>[-0-9]+) (?<msgid>[^ ]+) (?<extradata>(\[(.*)\]|-)) (?<message>.+)$'
     time_key:    'time'
     time_format: '%Y-%m-%dT%H:%M:%S.%L'
     time_keep:   true
 
-  - name:        'syslog-rfc3164-local'
+  'syslog-rfc3164-local':
     format:      'regex'
-    regex:       '^\<(?<pri>[0-9]+)\>(?<time>[^ ]* {1,2}[^ ]* [^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\])?(?:[^\:]*\:)? *(?<message>.*)$'
+    regex:       '^(?:\<(?<pri>[0-9]+)\>)?(?<time>[^ ]* {1,2}[^ ]* [^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\])?(?:[^\:]*\:)? *(?<message>.*)$'
     time_key:    'time'
     time_format: '%b %d %H:%M:%S'
     time_keep:   true
 
-  - name:        'syslog-rfc3164'
+  'syslog-rfc3164':
     format:      'regex'
-    regex:       '/^\<(?<pri>[0-9]+)\>(?<time>[^ ]* {1,2}[^ ]* [^ ]*) (?<host>[^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\])?(?:[^\:]*\:)? *(?<message>.*)$/'
+    regex:       '^(?:\<(?<pri>[0-9]+)\>)?(?<time>[^ ]* {1,2}[^ ]* [^ ]*) (?<host>[^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\])?(?:[^\:]*\:)? *(?<message>.*)$/'
     time_key:    'time'
     time_format: '%b %d %H:%M:%S'
     time_format: '%Y-%m-%dT%H:%M:%S.%L'
     time_keep:   true
 
-  - name:    'mongodb'
+  'mongodb':
     format:  'regex'
     regex:   '^(?<time>[^ ]*)\s+(?<severity>\w)\s+(?<component>[^ ]+)\s+\[(?<context>[^\]]+)]\s+(?<message>.*?) *(?<ms>(\d+))?(:?ms)?$'
     time_format: '%Y-%m-%dT%H:%M:%S.%L'
     time_keep:   true
     time_key: 'time'
 
-  - name: 'cri'
+  'cri':
     format: 'regex'
     regex: '^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<logtag>[^ ]*) (?<message>.*)$'
     time_key:    'time'
     time_format: '%Y-%m-%dT%H:%M:%S.%L%z'
 
-  - name:    'kube-custom'
+  'kube-custom':
     format:  'regex'
     regex:   '(?<tag>[^.]+)?\.?(?<pod_name>[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)_(?<namespace_name>[^_]+)_(?<container_name>.+)-(?<docker_id>[a-z0-9]{64})\.log$'
 
-  - name: 'cinder'
+  'cinder':
     format: 'regex'
     regex: '^(?<log_time>[^ ][-.\d\+:T]+[ ]*[.:\d]*)\s+(?<pid>[^ ]\d+)\s+(?<severity>[^ ][.-_\w]+)\s+(?<component>[^ ][.-_\w]+)(\s+\[(-|(?<req_id>[^ ][-\w]*) (?<req_user>[^ ][-\w]*) (?<req_project>[^ ][-\w]*) (?<req_domain>[^ ][-\w]*) (?<req_user_domain>[^ ][-\w]*) (?<req_project_domain>[^ ][-\w]*))\]){1}\s+(?<message>.*)$'
     time_format: '%Y-%m-%d %H:%M:%S.%L'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'glance'
+  'glance':
     format: 'regex'
     regex: '^(?<log_time>[^ ][-.\d\+:T]+[ ]*[.:\d]*)\s+(?<pid>[^ ]\d+)\s+(?<severity>[^ ][.-_\w]+)\s+(?<component>[^ ][.-_\w]+)(\s+\[(-|(?<req_id>[^ ][-\w]*) (?<req_user>[^ ][-\w]*) (?<req_project>[^ ][-\w]*) (?<req_domain>[^ ][-\w]*) (?<req_user_domain>[^ ][-\w]*) (?<req_project_domain>[^ ][-\w]*))\]){1}\s+(?<message>.*)$'
     time_format: '%Y-%m-%d %H:%M:%S.%L'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'heat'
+  'heat':
     format: 'regex'
     regex: '^(?<log_time>[^ ][-.\d\+:T]+[ ]*[.:\d]*)\s+(?<pid>[^ ]\d+)\s+(?<severity>[^ ][.-_\w]+)\s+(?<component>[^ ][.-_\w]+)(\s+\[(-|(?<req_id>[^ ][-\w]*) (?<req_user>[^ ][-\w]*) (?<req_project>[^ ][-\w]*) (?<req_domain>[^ ][-\w]*) (?<req_user_domain>[^ ][-\w]*) (?<req_project_domain>[^ ][-\w]*))\]){1}\s+(?<message>.*)$'
     time_format: '%Y-%m-%d %H:%M:%S.%L'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'keystone'
+  'keystone':
     format: 'regex'
     regex: '^(?<log_time>[^ ][-.\d\+:T]+[ ]*[.:\d]*)\s+(?<pid>[^ ]\d+)\s+(?<severity>[^ ][.-_\w]+)\s+(?<component>[^ ][.-_\w]+)(\s+\[(-|(?<req_id>[^ ][-\w]*) (?<req_user>[^ ][-\w]*) (?<req_project>[^ ][-\w]*) (?<req_domain>[^ ][-\w]*) (?<req_user_domain>[^ ][-\w]*) (?<req_project_domain>[^ ][-\w]*))\]){1}\s+(?<message>.*)$'
     time_format: '%Y-%m-%d %H:%M:%S.%L'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'neutron'
+  'neutron':
     format: 'regex'
     regex: '^(?<log_time>[^ ][-.\d\+:T]+[ ]*[.:\d]*)\s+(?<pid>[^ ]\d+)\s+(?<severity>[^ ][.-_\w]+)\s+(?<component>[^ ][.-_\w]+)(\s+\[(-|(?<req_id>[^ ][-\w]*) (?<req_user>[^ ][-\w]*) (?<req_project>[^ ][-\w]*) (?<req_domain>[^ ][-\w]*) (?<req_user_domain>[^ ][-\w]*) (?<req_project_domain>[^ ][-\w]*))\]){1}\s+(?<message>.*)$'
     time_format: '%Y-%m-%d %H:%M:%S.%L'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'nova'
+  'nova':
     format: 'regex'
     regex: '^(?<log_time>[^ ][-.\d\+:T]+[ ]*[.:\d]*)\s+(?<pid>[^ ]\d+)\s+(?<severity>[^ ][.-_\w]+)\s+(?<component>[^ ][.-_\w]+)(\s+\[(-|(?<req_id>[^ ][-\w]*) (?<req_user>[^ ][-\w]*) (?<req_project>[^ ][-\w]*) (?<req_domain>[^ ][-\w]*) (?<req_user_domain>[^ ][-\w]*) (?<req_project_domain>[^ ][-\w]*))\]){1}\s+(?<message>.*)$'
     time_format: '%Y-%m-%d %H:%M:%S.%L'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'crowbar'
+  'crowbar':
     format: 'regex'
     regex: '^.*\[(?<log_time>[^ ][-.\d\+:]+T[:\d]*)([^\]])*?\]\s+?(?<severity>[^ ]\w+)([\s-]*):?\s+(?<message>.*)'
     time_format: '%Y-%m-%dT%H:%M:%S'
     time_keep: false
     time_key: 'log_time'
 
-  - name: 'chefclient'
+  'chefclient':
     format: 'regex'
     regex: '^\[(?<log_time>[^ ][-.\d\+:]+T[:\d]*)([^\]])*?\]\s+(?<severity>[^ ]\w+):\s+(?<message>.*)$'
     time_format: '%Y-%m-%dT%H:%M:%S'
     time_keep: false
     time_key: 'log_time'
 
-  - name: 'mysql_error'
+  'mysql_error':
     format: 'regex'
     regex: '^(?<log_time>[^ +][-\d]+[\ T]*[:\dZ]+)\s*(?<myid>[^ ]\d+)\s+\[(?<severity>[^ ]\w+)\](\s+(?<subsystem>[^ ]\w+):){0,1}\s+(?<message>.*)$'
     time_format: '%Y-%m-%d %H:%M:%S'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'mysql_slow'
+  'mysql_slow':
     format: 'regex'
     regex: '^# User\@Host:\s+(?<user>[^\@][\w\[\]]+)[@\s]+(?<dbhost>[^ ][-.\w]+)\s+(\[(?<dbhost_address>[.\d]+)\]){0,1}\s+(?<message>.*)$'
 
-  - name: 'pacemaker'
+  'pacemaker':
     format: 'regex'
     regex: '^\s*(?<log_time>[^ ]* {1,2}[^ ]* [^ ]*) \[(?<pid>\d+)\] (?<node>[\-\w]*)\s*(?<component>\w*):\s+(?<severity>\w+):\s+(?<message>.*)$'
     time_format: '%b %d %H:%M:%S'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'rabbitmq'
+  'rabbitmq':
     format: 'regex'
     regex: '^=(?<severity>[^ ]\w+)\s+REPORT[=\s]*(?<log_time>[^ =][-:.\d\w]+)[\s=]+(?<message>.*)$'
     time_format: '%d-%b-%Y::%H:%M:%S'
     time_keep:   false
     time_key: 'log_time'
 
-  - name: 'universal'
+  'universal':
     format: 'regex'
     regex: '^(?<message>.*)$'
 
-  - name: 'uuid'
+  'uuid':
     format: 'regex'
     regex: '(?<uuid>[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12})'
 
-  - name: 'ceph'
+  'ceph':
     format: 'regex'
     regex: '^(?<log_time>[^ ][-.\d\+:T]+[ ]*[.:\d]*)\s+(?<message>.*)$'
     time_format: '%Y-%m-%d %H:%M:%S.%L'
     time_keep: false
     time_key: 'log_time'
 
-  - name:        'java_multiline'
+  'java_multiline':
     format:      'regex'
     regex:       '/^(?<time>\d{4}-\d{1,2}-\d{1,2} \d{1,2}:\d{1,2}:\d{1,2}) \[(?<thread>.*)\] (?<level>[^\s]+)(?<message>.*)/'
     time_key:    'time'
diff --git a/manifests/config.pp b/manifests/config.pp
index a51f1bb..c8877bf 100644
--- a/manifests/config.pp
+++ b/manifests/config.pp
@@ -12,23 +12,44 @@
   File {
     ensure  => file,
     mode    => $fluentbit::config_file_mode,
-    notify  => Class['fluentbit::service'],
   }
 
   $config_dir = dirname($fluentbit::config_file)
+  $plugin_dir = "${config_dir}/pipelines"
 
   if $fluentbit::manage_config_dir {
     file { $config_dir:
       ensure  => directory,
       purge   => true,
       recurse => true,
-      mode    => '0755',
+      mode    => $fluentbit::config_folder_mode,
+    }
+    -> file { $plugin_dir:
+      ensure  => directory,
+      purge   => true,
+      recurse => true,
+      mode    => $fluentbit::config_folder_mode,
+    }
+  }
+
+  if $fluentbit::manage_data_dir {
+    file { $fluentbit::data_dir:
+      ensure => directory,
+      mode   => $fluentbit::config_folder_mode,
+    }
+  }
+
+  if $fluentbit::manage_storage_dir and $fluentbit::storage_path {
+    file { $fluentbit::storage_path:
+      ensure => directory,
+      mode   => $fluentbit::config_folder_mode,
     }
   }
 
   $flush = $fluentbit::flush
+  $grace = $fluentbit::grace
   $daemon = bool2str($fluentbit::daemon, 'On', 'Off')
-  $log_file = $fluentbit::log_file
+  $dns_mode = $fluentbit::dns_mode
   $log_level = $fluentbit::log_level
   $parsers_file = $fluentbit::parsers_file
   $plugins_file = $fluentbit::plugins_file
@@ -41,27 +62,80 @@
   $storage_sync = $fluentbit::storage_sync
   $storage_checksum = bool2str($fluentbit::storage_checksum, 'On', 'Off')
   $storage_backlog_mem_limit = $fluentbit::storage_backlog_mem_limit
-  $variables = $fluentbit::variables
+  $storage_max_chunks_up = $fluentbit::storage_max_chunks_up
+  $storage_metrics = bool2str($fluentbit::storage_metrics, 'On', 'Off')
+  $storage_delete_irrecoverable_chunks = bool2str($fluentbit::storage_delete_irrecoverable_chunks, 'On', 'Off')
+  $scheduler_cap = $fluentbit::scheduler_cap
+  $scheduler_base = $fluentbit::scheduler_base
+  $json_convert_nan_to_null = $fluentbit::json_convert_nan_to_null
+  $variables = $fluentbit::variables.filter |$k, $v| {
+    length($v) > 0
+  }
+
+  $storage_config = $storage_path ? {
+    undef            => {},
+    default          => {
+      'storage.path'                        => $storage_path,
+      'storage.sync'                        => $storage_sync,
+      'storage.checksum'                    => $storage_checksum,
+      'storage.max_chunks_up'               => $storage_max_chunks_up,
+      'storage.backlog.mem_limit'           => $storage_backlog_mem_limit,
+      'storage.metrics'                     => $storage_metrics,
+      'storage.delete_irrecoverable_chunks' => $storage_delete_irrecoverable_chunks,
+    },
+  }
 
   file { $fluentbit::config_file:
-    content => template('fluentbit/td-agent-bit.conf.erb'),
+    mode    => $fluentbit::config_file_mode,
+    content => epp('fluentbit/fluent-bit.conf.epp',
+      {
+        variables => $variables,
+        service   => {
+          'flush'                    => $flush,
+          'grace'                    => $grace,
+          'daemon'                   => $daemon,
+          'dns.mode'                 => $dns_mode,
+          'log_level'                => $log_level,
+          'parsers_file'             => $parsers_file,
+          'plugins_file'             => $plugins_file,
+          'streams_file'             => $streams_file,
+          'http_server'              => $http_server,
+          'http_listen'              => $http_listen,
+          'http_port'                => $http_port,
+          'coro_stack_size'          => $coro_stack_size,
+          'scheduler.cap'            => $scheduler_cap,
+          'scheduler.base'           => $scheduler_base,
+          'json.convert_nan_to_null' => $json_convert_nan_to_null,
+        } + $storage_config,
+      },
+    ),
   }
 
   $parsers = $fluentbit::parsers
 
   file { $fluentbit::parsers_file:
-    content => template('fluentbit/parsers.conf.erb'),
+    content => epp('fluentbit/parsers.conf.epp', { parsers => $parsers }),
   }
 
   $plugins = $fluentbit::plugins
 
   file { $fluentbit::plugins_file:
-    content => template('fluentbit/plugins.conf.erb'),
+    content => epp('fluentbit/plugins.conf.epp', { plugins => $plugins }),
   }
 
   $streams = $fluentbit::streams
 
   file { $fluentbit::streams_file:
-    content => template('fluentbit/streams.conf.erb'),
+    content => epp('fluentbit/streams.conf.epp', { streams => $streams }),
+  }
+
+  systemd::unit_file { "${fluentbit::service_name}.service":
+    ensure  => present,
+    content => epp('fluentbit/fluent-bit.service.epp',
+      {
+        'binary_file' => $fluentbit::binary_file,
+        'config_file' => $fluentbit::config_file,
+      }
+    ),
   }
 }
diff --git a/manifests/filter/modify.pp b/manifests/filter/modify.pp
deleted file mode 100644
index 7cefcf0..0000000
--- a/manifests/filter/modify.pp
+++ /dev/null
@@ -1,51 +0,0 @@
-# @summary Filter to modify records using rules and conditions
-#
-# @param configfile
-#  Path to the filter configfile. Naming should be filter_*.conf to make sure
-#  it's getting included by the main config.
-# @param set
-#  Add a key/value pair with key KEY and value VALUE.
-#  If KEY already exists, this field is overwritten
-# @param add
-#  Add a key/value pair with key KEY and value VALUE if KEY does not exist
-# @param remove
-#  Remove a key/value pair with key KEY if it exists
-# @param remove_wildcard
-#  Remove all key/value pairs with key matching wildcard KEY
-# @param remove_regex
-#  Remove all key/value pairs with key matching regexp KEY
-# @param rename
-#  Rename a key/value pair with key KEY to RENAMED_KEY if KEY exists AND RENAMED_KEY does not exist
-# @param hard_rename
-#  Rename a key/value pair with key KEY to RENAMED_KEY if KEY exists.
-#  If RENAMED_KEY already exists, this field is overwritten
-# @param copy
-#  Copy a key/value pair with key KEY to COPIED_KEY if KEY exists AND COPIED_KEY does not exist
-# @param hard_copy
-#  Copy a key/value pair with key KEY to COPIED_KEY if KEY exists.
-#  If COPIED_KEY already exists, this field is overwritten
-# @example
-#   fluentbit::filter::modify { 'namevar': }
-define fluentbit::filter::modify (
-  String $configfile         = '/etc/td-agent-bit/filter_modify.conf',
-  String $match              = '*',
-  Optional $set              = undef,
-  Optional $add              = undef,
-  Optional $remove           = undef,
-  Optional $remove_wildcard  = undef,
-  Optional $remove_regex     = undef,
-  Optional $rename           = undef,
-  Optional $hard_rename      = undef,
-  Optional $copy             = undef,
-  Optional $hard_copy        = undef,
-) {
-  # create filter_modify.conf
-  # TODO: concat for multiple entries
-  file { $configfile:
-    ensure  => file,
-    mode    => '0644',
-    content => template('fluentbit/filter/modify.conf.erb'),
-    notify  => Class['fluentbit::service'],
-    require => Class['fluentbit::install'],
-  }
-}
diff --git a/manifests/init.pp b/manifests/init.pp
index ecbaa24..c424985 100644
--- a/manifests/init.pp
+++ b/manifests/init.pp
@@ -10,9 +10,6 @@
 # @param output_plugins Hash of the OUTPUT plugins to be configured
 # @param filter_plugins Hash of the filter to be configured
 #
-# @param manage_package_repo
-#   Installs the package repositories
-#
 # @param repo_key_fingerprint
 #   GPG key identifier of the repository
 #
@@ -40,23 +37,36 @@
 #   Whether the fluentbit service should be running.
 #   Default value: 'running'.
 #
-# @param service_manage
-#   Whether to manage the fluentbit service.
-#   Default value: true.
-#
-# @param service_hasstatus
+# @param service_has_status
 #   Whether the service has a functional status command.
 #   Default value: true.
 #
-# @param service_hasrestart
+# @param service_has_restart
 #   Whether the service has a restart command.
 #   Default value: true.
 #
+# @param service_restart_command
+#   Optional command for restarting service
+#
 # @param manage_config_dir
 #   Whether to manage the configuration directory.
 #   When enabled, will remove all unmanaged files from the directory the
 #   configuration resides in.
 #   Default value: true
+
+# @param manage_data_dir
+#   Whether to manage the data directory.
+#   Default value: true
+
+# @param manage_storage_dir
+#   Whether to manage the storage directory. `storage_path` must be defined.
+#   Default value: true
+#
+# @param data_dir
+#   Path to data directory that will be used by plugins using DB feature.
+#
+# @param binary_file
+#   Path of the daemon binary.
 #
 # @param config_file
 #   Path of the daemon configuration.
@@ -64,6 +74,9 @@
 # @param config_file_mode
 #   File mode to apply to the daemon configuration file
 #
+# @param config_folder_mode
+#   File mode to apply folders managed by the module
+#
 # @param storage_path
 #   Set an optional location in the file system to store streams and chunks of data.
 #   If this parameter is not set, Input plugins can only use in-memory buffering.
@@ -79,11 +92,27 @@
 #   Default value: false
 #
 # @param storage_backlog_mem_limit
+#   If the input plugin has enabled filesystem storage type, this property sets the maximum 
+#   number of Chunks that can be up in memory.
+#   This is the setting to use to control memory usage when you enable storage.type filesystem.
+#   Default value: 128
+#
+# @param storage_max_chunks_up
 #   If storage.path is set, Fluent Bit will look for data chunks that were
 #   not delivered and are still in the storage layer, these are called backlog data.
 #   This option configure a hint of maximum value of memory to use when processing these records.
 #   Default value: 5M
 #
+# @param storage_metrics
+#   If http_server option has been enabled in the main [SERVICE] section, this option registers a new 
+#   endpoint where internal metrics of the storage layer can be consumed.
+#   Default value: false
+#
+# @param storage_delete_irrecoverable_chunks
+#   When enabled, irrecoverable chunks will be deleted during runtime, and any other irrecoverable chunk 
+#   located in the configured storage path directory will be deleted when Fluent-Bit starts.
+#   Default value: false
+#
 # @param manage_plugins_file
 #   Whether to manage the enabled external plugins
 #
@@ -102,6 +131,9 @@
 # @param streams
 #   Stream processing tasks
 #
+# @param upstreams
+#   Upstreams used by forward plugins
+#
 # @param manage_parsers_file
 #   Whether to manage the parser definitions
 #
@@ -115,10 +147,12 @@
 #
 # @param flush
 #   Set the flush time in seconds. Everytime it timeouts, the engine will flush the records to the output plugin.
+# @param grace
+#   Set the grace time in seconds. The engine loop uses a Grace timeout to define the wait time on exit.
 # @param daemon
 #   Boolean value to set if Fluent Bit should run as a Daemon (background) or not. Allowed values are: yes, no, on and off.
-# @param log_file
-#   Absolute path for an optional log file.
+# @param dns_mode
+#   Sets the primary transport layer protocol used by the asynchronous DNS resolver.
 # @param log_level
 #   Set the logging verbosity level.
 #   Values are: error, info, debug and trace. Values are accumulative,
@@ -133,10 +167,16 @@
 # @param coro_stack_size
 #   Set the coroutines stack size in bytes.
 #   The value must be greater than the page size of the running system.
+# @param scheduler_cap
+#   Set a maximum retry time in seconds. The property is supported from v1.8.7.
+# @param scheduler_base
+#   Sets the base of exponential backoff. The property is supported from v1.8.7.
+# @param json_convert_nan_to_null
+#   If enabled, NaN is converted to null when fluent-bit converts msgpack to json.
 #
 # @param variables
 #   macro definitions to use in the configuration file
-#   the will be registered using the *@SET* command.
+#   the will be registered using the *@SET* command or using Env section in YAML syntax.
 #
 class fluentbit (
   Boolean $manage_package_repo,
@@ -147,24 +187,32 @@
   Boolean $manage_service,
   Boolean $service_enable,
   Boolean $service_has_status,
+  Optional[String[1]] $service_restart_command,
   Boolean $service_has_restart,
   Stdlib::Ensure::Service $service_ensure,
   String[1] $service_name,
   Boolean $manage_config_dir,
+  Boolean $manage_data_dir,
+  Boolean $manage_storage_dir,
 
-  Array[Fluentbit::Plugin] $input_plugins,
-  Array[Fluentbit::Plugin] $output_plugins,
-  Array[Fluentbit::Plugin] $filter_plugins,
+  Hash[String, Hash] $input_plugins,
+  Hash[String, Hash] $output_plugins,
+  Hash[String, Hash] $filter_plugins,
 
-  Array[Fluentbit::Parser] $parsers,
-  Array[Fluentbit::Stream] $streams,
+  Hash[String, Fluentbit::Parser] $parsers,
+  Hash[String, Fluentbit::Stream] $streams,
+  Hash[String, Hash] $upstreams,
   Array[Stdlib::Absolutepath] $plugins,
 
+  Stdlib::Absolutepath $binary_file,
+  Stdlib::Absolutepath $data_dir,
   Stdlib::Absolutepath $config_file,
   Stdlib::Filemode $config_file_mode,
+  Stdlib::Filemode $config_folder_mode,
   Integer $flush,
+  Integer $grace,
   Boolean $daemon,
-  Optional[Stdlib::Absolutepath] $log_file,
+  Enum['UDP', 'TCP'] $dns_mode,
   Enum['error', 'warning', 'info', 'debug', 'trace'] $log_level,
   Boolean $manage_parsers_file,
   Stdlib::Absolutepath $parsers_file,
@@ -175,47 +223,33 @@
   Boolean $http_server,
   Stdlib::IP::Address::Nosubnet $http_listen,
   Stdlib::Port $http_port,
+  Integer $scheduler_cap,
+  Integer $scheduler_base,
+  Boolean $json_convert_nan_to_null,
 
   Optional[Stdlib::Absolutepath] $storage_path,
   Optional[Enum['normal', 'full']] $storage_sync,
   Boolean $storage_checksum,
+  Optional[Integer] $storage_max_chunks_up,
+  Boolean $storage_metrics,
+  Boolean $storage_delete_irrecoverable_chunks,
   Optional[String[1]] $storage_backlog_mem_limit,
 
   Hash $variables,
   Integer $coro_stack_size,
 ) {
-
   contain fluentbit::repo
   contain fluentbit::install
   contain fluentbit::config
   contain fluentbit::service
 
   Class['::fluentbit::repo']
-    -> Class['::fluentbit::install']
-    -> Class['::fluentbit::config']
-    ~> Class['::fluentbit::service']
-
-  $input_plugins.each |$index, $plugin| {
-    $name = $plugin['name']
-
-    Resource["fluentbit::input::${name}"] {
-      "i${index}": * => merge($plugin['properties']);
-    }
-  }
-
-  $output_plugins.each |$index, $plugin| {
-    $name = $plugin['name']
-
-    Resource["fluentbit::output::${name}"] {
-      "o${index}": * => merge($plugin['properties']);
-    }
-  }
-
-  $filter_plugins.each |$index, $plugin| {
-    $name = $plugin['name']
+  -> Class['::fluentbit::install']
+  -> Class['::fluentbit::config']
+  ~> Class['::fluentbit::service']
 
-    Resource["fluentbit::filter::${name}"] {
-      "f${index}": * => merge($plugin['properties']);
-    }
-  }
+  create_resources(fluentbit::pipeline, $input_plugins, { type => 'input' })
+  create_resources(fluentbit::pipeline, $output_plugins, { type => 'output' })
+  create_resources(fluentbit::pipeline, $filter_plugins, { type => 'filter' })
+  create_resources(fluentbit::upstream, $upstreams)
 }
diff --git a/manifests/input/forward.pp b/manifests/input/forward.pp
deleted file mode 100644
index 2292a08..0000000
--- a/manifests/input/forward.pp
+++ /dev/null
@@ -1,35 +0,0 @@
-# @summary Inputs / listen and forward messages
-#
-# Forward is the protocol used by Fluent Bit and Fluentd to route messages between peers.
-# This plugin implements the input service to listen for Forward messages.
-#
-# @param configfile
-#  Path to the input configfile. Naming should be input_*.conf to make sure
-#  it's getting included by the main config.
-# @param listen
-#  Listener network interface
-# @param port
-#  TCP port to listen for incoming connections.
-# @param buffer_max_size
-#  Specify the maximum buffer memory size used to receive a Forward message.
-#  The value must be according to the Unit Size specification.
-# @param buffer_chunk_size
-#  By default the buffer to store the incoming Forward messages, do not allocate the maximum memory allowed,
-#  instead it allocate memory when is required. The rounds of allocations are set by Buffer_Chunk_Size.
-#  The value must be according to the Unit Size specification.
-#
-define fluentbit::input::forward (
-    String $configfile                    = "/etc/td-agent-bit/input_forward_${name}.conf",
-    Stdlib::IP::Address::Nosubnet $listen = '0.0.0.0',
-    Stdlib::Port $port                    = 24224,
-    Optional[String] $buffer_max_size     = undef,
-    Optional[String] $buffer_chunk_size   = undef,
-) {
-  file { $configfile:
-    ensure  => file,
-    mode    => $fluentbit::config_file_mode,
-    content => template('fluentbit/input/forward.conf.erb'),
-    notify  => Class['fluentbit::service'],
-    require => Class['fluentbit::install'],
-  }
-}
diff --git a/manifests/input/syslog.pp b/manifests/input/syslog.pp
deleted file mode 100644
index fb11c9b..0000000
--- a/manifests/input/syslog.pp
+++ /dev/null
@@ -1,60 +0,0 @@
-# @summary Ingest logs from (r)syslog
-#
-# Setups config file for fluentbit and configures listen
-# on rsyslog. Reloads services if something changes.
-#
-# @param mode
-#  Defines transport protocol mode: unix_udp (UDP over Unix socket), unix_tcp (TCP over Unix socket), tcp or udp
-# @param listen
-#  If Mode is set to tcp, specify the network interface to bind.
-# @param port
-#  If Mode is set to tcp, specify the TCP port to listen for incoming connections.
-# @param path
-#  If Mode is set to unix_tcp or unix_udp, set the absolute path to the Unix socket file.
-# @param parser
-#  Specify an alternative parser for the message. By default, the plugin uses the parser syslog-rfc3164.
-#  If your syslog messages have fractional seconds set this Parser value to syslog-rfc5424 instead.
-# @param rsyslog_config
-#  Specify the path to the rsyslog config file for fluentbit to enable listening.
-# @param configfile
-#  Path to the input configfile. Naming should be input_*.conf to make sure
-#  it's getting included by the main config.
-#
-define fluentbit::input::syslog(
-  String $mode = 'unix_udp',
-  Optional[String] $listen = $mode ? {
-    tcp => '0.0.0.0',
-    default => undef
-  },
-  Optional[String] $port   = $mode ? {
-    tcp => '5140',
-    default => undef,
-  },
-  Optional[String] $path   = $mode ? {
-    unix_tcp => '/tmp/fluent-bit.sock',
-    unix_udp => '/tmp/fluent-bit.sock',
-    default  => undef,
-  },
-  String $parser = 'syslog-rfc3164',
-  String $configfile = '/etc/td-agent-bit/input_syslog.conf',
-  String $rsyslog_config = '/etc/rsyslog.d/60-fluent-bit.conf',
-) {
-  # create input_syslog.conf
-  # TODO: concat for multiple entries
-  file { $configfile:
-    ensure  => file,
-    mode    => '0644',
-    content => template('fluentbit/input/syslog.conf.erb'),
-    notify  => Class['fluentbit::service'],
-    require => Class['fluentbit::install'],
-  }
-  # create rsyslog config and reload service
-  file { $rsyslog_config:
-    ensure  => file,
-    mode    => '0644',
-    notify  => Service['rsyslog'],
-    content => '$ModLoad omuxsock
-$OMUxSockSocket /tmp/fluent-bit.sock
-*.* :omuxsock:',
-  }
-}
diff --git a/manifests/input/tail.pp b/manifests/input/tail.pp
deleted file mode 100644
index 6d3def4..0000000
--- a/manifests/input/tail.pp
+++ /dev/null
@@ -1,120 +0,0 @@
-# @summary Input plugin to monitor one or several text files.
-#
-# The tail input plugin allows to monitor one or several text files.
-# It has a similar behavior like `tail -f` shell command.
-#
-# @param configfile
-#  Path to the input configfile. Naming should be input_*.conf to make sure
-#  it's getting included by the main config.
-#
-# @param path
-#   Pattern specifying a specific log files or multiple ones through the use of common wildcards.
-#
-# @param routing_tag
-#   Set a tag (with regex-extract fields) that will be placed on lines read.
-#
-# @param storage_type
-#   Specify the buffering mechanism to use.
-#   Values: memory, filesystem
-#
-# @param buffer_chunk_size
-#   Set the initial buffer size to read files data.
-#
-# @param buffer_max_size
-#   Set the limit of the buffer size per monitored file.
-#
-# @param path_key
-#   Appends the name of the monitored file as part of the record. The value assigned becomes the key in the map.
-#
-# @param exclude_path
-#   Shell patterns to exclude files matching a certain criteria.
-#
-# @param refresh_interval
-#   The interval of refreshing the list of watched files in seconds.
-#
-# @param rotate_wait
-#   Specify the number of extra time in seconds to monitor a file once is
-#   rotated in case some pending data is flushed.
-#
-# @param ignore_older
-#   Ignores files that have been last modified before this time.
-#
-# @param skip_long_lines
-#   When a monitored file reach it buffer capacity due to a very long line
-#   (Buffer_Max_Size), the default behavior is to stop monitoring that file.
-#   *Skip_Long_Lines* alters that behavior and instructs Fluent Bit to skip
-#   long lines and continue processing other lines that fits into the buffer size.
-#
-# @param db
-#   Specify the database file to keep track of monitored files and offsets.
-#
-# @param db_sync
-#   Set a default synchronization (I/O) method.
-#   Values: Extra, Full, Normal, Off.
-#
-# @param mem_buf_limit
-#   Set a limit of memory that Tail plugin can use when appending data to the Engine.
-#
-# @param key
-#   Name of the key to append any unstructured data to the output as.
-#
-# @param tag_regex
-#   Set a regex to exctract fields from the file.
-#
-# @param multiline
-#   Try to discover multiline messages and use the proper parsers to compose the outgoing messages.
-#
-# @param multiline_flush
-#   Wait period time in seconds to process queued multiline messages.
-#
-# @param parsers
-#   Name of parsers to use for disecting the message.
-#   When *multiline* is enabled, the first array entry becomes the configuration
-#   value for **Parser_Firstline**, with the remaining entry being assigned to
-#   the **Parser_N** attributes.
-#   When *multiline* is disabled, only the first item is used for the *Parser*
-#   configuration value.
-#
-# @param docker_mode
-#   Recombine split Docker log lines before passing them to any parser.
-#
-# @param docker_mode_flush
-#   Wait period time in seconds to flush queued unfinished split lines.
-#
-define fluentbit::input::tail (
-  Stdlib::Absolutepath $path,
-  Stdlib::Absolutepath $configfile                          = "/etc/td-agent-bit/input_tail_${name}.conf",
-  Optional[String[1]] $routing_tag                          = undef,
-  Optional[Enum['memory', 'filesystem']] $storage_type      = undef,
-
-  Optional[Fluentbit::Sizeunit] $buffer_chunk_size          = undef,
-  Optional[Fluentbit::Sizeunit] $buffer_max_size            = undef,
-  Optional[String[1]] $path_key                             = undef,
-  Array[String] $exclude_path                               = [],
-  Optional[Integer[1]] $refresh_interval                    = undef,
-  Optional[Integer[1]] $rotate_wait                         = undef,
-  Optional[Fluentbit::Timeunit] $ignore_older               = undef,
-  Boolean $skip_long_lines                                  = false,
-  Optional[Stdlib::Absolutepath] $db                        = undef,
-  Optional[Enum['Extra', 'Full', 'Normal', 'Off']] $db_sync = undef,
-  Optional[Fluentbit::Sizeunit] $mem_buf_limit              = undef,
-  Optional[String[1]] $key                                  = undef,
-  Optional[String[1]] $tag_regex                            = undef,
-
-  Boolean $multiline                                        = false,
-  Optional[Integer[1]] $multiline_flush                     = undef,
-  Array[String[1]] $parsers                                 = [],
-
-  Boolean $docker_mode                                      = false,
-  Optional[Integer[1]] $docker_mode_flush                   = undef,
-) {
-  $skip_long_lines_string = bool2str($skip_long_lines, 'On', 'Off')
-  $docker_mode_string = bool2str($docker_mode, 'On', 'Off')
-
-  file { $configfile:
-    ensure  => file,
-    mode    => $fluentbit::config_file_mode,
-    content => template('fluentbit/input/tail.conf.erb'),
-    notify  => Class['fluentbit::service'],
-  }
-}
diff --git a/manifests/install.pp b/manifests/install.pp
index 48ca991..5a71b7f 100644
--- a/manifests/install.pp
+++ b/manifests/install.pp
@@ -4,7 +4,7 @@
 class fluentbit::install {
   assert_private()
 
-  package{ 'fluentbit':
+  package { 'fluentbit':
     ensure => $fluentbit::package_ensure,
     name   => $fluentbit::package_name,
   }
diff --git a/manifests/output/es.pp b/manifests/output/es.pp
deleted file mode 100644
index 1d6bffd..0000000
--- a/manifests/output/es.pp
+++ /dev/null
@@ -1,86 +0,0 @@
-# @summary Plugin to output logs to a configured elasticsearch instance
-#
-# @param configfile
-#  Path to the output configfile. Naming should be output_*.conf to make sure
-#  it's getting included by the main config.
-# @param match
-#  Tag to route the output.
-# @param tls
-#  Turn TLS encrypted communication with Elasticsearch on / off.
-# @param host
-#  IP address or hostname of the target Elasticsearch instance
-# @param port
-#  TCP port of the target Elasticsearch instance
-# @param path
-#  Elasticsearch accepts new data on HTTP query path "/_bulk".
-#  But it is also possible to serve Elasticsearch behind a reverse proxy on a subpath.
-#  This option defines such path on the fluent-bit side.
-#  It simply adds a path prefix in the indexing HTTP POST URI.
-# @param buffer_size
-#  Specify the buffer size used to read the response from the Elasticsearch HTTP service.
-#  This option is useful for debugging purposes where is required to read full responses,
-#  note that response size grows depending of the number of records inserted.
-#  To set an unlimited amount of memory set this value to False,
-#  otherwise the value must be according to the Unit Size specification.
-# @param pipeline
-#  Newer versions of Elasticsearch allows to setup filters called pipelines.
-#  This option allows to define which pipeline the database should use.
-#  For performance reasons is strongly suggested to do parsing and filtering
-#  on Fluent Bit side, avoid pipelines.
-# @param http_user
-#  Optional username credential for Elastic X-Pack access
-# @param http_passwd
-#  Password for user defined in HTTP_User
-# @param index
-#  Index name
-# @param type
-#  Type name
-# @param logstash_format
-#  Enable Logstash format compatibility.
-# @param logstash_prefix
-#  When Logstash_Format is enabled, the Index name is composed using a prefix
-#  and the date, e.g: If Logstash_Prefix is equals to 'mydata' your index will
-#  become 'mydata-YYYY.MM.DD'. The last string appended belongs to the date
-#  when the data is being generated.
-# @param logstash_dateformat
-#  Time format (based on strftime) to generate the second part of the Index name.
-# @param time_key
-#  When Logstash_Format is enabled, each record will get a new timestamp field.
-#  The Time_Key property defines the name of that field.
-# @param time_key_format
-#  When Logstash_Format is enabled, this property defines the format of the timestamp.
-# @param include_tag_key
-#  When enabled, it append the Tag name to the record.
-# @param tag_key
-#  When Include_Tag_Key is enabled, this property defines the key name for the tag.
-# @param generate_id
-#  When enabled, generate _id for outgoing records. This prevents duplicate records when retrying ES.
-# @param replace_dots
-#  When enabled, replace field name dots with underscore, required by Elasticsearch 2.0-2.3.
-# @param trace_output
-#  When enabled print the elasticsearch API calls to stdout (for diag only)
-# @param current_time_index
-#  Use current time for index generation instead of message record
-# @param logstash_prefix_key
-#  Prefix keys with this string
-# @example
-#  include fluentbit::output::es
-define fluentbit::output::es (
-  String $configfile            = "/etc/td-agent-bit/output_es_${name}.conf",
-  String $match                 = '*',
-  Stdlib::Host $host            = '127.0.0.1',
-  Stdlib::Port $port            = 9200,
-  String $index                 = 'fluentbit',
-  String $type                  = 'flb_type',
-  Enum['on', 'off'] $tls        = 'off',
-  Optional[String] $http_user   = undef,
-  Optional[String] $http_passwd = undef,
-) {
-  file { $configfile:
-    ensure  => file,
-    mode    => $fluentbit::config_file_mode,
-    content => template('fluentbit/output/es.conf.erb'),
-    notify  => Class['fluentbit::service'],
-    require => Class['fluentbit::install'],
-  }
-}
diff --git a/manifests/output/http.pp b/manifests/output/http.pp
deleted file mode 100644
index ba4e1ab..0000000
--- a/manifests/output/http.pp
+++ /dev/null
@@ -1,127 +0,0 @@
-# @summary Manage the configuration of a HTTP output plugin.
-#
-# The http output plugin allows to flush your records into a HTTP endpoint.
-# This type manages the configuration for it.
-#
-# @param configfile
-#  Path to the output configfile. Naming should be output_*.conf to make sure
-#  it's getting included by the main config.
-#
-# @param match
-#  Tag to route the output.
-#
-# @param tls
-#   TLS configuration. By default TLS is disabled.
-#
-# @param headers
-#   Map of headers to add to requests.
-#
-# @param retry_limit
-#   Number of retries if upstream refuses the records.
-#   *false* will disable retries, *true* will cause it to retry once.
-#   All other values are passed on verbatim.
-#
-# @param host
-#   IP address or hostname of the target HTTP Server.
-#
-# @param port
-#   TCP port of the target HTTP Server.
-#
-# @param http_user
-#   Basic Auth Username.
-#
-# @param http_passwd
-#   Basic Auth Password.
-#   Requires HTTP_User to be set.
-#
-# @param proxy
-#   Specify an HTTP Proxy.
-#
-# @param uri
-#   Specify an optional HTTP URI for the target web server.
-#
-# @param format
-#   Specify the data format to be used in the HTTP request body.
-#
-# @param header_tag
-#   Specify an optional HTTP header field for the original message tag.
-#
-# @param json_date_key
-#   Specify the name of the date field in output.
-#
-# @param json_date_format
-#   Specify the format of the date.
-#
-# @param gelf_timestamp_key
-#   Specify the key to use for *timestamp* in **gelf** format.
-#
-# @param gelf_host_key
-#   Specify the key to use for *host* in **gelf** format.
-#
-# @param gelf_short_messge_key
-#   Specify the key to use for *short* in **gelf** format.
-#
-# @param gelf_full_message_key
-#   Specify the key to use for *full* in **gelf** format.
-#
-# @param gelf_level_key
-#   Specify the key to use for *level* in **gelf** format.
-#
-# @example
-#   fluentbit::output::http { 'logstash': }
-define fluentbit::output::http (
-  Stdlib::Absolutepath $configfile            = "/etc/td-agent-bit/output_http_${name}.conf",
-  Fluentbit::TLS $tls                         = {},
-  Hash[String[1], String[1]] $headers         = {},
-  Variant[Undef, Boolean, Integer[1]]
-    $retry_limit                              = undef,
-  Optional[String[1]] $match                  = undef,
-  Optional[Stdlib::Host] $host                = undef,
-  Optional[Stdlib::Port] $port                = undef,
-  Optional[String[1]] $http_user              = undef,
-  Optional[String[1]] $http_passwd            = undef,
-  Optional[Stdlib::HTTPUrl] $proxy            = undef,
-  Optional[Stdlib::Absolutepath] $uri         = undef,
-  Optional[Enum['msgpack', 'json', 'json_stream', 'json_lines', 'gelf']]
-    $format                                   = undef,
-  Optional[String[1]] $header_tag             = undef,
-  Optional[String[1]] $json_date_key          = undef,
-  Optional[Enum['double', 'iso8601']]
-    $json_date_format                         = undef,
-  Optional[String[1]] $gelf_timestamp_key     = undef,
-  Optional[String[1]] $gelf_host_key          = undef,
-  Optional[String[1]] $gelf_short_messge_key  = undef,
-  Optional[String[1]] $gelf_full_message_key  = undef,
-  Optional[String[1]] $gelf_level_key         = undef,
-) {
-
-  if empty($http_user) and !empty($http_passwd) {
-    fail('HTTP username must be provided when basic authentication is used')
-  }
-
-  $tls_enabled = $tls['enabled'] ? {
-    undef   => undef,
-    default => bool2str($tls['enabled'], 'On', 'Off'),
-  }
-  $tls_verify = $tls['verify'] ? {
-    undef   => undef,
-    default => bool2str($tls['verify'], 'On', 'Off'),
-  }
-  $tls_debug = $tls['debug'] ? {
-    undef   => undef,
-    default => bool2str($tls['debug'], 'On', 'Off'),
-  }
-  $tls_ca_file = $tls['ca_file']
-  $tls_ca_path = $tls['ca_path']
-  $tls_crt_file = $tls['crt_file']
-  $tls_key_file = $tls['key_file']
-  $tls_key_passwd = $tls['key_passwd']
-  $tls_vhost = $tls['vhost']
-
-  file { $configfile:
-    ensure  => file,
-    mode    => $fluentbit::config_file_mode,
-    content => template('fluentbit/output/http.conf.erb'),
-    notify  => Class['fluentbit::service'],
-  }
-}
diff --git a/manifests/pipeline.pp b/manifests/pipeline.pp
new file mode 100644
index 0000000..5699d84
--- /dev/null
+++ b/manifests/pipeline.pp
@@ -0,0 +1,60 @@
+# @summary Configures fluentbit pipeline (input, output or filter)
+#
+# @note This resource add extra configuration elements for some combinations of
+#  type-plugin_names, like db configuration for input plugins, or upstream configuration
+#  for output-forward plugin
+#
+# @example
+#   fluentbit::pipeline { 'input-dummy':
+#     type        => 'input',
+#     plugin_name => 'dummy',
+#   }
+# @example
+#   fluentbit {
+#     input_plugins => {
+#       'input-dummy' => { 'plugin_name' => 'dummy' },
+#     },
+#   }
+# @see https://docs.fluentbit.io/manual/administration/configuring-fluent-bit/yaml/configuration-file#config_pipeline
+#
+# @param type Defines the pipeline type to be configured
+# @param plugin_name fluent-bit plugin name to be used
+# @param properties Hash of rest of properties needed to configure the pipeline-plugin
+define fluentbit::pipeline (
+  Enum['input','filter','output'] $type,
+  String[1]                       $plugin_name,
+  Hash[String, Any]               $properties = {},
+) {
+  $db_settings = $type ? {
+    'input' => {
+      'db' => "${fluentbit::data_dir}/${title}",
+    },
+    default => {},
+  }
+
+  if $type == 'output' and $plugin_name == 'forward' {
+    $upstream_settings = $properties['upstream'] ? {
+      undef      => {},
+      default    => {
+        upstream => "${fluentbit::config::config_dir}/upstream-${properties['upstream']}.conf",
+      },
+    }
+  } else {
+    $upstream_settings = {}
+  }
+  file { "${fluentbit::config::plugin_dir}/${title}.conf":
+    mode    => $fluentbit::config_file_mode,
+    notify  => Service[$fluentbit::service_name],
+    content => epp('fluentbit/pipeline.conf.epp',
+      {
+        name       => $plugin_name,
+        type       => $type,
+        properties => merge(
+          $db_settings,
+          $properties,
+          $upstream_settings,
+        )
+      }
+    ),
+  }
+}
diff --git a/manifests/repo/debian.pp b/manifests/repo/debian.pp
index 4f43000..592b1d2 100644
--- a/manifests/repo/debian.pp
+++ b/manifests/repo/debian.pp
@@ -11,17 +11,20 @@
   $release = dig($facts, 'os', 'distro', 'codename')
   $supported = $flavour ? {
     'Debian' => [
-      'jessie',
       'stretch',
       'buster',
+      'bullseye',
+      'bookworm',
     ],
     'Ubuntu' => [
       'xenial',
       'bionic',
     ],
     'Raspbian' => [
-      'jessie',
       'stretch',
+      'buster',
+      'bullseye',
+      'bookworm',
     ],
     default => [],
   }
diff --git a/manifests/service.pp b/manifests/service.pp
index ae63911..f33a0c1 100644
--- a/manifests/service.pp
+++ b/manifests/service.pp
@@ -10,7 +10,8 @@
       enable     => $fluentbit::service_enable,
       hasstatus  => $fluentbit::service_has_status,
       hasrestart => $fluentbit::service_has_restart,
-      name       => $::fluentbit::service_name,
+      restart    => $fluentbit::service_restart_command,
+      name       => $fluentbit::service_name,
     }
   }
 }
diff --git a/manifests/upstream.pp b/manifests/upstream.pp
new file mode 100644
index 0000000..d5b18b3
--- /dev/null
+++ b/manifests/upstream.pp
@@ -0,0 +1,28 @@
+# @summary Configures fluentbit upstream
+#
+# @example
+#   fluentbit::upstream { 'upstreams':
+#     nodes  => {
+#       'n1' => { 'host' => '127.0.0.1', port => 1234 },
+#       'n2' => { 'host' => '127.0.0.1', port => 1235 },
+#     },
+#   }
+# @see https://docs.fluentbit.io/manual/administration/configuring-fluent-bit/classic-mode/upstream-servers
+#
+# @param upstream_name Name to be assigned to upstream, defaults to resource namevar
+# @param nodes Hash of nodes assigned to this upstream
+define fluentbit::upstream (
+  Hash[String, Fluentbit::UpstreamNode] $nodes,
+  String[1] $upstream_name = $name,
+) {
+  file { "${fluentbit::config::config_dir}/upstream-${upstream_name}.conf":
+    ensure  => present,
+    mode    => $fluentbit::config_file_mode,
+    content => epp('fluentbit/upstream.conf.epp',
+      {
+        'name'  => $upstream_name,
+        'nodes' => $nodes,
+      },
+    ),
+  }
+}
diff --git a/metadata.json b/metadata.json
index a39ff26..db1d891 100644
--- a/metadata.json
+++ b/metadata.json
@@ -15,14 +15,20 @@
     {
       "name": "puppetlabs/apt",
       "version_requirement": ">= 4.5.1 < 7.0.0"
+    },
+    {
+      "name": "puppet/systemd",
+      "version_requirement": ">= 2.0.0 < 6.0.0"
     }
   ],
   "operatingsystem_support": [
     {
       "operatingsystem": "Debian",
       "operatingsystemrelease": [
-        "8",
-        "9"
+        "9",
+        "10",
+        "11",
+        "12"
       ]
     },
     {
diff --git a/templates/filter/modify.conf.erb b/templates/filter/modify.conf.erb
deleted file mode 100644
index 0d1a87b..0000000
--- a/templates/filter/modify.conf.erb
+++ /dev/null
@@ -1,48 +0,0 @@
-# Managed by puppet
-# Filter to modify output
-[FILTER]
-    # configure plugins
-    Name  modify<%- if @match %>
-    Match <%= @match -%>
-<% end -%>
-<% if @set %>
-<% @set.each do |key, value| %>
-    Set <%= key %> <%= value -%>
-<% end -%>
-<% end -%>
-<% if @add -%>
-<%- @add.each do |key, value| %>
-    Add <%= key %> <%= value -%>
-<% end -%>
-<% end -%>
-<% if @remove -%>
-<%- @remove.each do |key, value| %>
-    Remove <%= key %> <%= value -%>
-<% end -%>
-<% end -%>
-<%- if @remove_wildcard %>
-    Remove_wildcard <%= @remove_wildcard -%>
-<% end -%>
-<%- if @remove_regex %>
-    Remove_regex <%= @remove_regex -%>
-<% end -%>
-<% if @rename -%>
-<%- @rename.each do |key, value| %>
-    Rename <%= key %> <%= value -%>
-<% end -%>
-<% end -%>
-<% if @hard_rename -%>
-<%- @hard_rename.each do |key, value| %>
-    Hard_rename <%= key %> <%= value -%>
-<% end -%>
-<% end -%>
-<% if @copy -%>
-<%- @copy.each do |key, value| %>
-    Copy <%= key %> <%= value -%>
-<% end -%>
-<% end -%>
-<% if @hard_copy -%>
-<%- @hard_copy.each do |key, value| %>
-    Hard_copy <%= key %> <%= value -%>
-<% end -%>
-<% end -%>
diff --git a/templates/fluent-bit.conf.epp b/templates/fluent-bit.conf.epp
new file mode 100644
index 0000000..e9192e4
--- /dev/null
+++ b/templates/fluent-bit.conf.epp
@@ -0,0 +1,38 @@
+# Managed by puppet
+<% $variables.each |$key, $value| { -%>
+@SET <%= $key %>=<%= $value %>
+<% } -%>
+
+[SERVICE]
+    Flush                     <%= $service['flush'] %>
+    Daemon                    <%= $service['daemon'] %>
+    Grace                     <%= $service['grace'] %>
+    dns.mode                  <%= $service['dns.mode'] %>
+    Log_Level                 <%= $service['log_level'] %>
+    Parsers_File              <%= $service['parsers_file'] %>
+    Plugins_File              <%= $service['plugins_file'] %>
+    Streams_file              <%= $service['streams_file'] %>
+    HTTP_Server               <%= $service['http_server'] %>
+    HTTP_Listen               <%= $service['http_listen'] %>
+    HTTP_Port                 <%= $service['http_port'] %>
+    Coro_Stack_Size           <%= $service['coro_stack_size'] %>
+    scheduler.cap             <%= $service['scheduler.cap'] %>
+    scheduler.base            <%= $service['scheduler.base'] %>
+    json.convert_nan_to_null  <%= $service['json.convert_nan_to_null'] %>
+<% if $service['storage.checksum'] { %>
+    storage.checksum          <%= $service['storage.checksum'] %>
+<% } -%>
+<% if $service['storage.path'] { %>
+    storage.path              <%= $service['storage.path'] %>
+<% } -%>
+<% if $service['storage.sync'] { %>
+    storage.sync              <%= $service['storage.sync'] %>
+<% } -%>
+<% if $service['storage.backlog_mem_limit'] { %>
+    storage.backlog.mem_limit <%= $service['storage.backlog_mem_limit'] %>
+<% } -%>
+    storage.max_chunks_up     <%= $service['storage.max_chunks_up'] %>
+    storage.metrics           <%= $service['storage.metrics'] %>
+    storage.delete_irrecoverable_chunks <%= $service['storage.delete_irrecoverable_chunks'] %>
+
+@INCLUDE pipelines/*.conf
diff --git a/templates/fluent-bit.service.epp b/templates/fluent-bit.service.epp
new file mode 100644
index 0000000..b3a7459
--- /dev/null
+++ b/templates/fluent-bit.service.epp
@@ -0,0 +1,16 @@
+[Unit]
+Description=Fluent Bit
+Documentation=https://docs.fluentbit.io/manual/
+Requires=network.target
+After=network.target
+
+[Service]
+Type=simple
+EnvironmentFile=-/etc/sysconfig/fluent-bit
+EnvironmentFile=-/etc/default/fluent-bit
+ExecStart=<%= $binary_file %> -c <%= $config_file %> --enable-hot-reload
+ExecReload=/bin/kill -HUP $MAINPID
+Restart=always
+
+[Install]
+WantedBy=multi-user.targe
diff --git a/templates/input/forward.conf.erb b/templates/input/forward.conf.erb
deleted file mode 100644
index aca9bd3..0000000
--- a/templates/input/forward.conf.erb
+++ /dev/null
@@ -1,15 +0,0 @@
-# Managed by puppet
-[INPUT]
-    Name     forward
-<%- if @listen -%>
-    Listen   <%= @listen %>
-<% end -%>
-<% if @port -%>
-    Port      <%= @port %>
-<% end -%>
-<% if @buffer_max_size -%>
-    Buffer_Max_Size   <%= @buffer_max_size %>
-<% end -%>
-<% if @buffer_chunk_size -%>
-    Buffer_Chunk_Size <%= @buffer_chunk_size %>
-<% end -%>
diff --git a/templates/input/syslog.conf.erb b/templates/input/syslog.conf.erb
deleted file mode 100644
index 8e8a72b..0000000
--- a/templates/input/syslog.conf.erb
+++ /dev/null
@@ -1,13 +0,0 @@
-# Managed by puppet
-# config for syslog input plugin
-[INPUT]
-    # configure plugins
-    Name     syslog<%- if @parser %>
-    Parser   <%= @parser -%>
-    <% end -%>
-    <% if @path %>
-    Path     <%= @path -%>
-    <% end -%>
-    <% if @mode %>
-    Mode     <%= @mode -%>
-    <% end -%>
diff --git a/templates/input/tail.conf.erb b/templates/input/tail.conf.erb
deleted file mode 100644
index e0b7795..0000000
--- a/templates/input/tail.conf.erb
+++ /dev/null
@@ -1,77 +0,0 @@
-# Managed by puppet
-[INPUT]
-    Name              tail
-<% if @routing_tag -%>
-    Tag               <%= @routing_tag %>
-<% end -%>
-<% if @storage_type -%>
-    storage.type      <%= @storage_type %>
-<% end -%>
-<% if @buffer_max_size -%>
-    Buffer_Max_Size   <%= @buffer_max_size %>
-<% end -%>
-<% if @buffer_chunk_size -%>
-    Buffer_Chunk_Size <%= @buffer_chunk_size %>
-<% end -%>
-<% if @path -%>
-    Path              <%= @path %>
-<% end -%>
-<% if @path_key -%>
-    Path_Key          <%= @path_key %>
-<% end -%>
-<% if @refresh_interval -%>
-    Refresh_Interval  <%= @refresh_interval %>
-<% end -%>
-<% if @rotate_wait -%>
-    Rotate_Wait       <%= @rotate_wait %>
-<% end -%>
-<% if @ignore_older -%>
-    Ignore_Older      <%= @ignore_older %>
-<% end -%>
-<% if @skip_long_lines_string -%>
-    Skip_Long_Lines   <%= @skip_long_lines_string %>
-<% end -%>
-<% if @db -%>
-    DB                <%= @db %>
-<% end -%>
-<% if @db_sync -%>
-    DB.Sync           <%= @db_sync %>
-<% end -%>
-<% if @mem_buf_limit -%>
-    Mem_Buf_Limit     <%= @mem_buf_limit %>
-<% end -%>
-<% if @key -%>
-    Key               <%= @key %>
-<% end -%>
-<% if @tag -%>
-    Tag               <%= @tag %>
-<% end -%>
-<% if @tag_regex -%>
-    Tag_Regex         <%= @tag_regex %>
-<% end -%>
-<% if @multiline_flush -%>
-    Multiline_Flush   <%= @multiline_flush %>
-<% end -%>
-<% if @docker_mode_string -%>
-    Docker_Mode       <%= @docker_mode_string %>
-<% end -%>
-<% if @docker_mode_flush -%>
-    Docker_Mode_Flush <%= @docker_mode_flush %>
-<% end -%>
-<% unless @exclude_path.empty? -%>
-    Exclude_Path      <%= @exclude_path.join(',') %>
-<% end -%>
-<% if @multiline -%>
-    Multiline         On
-  <%- unless @parsers.empty? -%>
-    Parser_Firstline  <%= @parsers.shift %>
-    <%- @parsers.each_with_index do |parser, index| -%>
-    Parser_<%= (index + 1) -%>  <%= parser %>
-    <%- end -%>
-  <%- end -%>
-<% else -%>
-    Multiline         Off
-  <%- unless @parsers.empty? -%>
-    Parser            <%= @parsers.first %>
-  <%- end -%>
-<% end -%>
diff --git a/templates/output/es.conf.erb b/templates/output/es.conf.erb
deleted file mode 100644
index dbf526b..0000000
--- a/templates/output/es.conf.erb
+++ /dev/null
@@ -1,27 +0,0 @@
-# Managed by puppet
-[OUTPUT]
-    Name  es
-<%- if @match -%>
-    Match <%= @match %>
-<% end -%>
-<% if @host -%>
-    Host  <%= @host %>
-<% end -%>
-<% if @port -%>
-    Port  <%= @port %>
-<% end -%>
-<% if @index -%>
-    Index <%= @index %>
-<% end -%>
-<% if @type -%>
-    Type  <%= @type %>
-<% end -%>
-<% if @tls -%>
-    tls   <%= @tls %>
-<% end -%>
-<% if @http_user -%>
-    HTTP_User <%= @http_user %>
-<% end -%>
-<% if @http_passwd -%>
-    HTTP_Passwd <%= @http_passwd %>
-<% end -%>
diff --git a/templates/output/http.conf.erb b/templates/output/http.conf.erb
deleted file mode 100644
index 77b4b20..0000000
--- a/templates/output/http.conf.erb
+++ /dev/null
@@ -1,88 +0,0 @@
-# Managed by puppet
-[OUTPUT]
-    Name                   http
-<% if @match -%>
-    Match                  <%= @match %>
-<% end -%>
-<% if @host -%>
-    Host                   <%= @host %>
-<% end -%>
-<% if @http_user -%>
-    HTTP_User              <%= @http_user %>
-<% end -%>
-<% if @http_passwd -%>
-    HTTP_Passwd            <%= @http_passwd %>
-<% end -%>
-<% if @port -%>
-    Port                   <%= @port %>
-<% end -%>
-<% if @proxy -%>
-    Proxy                  <%= @proxy %>
-<% end -%>
-<% if @uri -%>
-    URI                    <%= @uri %>
-<% end -%>
-<% if @format -%>
-    Format                 <%= @format %>
-<% end -%>
-<% if @header_tag -%>
-    header_tag             <%= @header_tag %>
-<% end -%>
-<% @headers.each do |key, value| -%>
-    Header                 <%= key %> <%= value %>
-<% end -%>
-<% if @json_date_key -%>
-    json_date_key          <%= @json_date_key %>
-<% end -%>
-<% if @json_date_format -%>
-    json_date_format       <%= @json_date_format %>
-<% end -%>
-<% if @gelf_timestamp_key -%>
-    gelf_timestamp_key     <%= @gelf_timestamp_key %>
-<% end -%>
-<% if @gelf_host_key -%>
-    gelf_host_key          <%= @gelf_host_key %>
-<% end -%>
-<% if @gelf_short_messge_key -%>
-    gelf_short_messge_key  <%= @gelf_short_messge_key %>
-<% end -%>
-<% if @gelf_full_message_key -%>
-    gelf_full_message_key  <%= @gelf_full_message_key %>
-<% end -%>
-<% if @gelf_level_key -%>
-    gelf_level_key         <%= @gelf_level_key %>
-<% end -%>
-<% if @tls_enabled -%>
-    tls                    <%= @tls_enabled %>
-<% end -%>
-<% if @tls_verify -%>
-    tls.verify             <%= @tls_verify %>
-<% end -%>
-<% if @tls_debug -%>
-    tls.debug              <%= @tls_debug %>
-<% end -%>
-<% if @tls_ca_file -%>
-    tls.ca_file            <%= @tls_ca_file %>
-<% end -%>
-<% if @tls_ca_path -%>
-    tls.ca_path            <%= @tls_ca_path %>
-<% end -%>
-<% if @tls_crt_file -%>
-    tls.crt_file           <%= @tls_crt_file %>
-<% end -%>
-<% if @tls_key_file -%>
-    tls.key_file           <%= @tls_key_file %>
-<% end -%>
-<% if @tls_key_passwd -%>
-    tls.key_passwd         <%= @tls_key_passwd %>
-<% end -%>
-<% if @tls_vhost -%>
-    tls.vhost              <%= @tls_vhost %>
-<% end -%>
-<% if @retry_limit == false -%>
-    Retry_Limit            False
-<% elsif @retry_limit == true -%>
-    Retry_Limit            1
-<% elsif @retry_limit.is_a?(Integer) -%>
-    Retry_Limit            <%= @retry_limit %>
-<% end -%>
diff --git a/templates/parsers.conf.epp b/templates/parsers.conf.epp
new file mode 100644
index 0000000..620ae12
--- /dev/null
+++ b/templates/parsers.conf.epp
@@ -0,0 +1,35 @@
+# Managed by puppet
+<% $parsers.each |$name, $parser| { -%>
+[PARSER]
+    Name          <%= $name %>
+<% if $parser['format'] { -%>
+    Format        <%= $parser['format'] %>
+<% } -%>
+<% if $parser['regex'] { -%>
+    Regex         <%= $parser['regex'] %>
+<% } -%>
+<% if $parser['time_key'] { -%>
+    Time_key      <%= $parser['time_key'] %>
+<% } -%>
+<% if $parser['time_format'] { -%>
+    Time_format   <%= $parser['time_format'] %>
+<% } -%>
+<% if $parser['time_offset'] { -%>
+    Time_offset   <%= $parser['time_offset'] %>
+<% } -%>
+<% if $parser['time_keep'] == true { -%>
+    Time_keep     On
+<% } elsif $parser['time_keep'] == false { -%>
+    Time_keep     Off
+<% } -%>
+<% if $parser['types'] { -%>
+    Types         <%= $parser['types'].map |$type| { [$type['field'], $type['type']].join(':') }.join(' ') %>
+<% } -%>
+<% if $parser['decode_field'] { -%>
+<% $parser['decode_field'].each |$decode| { -%>
+    Decode_field  <%= $decode['decoder'] %> <%= $decode['field'] %>
+<% } -%>
+<% } -%>
+
+<% } -%>
+# <%= $parsers.length %> parsers
diff --git a/templates/parsers.conf.erb b/templates/parsers.conf.erb
deleted file mode 100644
index 4a12233..0000000
--- a/templates/parsers.conf.erb
+++ /dev/null
@@ -1,35 +0,0 @@
-# Managed by puppet
-<% @parsers.each do |parser| -%>
-[PARSER]
-    Name          <%= parser['name'] %>
-<% if parser['format'] -%>
-    Format        <%= parser['format'] %>
-<% end -%>
-<% if parser['regex'] -%>
-    Regex         <%= parser['regex'] %>
-<% end -%>
-<% if parser['time_key'] -%>
-    Time_key      <%= parser['time_key'] %>
-<% end -%>
-<% if parser['time_format'] -%>
-    Time_format   <%= parser['time_format'] %>
-<% end -%>
-<% if parser['time_offset'] -%>
-    Time_offset   <%= parser['time_offset'] %>
-<% end -%>
-<% if parser['time_keep'] == true -%>
-    Time_keep     On
-<% elsif parser['time_keep'] == false -%>
-    Time_keep     Off
-<% end -%>
-<% if parser['types'] -%>
-    Types         <%= parser['types'].map { |type| [type['field'], type['type']].join(':') }.join(' ') %>
-<% end -%>
-<% if parser['decode_field'] -%>
-<% parser['decode_field'].each do |decode| -%>
-    Decode_field  <%= decode['decoder'] %> <%= decode['field'] %>
-<% end -%>
-<% end -%>
-
-<% end -%>
-# <%= @parsers.length %> parsers
diff --git a/templates/pipeline.conf.epp b/templates/pipeline.conf.epp
new file mode 100644
index 0000000..1bbc661
--- /dev/null
+++ b/templates/pipeline.conf.epp
@@ -0,0 +1,12 @@
+# Managed by Puppet
+[<%= $type.upcase %>]
+  Name  <%= $name %>
+<% $properties.each |$key, $value| { -%>
+  <%- if $value =~ Array { -%>
+    <%- $value.each |$v| { -%>
+  <%= $key.capitalize %> <%= $v %>
+    <%- } -%>
+  <%- } else { -%>
+  <%= $key.capitalize %> <%= $value %>
+  <%- } -%>
+<% } -%>
diff --git a/templates/plugins.conf.epp b/templates/plugins.conf.epp
new file mode 100644
index 0000000..7f8493a
--- /dev/null
+++ b/templates/plugins.conf.epp
@@ -0,0 +1,7 @@
+# Managed by puppet
+[PLUGINS]
+<% $plugins.each |$plugin| { -%>
+    Path <%= $plugin %>
+<% } -%>
+
+# <%= $plugins.length %> plugins
diff --git a/templates/plugins.conf.erb b/templates/plugins.conf.erb
deleted file mode 100644
index eac790e..0000000
--- a/templates/plugins.conf.erb
+++ /dev/null
@@ -1,7 +0,0 @@
-# Managed by puppet
-[PLUGINS]
-<% @plugins.each do |plugin| -%>
-    Path <%= plugin %>
-<%end -%>
-
-# <%= @plugins.length %> plugins
diff --git a/templates/streams.conf.epp b/templates/streams.conf.epp
new file mode 100644
index 0000000..80c37ad
--- /dev/null
+++ b/templates/streams.conf.epp
@@ -0,0 +1,8 @@
+# Managed by puppet
+<% $streams.each |$name, $stream| { -%>
+[STREAM_TASK]
+    Name   <%= $name %>
+    Exec   <%= $stream['exec'] %>
+
+<% } -%>
+# <%= $streams.length %> streams
diff --git a/templates/streams.conf.erb b/templates/streams.conf.erb
deleted file mode 100644
index 8c4ff82..0000000
--- a/templates/streams.conf.erb
+++ /dev/null
@@ -1,8 +0,0 @@
-# Managed by puppet
-<% @streams.each do |stream| -%>
-[STREAM_TASK]
-    Name   <%= stream['name'] %>
-    Exec   <%= stream['exec'] %>
-
-<% end -%>
-# <%= @streams.length %> streams
diff --git a/templates/td-agent-bit.conf.erb b/templates/td-agent-bit.conf.erb
deleted file mode 100644
index 974b4a1..0000000
--- a/templates/td-agent-bit.conf.erb
+++ /dev/null
@@ -1,33 +0,0 @@
-# Managed by puppet
-<% @variables.each do |key, value| -%>
-@SET <%= key %>=<%= value %>
-<% end -%>
-
-[SERVICE]
-    Flush                     <%= @flush %>
-    Daemon                    <%= @daemon %>
-<% if @log_file %>
-    Log_File                  <%= @log_file %>
-<% end -%>
-    Log_Level                 <%= @log_level %>
-    Parsers_File              <%= @parsers_file %>
-    Plugin_File               <%= @plugins_file %>
-    Streams_file              <%= @streams_file %>
-    HTTP_Server               <%= @http_server %>
-    HTTP_Listen               <%= @http_listen %>
-    HTTP_Port                 <%= @http_port %>
-    Coro_Stack_Size           <%= @coro_stack_size %>
-    storage.checksum          <%= @storage_checksum %>
-<% if @storage_path %>
-    storage.path              <%= @storage_path %>
-<% end -%>
-<% if @storage_sync %>
-    storage.sync              <%= @storage_sync %>
-<% end -%>
-<% if @storage_backlog_mem_limit %>
-    storage.backlog.mem_limit <%= @storage_backlog_mem_limit %>
-<% end -%>
-
-@INCLUDE input_*.conf
-@INCLUDE output_*.conf
-@INCLUDE filter_*.conf
diff --git a/templates/upstream.conf.epp b/templates/upstream.conf.epp
new file mode 100644
index 0000000..9c58d2a
--- /dev/null
+++ b/templates/upstream.conf.epp
@@ -0,0 +1,12 @@
+# Managed by puppet
+[UPSTREAM]
+  name  <%= $name %>
+
+<% $nodes.each |$node_name, $node| { -%>
+[NODE]
+  name  <%= $node_name %>
+  host  <%= $node['host'] %>
+  port  <%= $node['port'] %>
+
+<% } -%>
+# <%= $nodes.length %> nodes
diff --git a/types/parser.pp b/types/parser.pp
index cb15c60..5aa9a3b 100644
--- a/types/parser.pp
+++ b/types/parser.pp
@@ -1,5 +1,4 @@
 type Fluentbit::Parser = Struct[{
-  name                   => String[1],
   format                 => Enum['json', 'regex', 'ltsv', 'logfmt'],
   Optional[regex]        => String[1],
   Optional[time_key]     => String[1],
diff --git a/types/plugin.pp b/types/plugin.pp
deleted file mode 100644
index 8a9b744..0000000
--- a/types/plugin.pp
+++ /dev/null
@@ -1,4 +0,0 @@
-type Fluentbit::Plugin = Struct[{
-  name                 => String[1],
-  Optional[properties] => Hash,
-}]
diff --git a/types/stream.pp b/types/stream.pp
index 37d1c02..44f7b2b 100644
--- a/types/stream.pp
+++ b/types/stream.pp
@@ -1,4 +1,3 @@
 type Fluentbit::Stream = Struct[{
-  name => String[1],
   exec => String[1],
 }]
diff --git a/types/upstreamnode.pp b/types/upstreamnode.pp
new file mode 100644
index 0000000..e35c5fd
--- /dev/null
+++ b/types/upstreamnode.pp
@@ -0,0 +1,4 @@
+type Fluentbit::UpstreamNode = Struct[{
+  host => String[1],
+  port => Integer,
+}]