Class: Fluent::Plugin::AzureomsOutput

Inherits:
Output
  • Object
show all
Defined in:
lib/fluent/plugin/out_azureoms.rb

Instance Method Summary collapse

Instance Method Details

#build_signature(shared_key, date, content_length, method, content_type, resource) ⇒ Object 



131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 131

def build_signature(shared_key, date, content_length, method, content_type, resource)
  rfc1123date = date.utc.strftime("%a, %d %b %Y %H:%M:%S GMT")
  string_to_hash = "#{method}\n#{content_length}\n#{content_type}\nx-ms-date:#{rfc1123date}\n#{resource}"        
  decoded_key = Base64.decode64(shared_key)
  secure_hash = OpenSSL::HMAC.digest('SHA256', decoded_key, string_to_hash)
        
  encoded_hash = Base64.encode64(secure_hash).strip()
  authorization = "SharedKey #{workspace}:#{encoded_hash}"

  return authorization
end

#configure(conf) ⇒ Object 



37
38
39
40
41
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 37

def configure(conf)
  # This also calls config_param (don't access configuration parameters before

  # calling super)

  super        
end

#prefer_buffered_processing ⇒ Object

This output plugin uses the raw HTTP data collector API per docs.microsoft.com/en-us/azure/log-analytics/log-analytics-data-collector-api



46
47
48
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 46

def prefer_buffered_processing 
  true
end

#process(tag, es) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 50

def process(tag, es)
  print "Writing single record set (synchronous)\n"
  es.each do |time, record|           
    # Convert record into a JSON body payload for OMS

    record[:timestamp] = Time.at(time).iso8601          

    # Publish event

    send_data(workspace, key, record.to_json, log_name)        
  end 
end

#publish_data(log_name, signature, time, json) ⇒ Object 



99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 99

def publish_data(log_name, signature, time, json)
  url = "https://#{workspace}.ods.opinsights.azure.com/api/logs?api-version=2016-04-01"
  uri = URI url

  rfc1123date = time.utc.strftime("%a, %d %b %Y %H:%M:%S GMT")

  response = Net::HTTP.start(uri.hostname, uri.port, 
    :use_ssl => uri.scheme == 'https') do |http|

    req = Net::HTTP::Post.new(uri.to_s)
    req.body = json.to_s
    
    # Signature and headers

    req['Content-Type'] = 'application/json'
    req['Log-Type'] = log_name
    req['Authorization'] = signature
    req['x-ms-date'] = rfc1123date

    log.debug "Publishing record of length #{req.body.length} to OMS workspace #{workspace}"    
    http.request(req)          
  end

  case response 
  when Net::HTTPSuccess
    log.debug "Successfully published record of length #{json.length} to OMS workspace #{workspace}"                 
  else
    # TODO - throw error

    log.warn "Could not publish record of length #{json.length} to OMS workspace #{workspace} because #{response}"
  end 
  response
end

#send_data(customer_id, shared_key, content, log_type) ⇒ Object 



91
92
93
94
95
96
97
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 91

def send_data(customer_id, shared_key, content, log_type)        
    current_time = Time.now.utc
    signature = build_signature(
      shared_key, current_time, content.length, 
      "POST", "application/json", "/api/logs")
    publish_data(log_name, signature, current_time, content)
end

#write(chunk) ⇒ Object

Synchronous buffered output



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 62

def write(chunk) 
  log.debug "Writing buffered record set (synchronous)"
  log.debug "writing data to file", chunk_id: dump_unique_id_hex(chunk.unique_id)
  
  elements = Array.new
  
  chunk.each do |time, record|
    log.debug "Writing record #{record.inspect}"    

    # Fold the timestamp into the record

    record[:timestamp] = Time.at(time).iso8601

    # Append the record to the content in the appropriate format

    elements.push(record)

    # TODO - check size of content buffer and flush when it approaches 

    # watermark

    if false 
      log.debug "Elements buffer approaching max send size; flushing TODO"    
      send_data(workspace, key, elements.to_json, log_name)     
      elements.clear
    end 
  end

  if elements.length > 0
    send_data(workspace, key, elements.to_json, log_name)     
  end        
end