adding native types for input, output, streams etc

lib/puppet/provider/graylog2.rb

@@ -0,0 +1,200 @@
+require 'net/http'
+require 'json'
+require 'pp'
+require 'yaml'
+
+class Puppet::Provider::Graylog2 < Puppet::Provider
+
+  # Load in the YAML configuration file, check for errors, and return as hash 'cfg'
+  def self.load_config
+    @try_timeout = 20
+    #file that contains some special configuration for the native types
+    configfile = '/var/lib/puppet/module_data/graylog2/types_conf.yaml'
+    cfg = File.open(configfile)  { |yf| YAML::load( yf ) } if File.exists?(configfile)
+    # => Ensure loaded data is a hash. ie: YAML load was OK
+    if cfg.class != Hash
+       fail('ERROR: configuration - invalid format or parsing error.')
+    end
+    fail('config not loaded correctly! graylog_api_port is missing...') if cfg['graylog_api_port'].nil?
+    @http = Net::HTTP.new('localhost', cfg['graylog_api_port'])
+    fail('config not loaded correctly! graylog_admin_user is missing...') unless @user = cfg['graylog_admin_user']
+    fail('config not loaded correctly! graylog_admin_pass is missing...') unless @pass = cfg['graylog_admin_pass']
+    fail('config not loaded correctly! input_bypass_prefixes is missing...') unless @input_bypass_prefixes = cfg['input_bypass_prefixes']
+    fail('config not loaded correctly! output_bypass_prefixes is missing...') unless @output_bypass_prefixes = cfg['output_bypass_prefixes']
+    fail('config not loaded correctly! stream_bypass_prefixes is missing...') unless @stream_bypass_prefixes = cfg['stream_bypass_prefixes']
+    fail('config not loaded correctly! user_bypass_prefixes is missing...') unless @user_bypass_prefixes = cfg['user_bypass_prefixes']
+    @http.read_timeout = @try_timeout
+    @http.open_timeout = @try_timeout
+  end
+
+  # Generic HTTP call
+  def self.generic_http_call(request)
+    if @user.nil?
+      self.load_config
+    end
+    if @user.nil?
+      return nil
+    end
+    request.basic_auth(@user, @pass)
+    for i in 1..15
+      http_error = false
+      begin
+        response = @http.request(request)
+      rescue
+        http_error = true
+        Puppet.debug('generic_http_call: HTTP connection error!')
+      end
+      if !http_error
+        if /^20[0-9]$/.match(response.code)
+          if (!response.body.nil?) && (!response.body.empty?)
+            return JSON.parse(response.body)
+          else
+            return {}
+          end
+        else
+          if /^40[0-9]$/.match(response.code)
+            Puppet.debug("generic_http_call: response code= #{response.code} response message= #{response.message}")
+            return {}
+          end
+          Puppet.debug("generic_http_call: bad response code= #{response.code} response message= #{response.message}!")
+        end
+      end
+      Puppet.debug("generic_http_call: #{i} attempt failed... retrying in #{@try_timeout}s...")
+      sleep @try_timeout
+    end
+    raise 'Call to graylog failed!' # I wanted to stop Puppet if this happens but apparently it's not possible
+  end
+
+  # Generic GET/POST/PUT/DELETE calls
+  def self.generic_get(path)
+    Puppet.debug('generic_get: path= ' + path)
+    request = Net::HTTP::Get.new(path)
+    return self.generic_http_call(request)
+  end
+
+  def generic_post(path, body_map)
+    Puppet.debug('generic_post: path= ' + path)
+    request = Net::HTTP::Post.new(path)
+    request['Content-Type'] = 'application/json'
+    if body_map
+      request.body = body_map.to_json
+      Puppet.debug("generic_post: theres a bodymap= #{request.body}")
+    end
+    return self.class.generic_http_call(request) # self.class. is needed to go from the instance scope to the class scope
+  end
+
+  def generic_put(path, body_map)
+    Puppet.debug('generic_put: path= ' + path)
+    request = Net::HTTP::Put.new(path)
+    request.body = body_map.to_json
+    Puppet.debug("generic_put: bodymap= #{request.body}")
+    Puppet.debug(request.body)
+    request['Content-Type'] = 'application/json'
+    return self.class.generic_http_call(request)
+  end
+
+  def generic_del(path)
+    Puppet.debug('generic_del: path= ' + path)
+    request = Net::HTTP::Delete.new(path)
+    return self.class.generic_http_call(request)
+  end
+
+  # CONVERSION FUNCTIONS
+  def self.convert_property_map_to_str(hash_map)
+    new_hash_map = {}
+    hash_map.each do |k,v|
+      new_hash_map[k] = v.to_s
+    end
+    return new_hash_map
+  end
+
+  def self.convert_bool_to_symbol(value)
+    if value
+      return :true
+    end
+    return :false
+  end
+
+  def self.ignore_resource?(resource_name, prefix_list)
+    prefix_list.each do |prefix|
+      if Regexp.new('^' + prefix).match(resource_name)
+        return true
+      end
+    end
+    return false
+  end
+
+  # resources with names that are already in the list, we start labeling them with the tail ".DUPLICATED.X"
+  def self.mark_duplicated_name(name_base, name_list)
+    unless name_list[name_base]
+      return name_base
+    end
+    duplicated_number = 1
+    while true
+      name = "#{name_base}.DUPLICATED.#{duplicated_number}"
+      unless name_list[name]
+        return name
+      end
+      duplicated_number = duplicated_number + 1
+    end
+  end
+
+  def convert_symbol_to_bool(value)
+    return value == :true
+  end
+
+  # avoids badly formed json strings like {"port":"1221"} (there shouldnt be any quote around 1221)
+  # TODO do this inside the type-munge and not inside the provider
+  def unquote(hash_map)
+    new_hash_map = {}
+    hash_map.each do |k,v|
+      case k
+      # properties that shouldnt be around quotes
+      # TODO place up in EMOC/EMEC puppet module
+      when 'port', 'recv_buffer_size', 'connect_timeout', 'reconnect_delay', 'type', 'max_chunk_size', 'max_message_size', 'parallel_queues', 'prefetch', 'broker_port', 'amqp_connection_timeout', 'amqp_port'
+        new_hash_map[k] = v.to_i
+      when 'use_null_delimiter', 'allow_override_date', 'enable_cors', 'store_full_message'
+        if v == 'true'
+          new_hash_map[k] = true
+        else
+          new_hash_map[k] = false
+        end
+      else
+        new_hash_map[k] = v
+      end
+    end
+    return new_hash_map
+  end
+
+  # The prefetch method takes the provider instances that self.instances found and connects them
+  # to the existing resources in the puppet catalogue. Generic, works for all providers with self.instances
+  def self.prefetch(resources)
+    instances.each do |instance|
+      if resource = resources[instance.name]
+        resource.provider = instance
+      end
+    end
+  end
+
+  def create
+    Puppet.debug("Creating resource #{resource[:name]}")
+    @property_hash[:ensure] = :present
+    @property_hash[:new] = true
+    @property_hash[:name] = @resource[:name]
+    self.class.resource_type.validproperties.each do |property|
+      if val = resource.should(property)
+        @property_hash[property] = val
+      end
+    end
+  end
+
+  def destroy
+    Puppet.debug("Destroying resource #{resource[:name]}")
+    @property_hash[:ensure] = :absent
+  end
+
+  def exists?
+    @property_hash.fetch(:ensure, :absent) != :absent
+  end
+
+end

lib/puppet/provider/graylog2_extractor/graylog2.rb

@@ -0,0 +1,147 @@
+# needed to load the common file lib/puppet/provider/graylog2.rb
+require File.expand_path(File.join(File.dirname(__FILE__), '..', 'graylog2'))
+
+Puppet::Type.type(:graylog2_extractor).provide(:graylog2, :parent => Puppet::Provider::Graylog2) do
+
+  desc 'Support for graylog extractors.'
+
+  # creates a bunch of getters/setters for our properties/parameters
+  # it is only for prefetch/flush providers
+  # gets properties from @property_hash which contains the 'is' state of the resource
+  mk_resource_methods
+
+  def get_input_id(input_name)
+    Puppet.debug("get_input_id: input name is= #{input_name}")
+    inputs = self.class.generic_get('/system/inputs')['inputs']
+    inputs.each do |input|
+      if input['message_input']['title'] == input_name
+        return input['id']
+      end
+    end
+    return nil
+  end
+
+  def self.transform_converters(converters)
+    converters_new = {}
+    converters.each do |converter|
+      converters_new[converter['type']] = converter['config']
+    end
+    return converters_new
+  end
+
+  def self.get_extractors
+    extractors = []
+    gl_inputs = self.generic_get('/system/inputs')['inputs']
+    gl_inputs.each do |gl_input|
+      # There's a prefix for resources that are not taken into consideration
+      ignore_input = false
+      @input_bypass_prefixes.each do |prefix|
+        if Regexp.new('^' + prefix).match(gl_input['message_input']['title'])
+          ignore_input = true
+          break
+        end
+      end
+      unless ignore_input
+        extractor_list = self.generic_get("/system/inputs/#{gl_input['id']}/extractors")['extractors']
+        extractor_list.each do |extractor|
+          extractor = {
+            'input'            => gl_input['message_input']['title'],
+            'extractor_name'   => extractor['title'],
+            'title'            => "#{gl_input['message_input']['title']}:#{extractor['title']}",
+            'id'               => extractor['id'],
+            'source_field'     => extractor['source_field'],
+            'target_field'     => extractor['target_field'],
+            'type'             => extractor['type'],
+            'extractor_config' => extractor['extractor_config'],
+            'condition_type'   => extractor['condition_type'],
+            'condition_value'  => extractor['condition_value'],
+            'cursor_strategy'  => extractor['cursor_strategy'],
+            'converters'       => self.transform_converters(extractor['converters']),
+          }
+          extractors << extractor
+        end
+      end
+    end
+    return extractors
+  end
+
+  def create_extractor(name, element_properties)
+    input_id = get_input_id(name.split(':')[0])
+    extractor_id = self.generic_post("/system/inputs/#{input_id}/extractors", element_properties)['extractor_id']
+    return extractor_id
+  end
+
+  def modify_extractor(name, extractor_id, element_properties)
+    input_id = get_input_id(name.split(':')[0])
+    Puppet.debug("modify_extractor: body of the put call=\n#{element_properties.pretty_inspect}")
+    self.generic_put("/system/inputs/#{input_id}/extractors/#{extractor_id}", element_properties)
+  end
+
+  def delete_extractor(name, extractor_id)
+    input_id = get_input_id(name.split(':')[0])
+    self.generic_del("/system/inputs/#{input_id}/extractors/#{extractor_id}")
+  end
+
+  # This method collects on the node all existing instances of the things that this provider can manage.
+  # Regardless of wheter there's a corresponding resource declaration in a manifest.
+  # This should represent the 'is' state on a node.
+  # The method should return an array of provider-instances.
+  # Create a new instance with: new(:name => name, :property1 => current_state)
+  def self.instances
+    Puppet.debug('self.instances: retrieving extractors')
+    resources = []
+    gl_extractors = self.get_extractors()
+    gl_extractors.each do |gl_extractor|
+      element_properties = {
+        :name             => gl_extractor['title'],
+        :ensure           => :present,
+        :input            => gl_extractor['input'],
+        :extractor_name   => gl_extractor['extractor_name'],
+        :graylog_id       => gl_extractor['id'],
+        :source_field     => gl_extractor['source_field'],
+        :target_field     => gl_extractor['target_field'],
+        :extractor_type   => gl_extractor['type'],
+        :extractor_config => gl_extractor['extractor_config'],
+        :condition_type   => gl_extractor['condition_type'],
+        :condition_value  => gl_extractor['condition_value'],
+        :cut_or_copy      => gl_extractor['cursor_strategy'],
+        :converters       => gl_extractor['converters'],
+      }
+      Puppet.debug("self.instances: resource found=\n#{element_properties.pretty_inspect}")
+      resources << new(element_properties)
+    end
+    Puppet.debug('self.instances: number of resources= ' + resources.length.to_s)
+    return resources
+  end
+
+  # Function called at the end to apply all changes
+  def flush
+    Puppet.debug("flush: graylog2_extractor #{resource[:name]}")
+    case @property_hash[:ensure]
+    # if ensure is absent, we delete the input
+    when :absent
+      Puppet.debug("flush: graylog2_extractor #{resource[:name]} to be removed")
+      delete_extractor(@property_hash[:name], @property_hash[:graylog_id])
+    when :present
+      Puppet.debug("flush: update params:\n#{@property_hash.pretty_inspect}")
+      element_properties = {
+        :title            => @property_hash[:extractor_name],
+        :source_field     => @property_hash[:source_field],
+        :target_field     => @property_hash[:target_field],
+        :extractor_type   => @property_hash[:extractor_type],
+        :extractor_config => @property_hash[:extractor_config],
+        :condition_type   => @property_hash[:condition_type],
+        :condition_value  => @property_hash[:condition_value],
+        :cut_or_copy      => @property_hash[:cut_or_copy],
+        :converters       => @property_hash[:converters],
+      }
+      # if the resource is new, we have to create the input
+      if @property_hash[:new]
+        @property_hash[:graylog_id] = create_extractor(@property_hash[:name], element_properties)
+      # if the resource is not new, we have to modify the input... PUT changes
+      else
+        modify_extractor(@property_hash[:name], @property_hash[:graylog_id], element_properties)
+      end
+    end
+  end
+end