Add support for multiline parsers. Use concat to compose configuratio…

…n files. Split plugin configuration files into inputs, filters and outputs

data/common.yaml

@@ -59,6 +59,7 @@ fluentbit::streams: {}
 
 fluentbit::manage_parsers_file: true
 fluentbit::parsers_file: '/etc/fluent-bit/parsers.conf'
+fluentbit::multiline_parsers: {}
 fluentbit::parsers:
 
   'apache':

manifests/config.pp

@@ -16,6 +16,7 @@
 
   $config_dir = dirname($fluentbit::config_file)
   $plugin_dir = "${config_dir}/pipelines"
+  $scripts_dir = "${config_dir}/lua-scripts"
 
   if $fluentbit::manage_config_dir {
     file { $config_dir:
@@ -30,6 +31,44 @@
       recurse => true,
       mode    => $fluentbit::config_folder_mode,
     }
+    file { $scripts_dir:
+      ensure  => directory,
+      purge   => true,
+      recurse => true,
+      mode    => $fluentbit::config_folder_mode,
+      require => File[$config_dir],
+    }
+
+    Concat {
+      ensure         => present,
+      mode           => $fluentbit::config_file_mode,
+      require        => File[$config_dir],
+      ensure_newline => true,
+    }
+
+    concat {
+      [
+        "${plugin_dir}/inputs.conf",
+        "${plugin_dir}/outputs.conf",
+        "${plugin_dir}/filters.conf",
+      ]:
+    }
+
+    concat::fragment { 'inputs-header':
+      target  => "${plugin_dir}/inputs.conf",
+      content => "# Managed by Puppet\n",
+      order   => '01',
+    }
+    concat::fragment { 'outputs-header':
+      target  => "${plugin_dir}/outputs.conf",
+      content => "# Managed by Puppet\n",
+      order   => '01',
+    }
+    concat::fragment { 'filters-header':
+      target  => "${plugin_dir}/filters.conf",
+      content => "# Managed by Puppet\n",
+      order   => '01',
+    }
   }
 
   if $fluentbit::manage_data_dir {
@@ -89,8 +128,9 @@
     mode    => $fluentbit::config_file_mode,
     content => epp('fluentbit/fluent-bit.conf.epp',
       {
-        variables => $variables,
-        service   => {
+        manage_config_dir => $fluentbit::manage_config_dir,
+        variables         => $variables,
+        service           => {
           'flush'                    => $flush,
           'grace'                    => $grace,
           'daemon'                   => $daemon,
@@ -112,9 +152,15 @@
   }
 
   $parsers = $fluentbit::parsers
+  $multiline_parsers = $fluentbit::multiline_parsers
 
   file { $fluentbit::parsers_file:
-    content => epp('fluentbit/parsers.conf.epp', { parsers => $parsers }),
+    content => epp('fluentbit/parsers.conf.epp',
+      {
+        parsers           => $parsers,
+        multiline_parsers => $multiline_parsers,
+      }
+    ),
   }
 
   $plugins = $fluentbit::plugins

manifests/init.pp

@@ -145,6 +145,11 @@
 #   The default value consists of all the available definitions provided by the
 #   upstream project as of version 1.3
 #
+# @param multiline_parsers
+#   List of parser definitions.
+#   The default value consists of all the available definitions provided by the
+#   upstream project as of version 2.1
+#
 # @param flush
 #   Set the flush time in seconds. Everytime it timeouts, the engine will flush the records to the output plugin.
 # @param grace
@@ -200,6 +205,7 @@
   Hash[String, Hash] $filter_plugins,
 
   Hash[String, Fluentbit::Parser] $parsers,
+  Hash[String, Variant[Fluentbit::MultilineParser]] $multiline_parsers,
   Hash[String, Fluentbit::Stream] $streams,
   Hash[String, Hash] $upstreams,
   Array[Stdlib::Absolutepath] $plugins,

manifests/pipeline.pp

@@ -19,17 +19,22 @@
 #
 # @param type Defines the pipeline type to be configured
 # @param plugin_name fluent-bit plugin name to be used
+# @param order Order to be applied to concat::fragment
 # @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,
+  String[1]                       $order      = '10',
   Hash[String, Any]               $properties = {},
 ) {
-  $db_settings = $type ? {
-    'input' => {
+  $db_compatible_plugins = ['tail', 'systemd']
+
+  if $type == 'input' and $plugin_name in $db_compatible_plugins {
+    $db_settings = {
       'db' => "${fluentbit::data_dir}/${title}",
-    },
-    default => {},
+    }
+  } else {
+    $db_settings = {}
   }
 
   if $type == 'output' and $plugin_name == 'forward' {
@@ -42,9 +47,25 @@
   } else {
     $upstream_settings = {}
   }
-  file { "${fluentbit::config::plugin_dir}/${title}.conf":
-    mode    => $fluentbit::config_file_mode,
-    notify  => Service[$fluentbit::service_name],
+
+  if $type == 'filter' and $plugin_name == 'lua' and $properties['code'] {
+    # Catch 'code' property for lua scripts and write it to disk
+    file { "${fluentbit::config::scripts_dir}/${title}.lua":
+      ensure  => present,
+      mode    => $fluentbit::config_file_mode,
+      content => $properties['code'],
+      notify  => Service[$fluentbit::service_name],
+    }
+    $script_settings = {
+      'script' => "${fluentbit::config::scripts_dir}/${title}.lua",
+      'code'   => undef,
+    }
+  } else {
+    $script_settings = {}
+  }
+
+  concat::fragment { "fragment-${title}":
+    target  => "${fluentbit::config::plugin_dir}/${type}s.conf",
     content => epp('fluentbit/pipeline.conf.epp',
       {
         name       => $plugin_name,
@@ -55,6 +76,7 @@
             alias => $title,
           },
           $properties,
+          $script_settings,
           $upstream_settings,
         )
       }

metadata.json

@@ -19,6 +19,10 @@
     {
       "name": "puppet/systemd",
       "version_requirement": ">= 2.0.0 < 6.0.0"
+    },
+    {
+      "name": "puppetlabs-concat",
+      "version_requirement": ">= 7.0.0 < 10.0.0"
     }
   ],
   "operatingsystem_support": [

templates/fluent-bit.conf.epp

@@ -35,4 +35,6 @@
     storage.metrics           <%= $service['storage.metrics'] %>
     storage.delete_irrecoverable_chunks <%= $service['storage.delete_irrecoverable_chunks'] %>
 
+<%- if $manage_config_dir { %>
 @INCLUDE pipelines/*.conf
+<%- } %>

templates/parsers.conf.epp

@@ -32,4 +32,22 @@
 <% } -%>
 
 <% } -%>
-# <%= $parsers.length %> parsers
+<% $multiline_parsers.each |$name, $parser| { -%>
+[MULTILINE_PARSER]
+    Name          <%= $name %>
+    Type          <%= $parser['type'] %>
+<% if $parser['parser'] { -%>
+    Parser        <%= $parser['time_format'] %>
+<% } -%>
+<% if $parser['key_content'] { -%>
+    Key_content   <%= $parser['key_content'] %>
+<% } -%>
+<% if $parser['flush_timeout'] { -%>
+    Flush_timeout <%= $parser['flush_timeout'] %>
+<% } -%>
+<% $parser['rules'].each |$rule| { -%>
+    Rule  "<%= $rule['state'] %>" "<%= $rule['regex'] %>" "<%= $rule['next_state'] %>"
+<% } -%>
+
+<% } -%>
+# <%= $parsers.length + $multiline_parsers.length %> parsers

templates/pipeline.conf.epp

@@ -1,12 +1,14 @@
-# Managed by Puppet
+
 [<%= $type.upcase %>]
   Name  <%= $name %>
 <% $properties.each |$key, $value| { -%>
+<%- if $value != undef { -%>
   <%- if $value =~ Array { -%>
     <%- $value.each |$v| { -%>
   <%= $key.capitalize %> <%= $v %>
     <%- } -%>
   <%- } else { -%>
   <%= $key.capitalize %> <%= $value %>
   <%- } -%>
+<%- } -%>
 <% } -%>

types/multilineparser.pp

@@ -0,0 +1,7 @@
+type Fluentbit::MultilineParser = Struct[{
+  type                    => Enum['regex'],
+  rules                   => Array[Fluentbit::MultilineParserRule],
+  Optional[parser]        => String[1],
+  Optional[key_content]   => String[1],
+  Optional[flush_timeout] => Integer,
+}]

types/multilineparserrule.pp

@@ -0,0 +1,5 @@
+type Fluentbit::MultilineParserRule = Struct[{
+  state      => String[1],
+  regex      => String[1],
+  next_state => String[1],
+}]