Class: KustoErrorHandler

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

Overview

KustoErrorHandler parses and classifies errors returned from Azure Data Explorer (Kusto) API responses. Provides methods for error extraction, classification, and logging.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(error_response) ⇒ KustoErrorHandler

Returns a new instance of KustoErrorHandler.



8
9
10
11
12
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 8

def initialize(error_response)
  # Parse error response and extract message
  @error = parse_error(error_response)
  @message = @error['Message'] || @error['message'] || error_response
end

Instance Attribute Details

#message ⇒ Object (readonly)

Returns the value of attribute message.



6
7
8
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 6

def message
  @message
end

Class Method Details

.extract_kusto_error_type(error) ⇒ Object

rubocop:disable Metrics/MethodLength Extracts the Kusto error type (code) from an exception or error response



21
22
23
24
25
26
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 21

def self.extract_kusto_error_type(error)
  error_json = parse_error_json(error)
  return error_json['error']['code'] if error_json && error_json['error'] && error_json['error']['code']

  nil
end

.from_kusto_error_type(error_type, message) ⇒ Object

Factory method to create a KustoErrorHandler from error type and message



47
48
49
50
51
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 47

def self.from_kusto_error_type(error_type, message)
  # You can expand this logic to map error_type to more structured fields if needed
  error_response = { 'error' => { 'code' => error_type, 'message' => message } }.to_json
  new(error_response)
end

.handle_kusto_error(logger, e, unique_id) ⇒ Object

Handles Kusto errors and logs them appropriately



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 54

def self.handle_kusto_error(logger, e, unique_id)
  kusto_error_type = extract_kusto_error_type(e)
  if kusto_error_type
    kusto_error = from_kusto_error_type(kusto_error_type, e.message)
    log_kusto_data_error(logger, kusto_error)
    log_kusto_drop_chunk(logger, kusto_error, unique_id) if kusto_error.is_permanent?
    # Always raise the custom error if present
    raise kusto_error if kusto_error.is_a?(StandardError)
    raise kusto_error unless kusto_error.is_permanent?

    nil
  else
    log_failed_ingest(logger, unique_id, e)
    raise
  end
end

.handle_try_write_error(logger, e, chunk_id) ⇒ Object

Handles errors during try_write and logs them appropriately



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 72

def self.handle_try_write_error(logger, e, chunk_id)
  kusto_error_type = extract_kusto_error_type(e)
  if kusto_error_type
    kusto_error = from_kusto_error_type(kusto_error_type, e.message)
    log_kusto_data_error(logger, kusto_error)
    if kusto_error.is_permanent?
      log_kusto_drop_chunk(logger, kusto_error, chunk_id)
      return nil
    end
    # Always raise the custom error if present
    raise kusto_error if kusto_error.is_a?(StandardError)
    raise kusto_error unless kusto_error.is_permanent?

    nil
  else
    logger.error(
      "Failed to ingest chunk #{chunk_id}: #{e.full_message}"
    )
    raise
  end
end

.log_failed_ingest(logger, unique_id, e) ⇒ Object

Log failed ingestion event



111
112
113
114
115
116
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 111

def self.log_failed_ingest(logger, unique_id, e)
  logger.error(
    "Failed to ingest event to Kusto : #{unique_id}\n" \
    "#{e.full_message}"
  )
end

.log_kusto_data_error(logger, kusto_error) ⇒ Object

Log details of a Kusto data error



95
96
97
98
99
100
101
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 95

def self.log_kusto_data_error(logger, kusto_error)
  logger.error(
    "KustoDataError: #{kusto_error.message} " \
    "(Code: #{kusto_error.failure_code}, Reason: #{kusto_error.failure_sub_code}, " \
    "Permanent: #{kusto_error.is_permanent?})"
  )
end

.log_kusto_drop_chunk(logger, kusto_error, chunk_id) ⇒ Object

Log when a chunk is dropped due to a permanent error



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

def self.log_kusto_drop_chunk(logger, kusto_error, chunk_id)
  logger.error(
    "Dropping chunk #{chunk_id} due to permanent Kusto error: #{kusto_error.message}"
  )
end

.parse_error_json(error) ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 28

def self.parse_error_json(error)
  # Parse error JSON from string or exception
  if error.is_a?(String)
    begin
      JSON.parse(error)
    rescue StandardError
      nil
    end
  elsif error.respond_to?(:message)
    begin
      JSON.parse(error.message)
    rescue StandardError
      nil
    end
  end
end

Instance Method Details

#permanent_error? ⇒ Boolean

Returns:

  • (Boolean) 


14
15
16
17
 
# File 'lib/fluent/plugin/kusto_error_handler.rb', line 14

def permanent_error?
  # Check if error is marked as permanent
  @error && [true, 'true'].include?(@error['@permanent'])
end