class Fluent::AzureMonitorLogInput

Public Class Methods

new() click to toggle source
Calls superclass method
# File lib/fluent/plugin/in_azuremonitorlog.rb, line 28
def initialize
  super
end

Public Instance Methods

configure(conf) click to toggle source
Calls superclass method
# File lib/fluent/plugin/in_azuremonitorlog.rb, line 32
def configure(conf)
  super

  provider = MsRestAzure::ApplicationTokenProvider.new(@tenant_id, @client_id, @client_secret)
  credentials = MsRest::TokenCredentials.new(provider)
  @client = Azure::ARM::Monitor::MonitorManagementClient.new(credentials);
  @client.subscription_id = @subscription_id
end
set_query_options(filter, custom_headers) click to toggle source
# File lib/fluent/plugin/in_azuremonitorlog.rb, line 54
def set_query_options(filter, custom_headers)
  fail ArgumentError, 'path is nil' if @client.subscription_id.nil?

  request_headers = {}

  # Set Headers
  request_headers['x-ms-client-request-id'] = SecureRandom.uuid
  request_headers['accept-language'] = @client.accept_language unless @client.accept_language.nil?

  {
      middlewares: [[MsRest::RetryPolicyMiddleware, times: 3, retry: 0.02], [:cookie_jar]],
      path_params: {'subscriptionId' => @client.subscription_id},
      query_params: {'api-version' => @api_version, '$filter' => filter, '$select' => @select},
      headers: request_headers.merge(custom_headers || {}),
      base_url: @client.base_url
  }
end
shutdown() click to toggle source
Calls superclass method
# File lib/fluent/plugin/in_azuremonitorlog.rb, line 47
def shutdown
  super
  @finished = true
  @watcher.terminate
  @watcher.join
end
start() click to toggle source
Calls superclass method
# File lib/fluent/plugin/in_azuremonitorlog.rb, line 41
def start
  super
  @finished = false
  @watcher = Thread.new(&method(:watch))
end

Private Instance Methods

get_monitor_log_async(filter = nil, custom_headers = nil) click to toggle source
# File lib/fluent/plugin/in_azuremonitorlog.rb, line 105
def get_monitor_log_async(filter = nil, custom_headers = nil)
  options = set_query_options(filter, custom_headers)
  path_template = '/subscriptions/{subscriptionId}/providers/microsoft.insights/eventtypes/management/values'
  promise = @client.make_request_async(:get, path_template, options)

  promise = promise.then do |result|
    http_response = result.response
    status_code = http_response.status
    response_content = http_response.body
    unless status_code == 200
      error_model = JSON.load(response_content)
      log.error(error_model['error']['message'])
    end

    result.request_id = http_response['x-ms-request-id'] unless http_response['x-ms-request-id'].nil?
    # Deserialize Response
    if status_code == 200
      begin
        result.body = response_content.to_s.empty? ? nil : JSON.load(response_content)
      rescue Exception => e
        log.error("Error occurred in parsing the response")
      end
    end

    result
  end

  promise.execute
end
watch() click to toggle source
# File lib/fluent/plugin/in_azuremonitorlog.rb, line 74
def watch
  log.debug "azure monitorlog: watch thread starting"

  @next_fetch_time = Time.now

  until @finished
    start_time = @next_fetch_time - @interval
    end_time = @next_fetch_time
    log.debug "start time: #{start_time}, end time: #{end_time}"
    filter = "eventTimestamp ge '#{start_time}' and eventTimestamp le '#{end_time}'"

    if !@filter.empty?
      filter += " and #{@filter}"
    end

    monitor_logs_promise = get_monitor_log_async(filter)
    monitor_logs = monitor_logs_promise.value!

    if !monitor_logs.body['value'].nil? and  monitor_logs.body['value'].any?
      monitor_logs.body['value'].each {|val|
        router.emit(@tag, Time.now.to_i, val)
      }
    else
      log.debug "empty"
    end
    @next_fetch_time += @interval
    sleep @interval
  end

end