Class: Fluent::Plugin::AzureomsOutput

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

Constant Summary collapse

DEFAULT_BUFFER_TYPE = 
"memory"

Instance Method Summary collapse

Instance Method Details

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



144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 144

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

Raises:

  • (Fluent::ConfigError) 


44
45
46
47
48
49
50
51
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 44

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

  # Require chunking by tag keys
  raise Fluent::ConfigError, "'tag' in chunk_keys is required." if not @chunk_key_tag
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



56
57
58
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 56

def prefer_buffered_processing 
  true
end

#process(tag, es) ⇒ Object 



60
61
62
63
64
65
66
67
68
69
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 60

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, tag)        
  end 
end

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



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 112

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_name) ⇒ Object 



104
105
106
107
108
109
110
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 104

def send_data(customer_id, shared_key, content, log_name)        
    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



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/fluent/plugin/out_azureoms.rb', line 72

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)
  log.debug "writing data to log type", chunk.metadata.tag

  tag = chunk.metadata.tag
  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 #{elements.length} to logname #{tag}"    
      send_data(workspace, key, elements.to_json, tag)     
      elements.clear
    end 
  end

  if elements.length > 0
    log.debug "Flushing #{elements.length} to logname #{tag}"    
    send_data(workspace, key, elements.to_json, tag)     
  end        
end