Class: Fluent::SpectrumInput

Inherits:

Input

• Object
• Input
• Fluent::SpectrumInput

Defined in:

lib/fluent/plugin/in_spectrum.rb

Defined Under Namespace

Classes: MemoryStateStore, StateStore, TimerWatcher

Constant Summary

INTERVAL_MIN =

Configurations

10

Instance Method Summary

• #configure(conf) ⇒ Object

  def initialize.

• #initialize ⇒ SpectrumInput constructor

  A new instance of SpectrumInput.

• #on_timer ⇒ Object

  def run.

• #parseAttributes(alarmAttribute) ⇒ Object
• #resource ⇒ Object

  URL Resource.

• #run ⇒ Object

  def shutdown.

• #shutdown ⇒ Object

  def start.

• #start ⇒ Object

  def configure.

• #to_utf8(str) ⇒ Object

  function to UTF8 encode.

Constructor Details

#initialize ⇒ SpectrumInput

Returns a new instance of SpectrumInput.

# File 'lib/fluent/plugin/in_spectrum.rb', line 85

def initialize
  require 'rest-client'
  require 'json'
  super
end

Instance Method Details

#configure(conf) ⇒ Object

def initialize

# File 'lib/fluent/plugin/in_spectrum.rb', line 91

def configure(conf)
  super 
  @conf = conf
  # Verify configs
  # Stop if required fields are not set
  unless @endpoint && @username && @password
    raise ConfigError, "Spectrum :: ConfigError 'endpoint' and 'username' and 'password' must be all specified."
  end
  # Enforce min interval
  if @interval.to_i < INTERVAL_MIN
    raise ConfigError, "Spectrum :: ConfigError 'interval' must be #{INTERVAL_MIN} or over."
  end
  # Warn about optional state file
  unless @state_file
    $log.warn "Spectrum :: 'state_file PATH' parameter is not set to a valid source."
    $log.warn "Spectrum :: this parameter is highly recommended to save the last known good timestamp to resume event consuming"
  end
  # map of Spectrum attribute codes to names
  @spectrum_access_code={
    "0x11f9c" => "ALARM_ID",
    "0x11f4e" => "CREATION_DATE",
    "0x11f56" => "SEVERITY",
    "0x12b4c" => "ALARM_TITLE",
    "0x1006e" => "HOSTNAME",
    "0x12d7f" => "IP_ADDRESS",
    "0x1296e" => "ORIGINATING_EVENT_ATTR",
    "0x10000" => "MODEL_STRING",  
    "0x11f4d" => "ACKNOWLEDGED",
    "0x11f4f" => "ALARM_STATUS",
    "0x11fc5" => "OCCURRENCES",
    "0x11f57" => "TROUBLE_SHOOTER",
    "0x11f9b" => "USER_CLEARABLE",
    "0x12022" => "TROUBLE_TICKET_ID",
    "0x12942" => "PERSISTENT",
    "0x12adb" => "GC_NAME",
    "0x57f0105" => "Custom_Project",
    #{}"0x11f51" => "CLEARED_BY_USER_NAME",
    #{}"0x11f52" => "EVENT_ID_LIST",
    #{}"0x11f53" => "MODEL_HANDLE",
    #{}"0x11f54" => "PRIMARY_ALARM",
    #{}"0x11fc4" => "ALARM_SOURCE",
    #{}"0x11fc6" => "TROUBLE_SHOOTER_MH",
    #{}"0x12a6c" => "TROUBLE_SHOOTER_EMAIL",
    #{}"0x1290d" => "IMPACT_SEVERITY",
    #{}"0x1290e" => "IMPACT_SCOPE",
    #{}"0x1298a" => "IMPACT_TYPE_LIST",
    #{}"0x12948" => "DIAGNOSIS_LOG",
    #{}"0x129aa" => "MODEL_ID",
    #{}"0x129ab" => "MODEL_TYPE_ID",
    #{}"0x129af" => "CLEAR_DATE",
    #{}"0x12a04" => "SYMPTOM_LIST_ATTR",
    #{}"0x12a6f" => "EVENT_SYMPTOM_LIST_ATTR",
    #{}"0x12a05" => "CAUSE_LIST_ATTR",
    #{}"0x12a06" => "SYMPTOM_COUNT_ATTR",
    #{}"0x12a70" => "EVENT_SYMPTOM_COUNT_ATTR",
    #{}"0x12a07" => "CAUSE_COUNT_ATTR",
    #{}"0x12a63" => "WEB_CONTEXT_URL",
    #{}"0x12a6b" => "COMBINED_IMPACT_TYPE_LIST",
    #{}"0x11f50" => "CAUSE_CODE",
    #{}"0x10009" => "SECURITY_STRING"
  }
  # Create XML chunk for attributes we care about
  @attr_of_interest=""
  @spectrum_access_code.each do |key, array|
    @attr_of_interest += " <rs:requested-attribute id=\"#{key}\"/>"
  end
  # URL Resource
  def resource
    @url = 'http://' + @endpoint.to_s + '/spectrum/restful/alarms'
    RestClient::Resource.new(@url, :user => @username, :password => @password, :open_timeout => 5, :timeout => (@interval * 3))
  end
end

#on_timer ⇒ Object

def run

# File 'lib/fluent/plugin/in_spectrum.rb', line 186

def on_timer
  if not @stop_flag
    pollingStart = Engine.now.to_i
    if @state_store.last_records.has_key?("spectrum") 
      alertStartTime = @state_store.last_records['spectrum']
    else
      alertStartTime = (pollingStart.to_i - @interval.to_i)
    end
    pollingEnd = ''
    pollingDuration = ''
    # Format XML for spectrum post
    @xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>
    <rs:alarm-request throttlesize=\"#{select_limit}\"
    xmlns:rs=\"http://www.ca.com/spectrum/restful/schema/request\"
    xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"
    xsi:schemaLocation=\"http://www.ca.com/spectrum/restful/schema/request ../../../xsd/Request.xsd \">
    <rs:attribute-filter>
      <search-criteria xmlns=\"http://www.ca.com/spectrum/restful/schema/filter\">
      <filtered-models>
        <greater-than>
          <attribute id=\"0x11f4e\">
            <value> #{alertStartTime} </value>
          </attribute>
        </greater-than>
      </filtered-models>
      </search-criteria>
    </rs:attribute-filter>
    #{@attr_of_interest}
    </rs:alarm-request>"

    # Post to Spectrum and parse results
    begin
      res=resource.post @xml,:content_type => 'application/xml',:accept => 'application/json'
      body = JSON.parse(res.body)
      pollingEnd = Time.parse(res.headers[:date]).to_i
      @state_store.last_records['spectrum'] = pollingEnd
      pollingDuration = Engine.now.to_i - pollingStart
    end  

    # Processing for multiple alerts returned
    if body['ns1.alarm-response-list']['@total-alarms'].to_i > 1
      $log.info "Spectrum :: returned #{body['ns1.alarm-response-list']['@total-alarms'].to_i} alarms for period < #{alertStartTime.to_i} took #{pollingDuration.to_i} seconds, ended at #{pollingEnd}"
      # iterate through each alarm
      body['ns1.alarm-response-list']['ns1.alarm-responses']['ns1.alarm'].each do |alarm|
        # Create initial structure
        record_hash = Hash.new # temp hash to hold attributes of alarm
        raw_array = Array.new # temp hash to hold attributes of alarm for raw
        record_hash['event_type'] = @tag.to_s
        record_hash['intermediary_source'] = @endpoint.to_s
        record_hash['receive_time_input'] = pollingEnd.to_s
        # iterate though alarm attributes
        alarm['ns1.attribute'].each do |attribute|
          key,value = parseAttributes(attribute)
          record_hash[key] = value
          if @include_raw.to_s == "true"
            raw_array << { "#{key}" => "#{value}" }
          end
        end
        # append raw object
        if @include_raw.to_s == "true"  
          record_hash[:raw] = raw_array
        end
        Engine.emit(@tag, record_hash['CREATION_DATE'].to_i,record_hash)
      end
    # Processing for single alarm returned  
    elsif body['ns1.alarm-response-list']['@total-alarms'].to_i == 1
      $log.info "Spectrum :: returned #{body['ns1.alarm-response-list']['@total-alarms'].to_i} alarms for period < #{alertStartTime.to_i} took #{pollingDuration.to_i} seconds, ended at #{pollingEnd}"
      # Create initial structure
      record_hash = Hash.new # temp hash to hold attributes of alarm
      raw_array = Array.new # temp hash to hold attributes of alarm for raw
      record_hash['event_type'] = @tag.to_s
      record_hash['intermediary_source'] = @endpoint.to_s
      record_hash['receive_time_input'] = pollingEnd.to_s
      # iterate though alarm attributes and add to temp hash  
      body['ns1.alarm-response-list']['ns1.alarm-responses']['ns1.alarm']['ns1.attribute'].each do |attribute|
        key,value = parseAttributes(attribute)
        record_hash[key] = value
        if @include_raw.to_s == "true"
          raw_array << { "#{key}" => "#{value}" }
        end
      end
      # append raw object
      if @include_raw.to_s == "true"  
        record_hash[:raw] = raw_array
      end
      Engine.emit(@tag, record_hash['CREATION_DATE'].to_i,record_hash)
    # No alarms returned
    else
      $log.info "Spectrum :: returned #{body['ns1.alarm-response-list']['@total-alarms'].to_i} alarms for period < #{alertStartTime.to_i} took #{pollingDuration.to_i} seconds, ended at #{pollingEnd}"
    end
    @state_store.update!  
  end
end

#parseAttributes(alarmAttribute) ⇒ Object

# File 'lib/fluent/plugin/in_spectrum.rb', line 79

def parseAttributes(alarmAttribute)
  key = @spectrum_access_code[alarmAttribute['@id'].to_s].to_s
  value = ((to_utf8(alarmAttribute['$'].to_s)).strip).gsub(/\r?\n/, " ")
  return key,value
end

#resource ⇒ Object

URL Resource

# File 'lib/fluent/plugin/in_spectrum.rb', line 158

def resource
  @url = 'http://' + @endpoint.to_s + '/spectrum/restful/alarms'
  RestClient::Resource.new(@url, :user => @username, :password => @password, :open_timeout => 5, :timeout => (@interval * 3))
end

#run ⇒ Object

def shutdown

# File 'lib/fluent/plugin/in_spectrum.rb', line 179

def run
  @loop.run
rescue
  $log.error "unexpected error", :error=>$!.to_s
  $log.error_backtrace
end

#shutdown ⇒ Object

def start

# File 'lib/fluent/plugin/in_spectrum.rb', line 172

def shutdown
  #@loop.watchers.each {|w| w.detach}
  @stop_flag = true
  @loop.stop
  @thread.join
end

#start ⇒ Object

def configure

# File 'lib/fluent/plugin/in_spectrum.rb', line 164

def start
  @stop_flag = false
  @state_store = @state_file.nil? ? MemoryStateStore.new : StateStore.new(@state_file)
  @loop = Coolio::Loop.new
  @loop.attach(TimerWatcher.new(@interval, true, &method(:on_timer)))
  @thread = Thread.new(&method(:run))
end

#to_utf8(str) ⇒ Object

function to UTF8 encode

# File 'lib/fluent/plugin/in_spectrum.rb', line 73

def to_utf8(str)
  str = str.force_encoding('UTF-8')
  return str if str.valid_encoding?
  str.encode("UTF-8", 'binary', invalid: :replace, undef: :replace, replace: '')
end