Reworking the processor implementation.

README.md

@@ -16,6 +16,7 @@
       - [JSON logs](#json-logs)
     - [Prospectors in hiera](#prospectors-in-hiera)
     - [Usage on Windows](#usage-on-windows)
+    - [Processors](#processors)
 4. [Reference](#reference)
     - [Public Classes](#public-classes)
     - [Private Classes](#private-classes)
@@ -154,55 +155,64 @@ but is expected to exist as a directory that puppet can write to.
 ### Processors
 
 Filebeat 5.0 and greater includes a new libbeat feature for filtering and/or enhancing all
-exported data through processors before geing sent to the configured output(s). By populating
-the `processors` parameter any number of processors may be configured to work on all events
-or only those that match certain conditions.
+exported data through processors before geing sent to the configured output(s). They can be
+defined as a hash added to the class declaration (also used for automatically creating
+processors using hiera), or as their own defined resources.
 
 To drop the offset and input_type fields from all events:
 
 ```puppet
 class{"filebeat":
-  processors => [
-    {
-      "name" => "drop_fields",
+  processors => {
+    "drop_fields" => {
       "params" => {"fields" => ["input_type", "offset"]}
     },
-  ],
+  },
 }
 ```
 
 To drop all events that have the http response code equal to 200:
 
 ```puppet
 class{"filebeat":
-  processors => [
-    {
-      "name" => "drop_event",
+  processors => {
+    "drop_event" => {
       "when" => {"equals" => {"http.code" => 200}}
     },
-  ],
+  },
 }
 ```
 
 Now to combine these examples into a single definition:
 
 ```puppet
 class{"filebeat":
-  processors => [
-    {
-      "name" => "drop_fields",
-      "params" => {"fields" => ["input_type", "offset"]}
+  processors => {
+    "drop_fields" => {
+      "params"   => {"fields" => ["input_type", "offset"]},
+      "priority" => 1,
     },
-    {
-      "name" => "drop_event",
-      "when" => {"equals" => {"http.code" => 200}}
+    "drop_event" => {
+      "when"     => {"equals" => {"http.code" => 200}},
+      "priority: => 2,
     },
-  ],
+  },
 }
 ```
 
 For more information please review the documentation [here](https://www.elastic.co/guide/en/beats/filebeat/5.1/configuration-processors.html).
 
+#### Processors in Hiera
+
+Processors can be declared in hiera using the `processors` parameter. By default, hiera will not merge
+processor declarations down the hiera hierarchy. To change the behavior in puppet 3 use the `processors_merge`
+parameter. In puppet 4, you can use `processors_merge`, but can also use the
+[lookup_options](https://docs.puppet.com/puppet/latest/reference/lookup_quick.html#setting-lookupoptions-in-data)
+flag.
+
+When `processors_merge` is set to true, `processors` will be replaced by the output of
+`hiera_hash('filebeat::processors')`.
+
 ## Reference
  - [**Public Classes**](#public-classes)
     - [Class: filebeat](#class-filebeat)
@@ -216,6 +226,7 @@ For more information please review the documentation [here](https://www.elastic.
     - [Class: filebeat::install::windows](#class-filebeatinstallwindows)
  - [**Public Defines**](#public-defines)
     - [Define: filebeat::prospector](#define-filebeatprospector)
+    - [Define: filebeat::processors](#define-filebeatprocessor)
 
 ### Public Classes
 
@@ -332,6 +343,23 @@ to fully understand what these parameters do.
   - `multiline`: [Hash] Options that control how Filebeat handles log messages that span multiple lines.
     [See above](#multiline-logs). (default: {})
 
+#### Define: `filebeat::processor`
+
+Installs a configuration file for a processor.
+
+Be sure to read the [processor configuration details](https://www.elastic.co/guide/en/beats/filebeat/current/configuration-processors.html)
+to fully understand what these parameters do.
+
+**Parameters for `filebeat::processor`**
+  - `ensure`: The ensure parameter on the prospector configuration file. (default: present)
+  - `priority`: [Integer] Used to help alpha-numerically sort the processor files in the config dir
+  location. (default: 10)
+  - `processor_name`: [String] The name of the processor. (default: $name)
+  - `params`: [Hash] The key-value pairs to pass to the processor as parameters. Not required on all
+  processors, review documentation for more details. (default: undef)
+  - `when`: Optional[Hash] Run this processor on any event that matches these conditions. Populates
+  the `when` option. (default: undef)
+
 
 ## Limitations
 This module doesn't load the [elasticsearch index template](https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-getting-started.html#filebeat-template) into elasticsearch (required when shipping

manifests/init.pp

@@ -42,7 +42,7 @@
 # @param max_procs [Number] The maximum number of CPUs that can be simultaneously used
 # @param fields [Hash] Optional fields that should be added to each event output
 # @param fields_under_root [Boolean] If set to true, custom fields are stored in the top level instead of under fields
-# @param processors [Array] An optional list of hashes used to configure filebeat processors
+# @param processors [Hash] Processors that will be added. Commonly used to create processors using hiera.
 # @param prospectors [Hash] Prospectors that will be created. Commonly used to create prospectors using hiera
 # @param prospectors_merge [Boolean] Whether $prospectors should merge all hiera sources, or use simple automatic parameter lookup
 class filebeat (
@@ -78,7 +78,8 @@
   $max_procs            = $filebeat::params::max_procs,
   $fields               = $filebeat::params::fields,
   $fields_under_root    = $filebeat::params::fields_under_root,
-  $processors           = $filebeat::params::processors,
+  $processors           = {},
+  $processors_merge     = false,
   #### End v5 onlly ####
   $prospectors          = {},
   $prospectors_merge    = false,
@@ -88,7 +89,7 @@
 
   $kernel_fail_message = "${::kernel} is not supported by filebeat."
 
-  validate_bool($manage_repo, $prospectors_merge)
+  validate_bool($manage_repo, $processors_merge, $prospectors_merge)
 
   if $major_version == undef and $::filebeat_version == undef {
     $real_version = '5'
@@ -123,11 +124,16 @@
     $prospectors_final = $prospectors
   }
 
+  if $processors_merge {
+    $processors_final = hiera_hash('filebeat::processors', $processors)
+  } else {
+    $processors_final = $processors
+  }
+
   if $config_file != $filebeat::params::config_file {
     warning('You\'ve specified a non-standard config_file location - filebeat may fail to start unless you\'re doing something to fix this')
   }
 
-  validate_array($processors)
   validate_hash($outputs, $logging, $prospectors_final)
   validate_string($idle_timeout, $registry_file, $config_dir, $package_ensure)
 
@@ -144,4 +150,7 @@
   if !empty($prospectors_final) {
     create_resources('filebeat::prospector', $prospectors_final)
   }
+  if !empty($processors_final) {
+    create_resources('filebeat::processors', $processors_final)
+  }
 }

manifests/params.pp

@@ -22,7 +22,6 @@
   $logging              = {}
   $run_options          = {}
   $use_generic_template = false
-  $processors           = []
 
   # These are irrelevant as long as the template is set based on the major_version parameter
   # if versioncmp('1.9.1', $::rubyversion) > 0 {

manifests/processor.pp

@@ -0,0 +1,75 @@
+define filebeat::processor(
+  $ensure         = present,
+  $priority       = 10,
+  $processor_name = $name,
+  $params         = undef,
+  $when           = undef,
+) {
+  include ::filebeat
+
+  validate_integer($priority)
+  validate_string($processor_name)
+
+  if versioncmp($filebeat::real_version, '5') < 0 {
+    fail('Processors only work on Filebeat 5.0 and higher')
+  }
+
+  if $priority < 10 {
+    $_priority = "0${priority}"
+  }
+  else {
+    $_priority = $priority
+  }
+
+  if $processor_name == 'drop_field' and $when == undef {
+    fail('drop_event processors require a condition, without one ALL events are dropped')
+  }
+  elsif $processor_name != 'add_cloud_metadata' and $params == undef {
+    fail("${processor_name} requires parameters to function as expected")
+  }
+
+  if $processor_name == 'add_cloud_metadata' {
+    $_configuration = delete_undef_values(merge({'timeout' => '3s'}, $params))
+  }
+  elsif $processor_name == 'drop_field' {
+    $_configuration = $when
+  }
+  else {
+    $_configuration = delete_undef_values(merge({'when' => $when}, $params))
+  }
+
+  $processor_config = delete_undef_values({
+    'processors' => [
+      {
+        $processor_name => $_configuration
+      },
+    ],
+  })
+
+  case $::kernel {
+    'Linux': {
+      file{"filebeat-processor-${name}":
+        ensure       => $ensure,
+        path         => "${filebeat::config_dir}/${_priority}-processor-${name}.yml",
+        owner        => 'root',
+        group        => 'root',
+        mode         => $::filebeat::config_file_mode,
+        content      => inline_template('<%= @processor_config.to_yaml() %>'),
+        validate_cmd => '/usr/share/filebeat/bin/filebeat -N -configtest -c %',
+        notify       => Class['filebeat::service'],
+      }
+    }
+    'Windows': {
+      file{"filebeat-processor-${name}":
+        ensure       => $ensure,
+        path         => "${filebeat::config_dir}/${_priority}-processor-${name}.yml",
+        content      => inline_template('<%= @processor_config.to_yaml() %>'),
+        validate_cmd => 'c:\Program Files\Filebeat\filebeat.exe -N -configtest -c %',
+        notify       => Class['filebeat::service'],
+      }
+    }
+    default: {
+      fail($filebeat::kernel_fail_message)
+    }
+  }
+}

spec/defines/processor_spec.rb

@@ -0,0 +1,73 @@
+require 'spec_helper'
+
+describe 'filebeat::processor', type: :define do
+  let :pre_condition do
+    'class{"filebeat":
+      outputs => {
+        "logstash" => {
+          "hosts" => [
+            "localhost:5044",
+          ],
+        },
+      },
+    }'
+  end
+  let :title do
+    'test-processor'
+  end
+
+  context 'with no parameters' do
+    it { is_expected.to raise_error(Puppet::Error) }
+  end
+
+  context 'on Linux' do
+    let :facts do
+      {
+        kernel: 'Linux',
+        osfamily: 'Linux',
+        rubyversion: '2.3.1',
+        filebeat_version: '5.1.2'
+      }
+    end
+
+    context 'add_cloud_metadata processor' do
+      let :params do
+        {
+          processor_name: 'add_cloud_metadata'
+        }
+      end
+
+      it do
+        is_expected.to contain_file('filebeat-processor-test-processor').with(
+          mode: '0644',
+          path: '/etc/filebeat/conf.d/10-processor-test-processor.yml',
+          content: '---
+processors:
+- add_cloud_metadata:
+    timeout: 3s
+'
+        )
+      end
+    end
+
+    context 'drop_event processor with no conditions' do
+      let :params do
+        {
+          processor_name: 'drop_event'
+        }
+      end
+
+      it { is_expected.to raise_error(Puppet::Error) }
+    end
+
+    context 'drop_field processor with no params' do
+      let :params do
+        {
+          processor_name: 'drop_field'
+        }
+      end
+
+      it { is_expected.to raise_error(Puppet::Error) }
+    end
+  end
+end

templates/filebeat5.yml.erb

@@ -1,16 +1,3 @@
-<%-
-  def yaml_indent(conds)
-    return_val = []
-    tmp_val = conds.to_yaml.split("\n")
-    tmp_val = tmp_val.collect{|val| val.rstrip || val}
-    tmp_val.delete('---')
-    tmp_val.each do |val|
-      return_val << "        " + val
-    end
-
-    return_val.join("\n")
-  end
--%>
 #========================= Filebeat global options ============================
 
 filebeat.spool_size: <%= @filebeat_config['filebeat']['spool_size'] %>
@@ -76,21 +63,6 @@ max_procs: <%= @filebeat_config['max_procs'] %>
 #       equals:
 #           http.code: 200
 #
-<%- unless @filebeat_config['processors'].empty? -%>
-processors:
-  <%- @filebeat_config['processors'].each do |_proc| -%>
-  - <%= _proc['name'] %>:
-    <%- unless _proc['when'].nil? or _proc['when'].empty? -%>
-      when:
-<%=  yaml_indent(_proc['when']) %>
-    <%- end -%>
-    <%- unless _proc['params'].nil? or _proc['params'].empty? -%>
-    <%- _proc['params'].each do |key,val|-%>
-      <%= key %>: <%= val %>
-    <%- end -%>
-    <%- end -%>
-  <%- end -%>
-<%- end -%>
 
 #================================ Outputs =====================================