Class: Fluent::Plugin::KustoOutput

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

Instance Method Summary collapse

Instance Method Details

#check_data_on_server(chunk_id, row_count, resolved_table) ⇒ Object 



292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
 
# File 'lib/fluent/plugin/out_kusto.rb', line 292

def check_data_on_server(chunk_id, row_count, resolved_table)
  # Query Kusto to verify chunk ingestion
  begin
    # Sanitize inputs to prevent injection attacks
    safe_table_name = resolved_table.to_s.gsub(/[^a-zA-Z0-9_]/, '')
    safe_chunk_id = chunk_id.to_s.gsub(/[^a-zA-Z0-9_-]/, '')
    query = "#{safe_table_name} | extend record_dynamic = parse_json(record) | where record_dynamic.chunk_id == '#{safe_chunk_id}' | count"
    result = run_kusto_api_query(query, @outconfiguration.kusto_endpoint, @ingester.token_provider,
                                 use_ingest_endpoint: false, database_name: @database_name)
    if result.is_a?(Array) && result[0].is_a?(Array)
      count_val = result[0][0].to_i
      return count_val == row_count
    elsif @logger.respond_to?(:error)
      if @logger.respond_to?(:error)
        @logger.error("Kusto query failed or returned unexpected result: #{result.inspect}")
      end
    end
  rescue StandardError => e
    @logger.error("Failed to get chunk_id count: #{e}") if @logger.respond_to?(:error)
  end
  false
end

#compress_data(data) ⇒ Object 



154
155
156
157
158
159
160
161
 
# File 'lib/fluent/plugin/out_kusto.rb', line 154

def compress_data(data)
  # Compress data using gzip
  sio = StringIO.new
  gz = Zlib::GzipWriter.new(sio)
  gz.write(data)
  gz.close
  sio.string
end

#configure(conf) ⇒ Object 



59
60
61
62
63
64
65
66
67
 
# File 'lib/fluent/plugin/out_kusto.rb', line 59

def configure(conf)
  # Configure plugin and validate parameters
  compat_parameters_convert(conf, :buffer)
  super
  validate_buffer_config(conf)
  validate_delayed_config
  validate_required_params
  @table_name_template = table_name
end

#dump_unique_id_hex(unique_id) ⇒ Object 



146
147
148
149
150
151
152
 
# File 'lib/fluent/plugin/out_kusto.rb', line 146

def dump_unique_id_hex(unique_id)
  # Convert unique_id to hex string
  return 'noid' if unique_id.nil?
  return unique_id.unpack1('H*') if unique_id.respond_to?(:unpack1)

  unique_id.to_s
end

#extract_tag(record, tag) ⇒ Object 



90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/fluent/plugin/out_kusto.rb', line 90

def extract_tag(record, tag)
  # Extract tag from record or fallback to defaults
  return tag if !record.is_a?(Hash) || record.nil?
  return record['tag'] if record['tag']
  return tag if tag
  return record['host'] if record['host']
  return record['user'] if record['user']
  return ::Regexp.last_match(1) if record['message'] && record['message'] =~ /(\d{1,3}(?:\.\d{1,3}){3})/

  'default_tag'
end

#extract_tag_from_metadata(metadata) ⇒ Object 



198
199
200
201
202
203
204
 
# File 'lib/fluent/plugin/out_kusto.rb', line 198

def extract_tag_from_metadata(metadata)
  # Extract tag from chunk metadata
  return 'default_tag' if metadata.nil?
  return metadata.tag || 'default_tag' if metadata.respond_to?(:tag)

  'default_tag'
end

#extract_timestamp(record, time) ⇒ Object 



102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/fluent/plugin/out_kusto.rb', line 102

def extract_timestamp(record, time)
  # Extract datetime from record or fallback to time
  timestamp = find_time_or_date_key(record)
  return timestamp if timestamp && !timestamp.to_s.empty?

  timestamp = (time ? Time.at(time).utc.iso8601 : '')
  return timestamp unless timestamp.to_s.empty?

  timestamp = find_timestamp_by_regex(record)
  timestamp ||= ''
  timestamp
end

#find_time_or_date_key(record) ⇒ Object 



115
116
117
118
119
120
121
122
123
 
# File 'lib/fluent/plugin/out_kusto.rb', line 115

def find_time_or_date_key(record)
  # Find time/date key in record
  return nil unless record.is_a?(Hash)

  record.each do |k, v|
    return v if k.to_s.downcase.include?('time') || k.to_s.downcase.include?('date')
  end
  nil
end

#find_timestamp_by_regex(record) ⇒ Object 



125
126
127
128
129
130
131
132
 
# File 'lib/fluent/plugin/out_kusto.rb', line 125

def find_timestamp_by_regex(record)
  # Find datetime by regex in record values
  record.each_value do |v|
    next unless v.is_a?(String)
    return ::Regexp.last_match(1) if v =~ %r{(\[\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} [+-]\d{4}\].*)}
  end
  nil
end

#format(tag, time, record) ⇒ Object 



82
83
84
85
86
87
88
 
# File 'lib/fluent/plugin/out_kusto.rb', line 82

def format(tag, time, record)
  # Format a record for ingestion
  tag_val = extract_tag(record, tag)
  timestamp = extract_timestamp(record, time)
  safe_record = sanitize_record_for_json(record)
  "#{format_record_json(tag_val, timestamp, safe_record)}\n"
end

#format_record_json(tag_val, timestamp, safe_record) ⇒ Object 



134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/fluent/plugin/out_kusto.rb', line 134

def format_record_json(tag_val, timestamp, safe_record)
  # Format record as JSON for ingestion
  record_value = if safe_record.is_a?(Hash)
                   safe_record.reject do |k, _|
                     %w[tag time].include?(k)
                   end
                 else
                   safe_record || {}
                 end
  { 'tag' => tag_val, 'timestamp' => timestamp, 'record' => record_value }.to_json
end

#handle_kusto_error(e, unique_id) ⇒ Object 



206
207
208
209
 
# File 'lib/fluent/plugin/out_kusto.rb', line 206

def handle_kusto_error(e, unique_id)
  # Handle and log Kusto errors
  KustoErrorHandler.handle_kusto_error(@logger, e, dump_unique_id_hex(unique_id))
end

#multi_workers_ready? ⇒ Boolean

Returns:

  • (Boolean) 


54
55
56
57
 
# File 'lib/fluent/plugin/out_kusto.rb', line 54

def multi_workers_ready?
  # Enable multi-worker support
  true
end

#process(tag, es) ⇒ Object 



163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/fluent/plugin/out_kusto.rb', line 163

def process(tag, es)
  resolved_table = resolve_table_name(tag)
  es.each do |time, record|
    formatted = format(tag, time, record).encode('UTF-8', invalid: :replace, undef: :replace, replace: '_')
    safe_tag = tag.to_s.encode('UTF-8', invalid: :replace, undef: :replace, replace: '_').gsub(/[^0-9A-Za-z.-]/,
                                                                                               '_')
    blob_name = "fluentd_event_#{safe_tag}.json"
    @ingester.upload_data_to_blob_and_queue(formatted, blob_name, @database_name, resolved_table,
                                            compression_enabled, @ingestion_mapping_reference)
  rescue StandardError => e
    @logger&.error("Failed to ingest event to Kusto: #{e}\nEvent skipped: #{record.inspect}\n#{e.backtrace.join("\n")}")
    next
  end
end

#resolve_table_name(tag) ⇒ Object 



347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
 
# File 'lib/fluent/plugin/out_kusto.rb', line 347

def resolve_table_name(tag)
  # Resolve table name from template with placeholders
  if @table_name_template.nil? || @table_name_template.empty?
    @logger&.error('Table name template is nil or empty')
    raise Fluent::ConfigError, 'table_name must be set and non-empty'
  end
  
  return @table_name_template unless @table_name_template.include?('${')

  tag_str = tag.to_s
  
  # Validate tag is not empty when using placeholders
  if tag_str.empty?
    @logger&.warn("Tag is empty when resolving dynamic table name, using template as fallback: #{@table_name_template}")
    return @table_name_template.gsub(/\$\{[^}]+\}/, 'unknown').gsub(/[^0-9A-Za-z_]/, '_')
  end
  
  tag_parts = tag_str.split('.')
  result = @table_name_template.dup

  # Replace ${tag} with full tag (dots converted to underscores)
  result = result.gsub('${tag}', tag_str.gsub('.', '_'))

  # Replace ${tag_parts[N]} with Nth part
  result = result.gsub(/\$\{tag_parts\[(\d+)\]\}/) do
    index = ::Regexp.last_match(1).to_i
    tag_parts[index] || 'unknown'
  end

  # Replace ${tag_prefix[N]} with first N parts
  result = result.gsub(/\$\{tag_prefix\[(\d+)\]\}/) do
    count = ::Regexp.last_match(1).to_i
    parts = tag_parts.take(count)
    parts.empty? ? 'unknown' : parts.join('_')
  end

  # Replace ${tag_suffix[N]} with last N parts
  result = result.gsub(/\$\{tag_suffix\[(\d+)\]\}/) do
    count = ::Regexp.last_match(1).to_i
    parts = tag_parts.last(count)
    parts.empty? ? 'unknown' : parts.join('_')
  end

  # Sanitize: replace special characters with underscores and collapse consecutive underscores
  sanitized = result.gsub(/[^0-9A-Za-z_]/, '_').gsub(/_+/, '_')
  
  # Final validation: ensure we don't have an empty table name
  if sanitized.empty? || sanitized == '_'
    @logger&.error("Resolved table name is empty or invalid for tag '#{tag}' with template '#{@table_name_template}'")
    raise Fluent::ConfigError, "table_name resolved to empty or invalid value for tag '#{tag}'"
  end
  
  sanitized
end

#shutdown ⇒ Object 



315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
 
# File 'lib/fluent/plugin/out_kusto.rb', line 315

def shutdown
  # Handle plugin shutdown and cleanup threads
  @shutdown_called = true
  
  # Give deferred threads a chance to finish gracefully
  if @deferred_threads&.any?
    @logger&.info("Shutting down with #{@deferred_threads.size} active deferred commit threads")
    
    # Wait up to 10 seconds for threads to complete naturally
    deadline = Time.now + 10
    
    while Time.now < deadline && @deferred_threads.any?(&:alive?)
      alive_count = @deferred_threads.count(&:alive?)
      @logger&.debug("Waiting for #{alive_count} deferred threads to complete...")
      sleep 0.5
    end
    
    # Force kill any remaining threads
    @deferred_threads.each do |t|
      if t.alive?
        t.kill
        @logger&.info('delayed commit for buffer chunks was cancelled in shutdown chunk_id=unknown')
      end
    end
    
    @deferred_threads.clear
  end
  
  @ingester.shutdown if @ingester.respond_to?(:shutdown)
  super
end

#start ⇒ Object 



69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/fluent/plugin/out_kusto.rb', line 69

def start
  # Initialize output configuration and ingester
  super
  setup_outconfiguration
  setup_ingester_and_logger
  @table_name_template = @outconfiguration&.table_name
  @database_name = @outconfiguration&.database_name
  @shutdown_called = false
  @deferred_threads = []
  @plugin_start_time = Time.now
  @total_bytes_ingested = 0
end

#start_deferred_commit_thread(chunk_id, chunk, row_count, resolved_table) ⇒ Object 



250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
 
# File 'lib/fluent/plugin/out_kusto.rb', line 250

def start_deferred_commit_thread(chunk_id, chunk, row_count, resolved_table)
  # Start a thread to commit chunk after verifying ingestion
  return nil if @shutdown_called

  Thread.new do
    max_wait_time = @deferred_commit_timeout # Maximum wait time in seconds
    check_interval = 1 # Check every 1 second
    attempts = 0
    max_attempts = max_wait_time / check_interval

    loop do
      break if @shutdown_called
      
      attempts += 1
      
      if check_data_on_server(chunk_id, row_count, resolved_table)
        commit_write(chunk.unique_id)
        @logger&.info("Successfully committed chunk_id=#{chunk_id} after #{attempts} attempts")
        break
      end

      # If we've exceeded max attempts, commit anyway to avoid hanging
      if attempts >= max_attempts
        commit_write(chunk.unique_id)
        @logger&.warn("Force committing chunk_id=#{chunk_id} after #{max_wait_time}s timeout (#{attempts} verification attempts)")
        break
      end

      sleep check_interval
    end
  rescue StandardError => e
    @logger&.error("Error in deferred commit thread for chunk_id=#{chunk_id}: #{e}")
    # Ensure chunk is committed even on error to avoid hanging
    begin
      commit_write(chunk.unique_id)
      @logger&.warn("Force committed chunk_id=#{chunk_id} due to error in verification thread")
    rescue StandardError => commit_error
      @logger&.error("Failed to commit chunk_id=#{chunk_id} after thread error: #{commit_error}")
    end
  end
end

#try_write(chunk) ⇒ Object 



211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
 
# File 'lib/fluent/plugin/out_kusto.rb', line 211

def try_write(chunk)
  @deferred_threads ||= []
  tag = extract_tag_from_metadata(chunk.metadata)
  resolved_table = resolve_table_name(tag)
  safe_tag = tag.to_s.encode('UTF-8', invalid: :replace, undef: :replace, replace: '_').gsub(/[^0-9A-Za-z.-]/,
                                                                                             '_')
  chunk_id = dump_unique_id_hex(chunk.unique_id)
  ext = compression_enabled ? '.json.gz' : '.json'
  blob_name = "fluentd_event_#{safe_tag}_#{chunk_id}#{ext}"
  raw_data = chunk.read || ''
  records = raw_data.split("\n").map do |line|
    rec = JSON.parse(line)
    rec['record']['chunk_id'] = chunk_id if rec.is_a?(Hash) && rec['record'].is_a?(Hash)
    rec.to_json
  rescue StandardError
    line
  end
  updated_raw_data = records.join("\n")
  row_count = records.size
  data_to_upload = compression_enabled ? compress_data(updated_raw_data) : updated_raw_data
  begin
    @ingester.upload_data_to_blob_and_queue(data_to_upload, blob_name, @database_name, resolved_table,
                                            compression_enabled, @ingestion_mapping_reference)
    if @shutdown_called || !@delayed
      commit_write(chunk.unique_id)
      if @shutdown_called
        @logger&.info("Immediate commit for chunk_id=#{chunk_id} due to shutdown")
      else
        @logger&.info("Immediate commit for chunk_id=#{chunk_id} (delayed=false)")
      end
    else
      thread = start_deferred_commit_thread(chunk_id, chunk, row_count, resolved_table)
      @deferred_threads << thread if thread
    end
  rescue StandardError => e
    KustoErrorHandler.handle_try_write_error(@logger, e, chunk_id)
  end
end

#write(chunk) ⇒ Object 



178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/fluent/plugin/out_kusto.rb', line 178

def write(chunk)
  # Write a chunk of events to Kusto
  worker_id = Fluent::Engine.worker_id
  raw_data = chunk.read
  tag = extract_tag_from_metadata(chunk.metadata)
  resolved_table = resolve_table_name(tag)
  safe_tag = tag.to_s.encode('UTF-8', invalid: :replace, undef: :replace, replace: '_').gsub(/[^0-9A-Za-z.-]/,
                                                                                             '_')
  unique_id = chunk.unique_id
  ext = compression_enabled ? '.json.gz' : '.json'
  blob_name = "fluentd_event_worker#{worker_id}_#{safe_tag}_#{dump_unique_id_hex(unique_id)}#{ext}"
  data_to_upload = compression_enabled ? compress_data(raw_data) : raw_data
  begin
    @ingester.upload_data_to_blob_and_queue(data_to_upload, blob_name, @database_name, resolved_table,
                                            compression_enabled, @ingestion_mapping_reference)
  rescue StandardError => e
    handle_kusto_error(e, unique_id)
  end
end