Top Level Namespace

Defined Under Namespace

Modules: Fluent, KustoConstants Classes: AadTokenProvider, AbstractTokenProvider, AzCliTokenProvider, Client, Ingester, KustoErrorHandler, ManagedIdentityTokenProvider, OutputConfiguration, WorkloadIdentity

Constant Summary collapse

AZURE_CLOUDS = 
{
  'AzureCloud' => { 'aad' => 'https://login.microsoftonline.com' },
  'AzureChinaCloud' => { 'aad' => 'https://login.chinacloudapi.cn' },
  'AzureUSGovernment' => { 'aad' => 'https://login.microsoftonline.us' },
  'AzureUSGovernmentCloud' => { 'aad' => 'https://login.microsoftonline.us' }
}.freeze

Instance Method Summary collapse

Instance Method Details

#run_kusto_api_query(query, data_endpoint, token_provider, use_ingest_endpoint: false, database_name: nil) ⇒ Object

Runs a Kusto API query against the specified endpoint. Handles both management and query endpoints, builds request, and parses response.



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/fluent/plugin/kusto_query.rb', line 20

def run_kusto_api_query(query, data_endpoint, token_provider, use_ingest_endpoint: false, database_name: nil)
  access_token = token_provider.get_token
  endpoint = use_ingest_endpoint ? to_ingest_endpoint(data_endpoint) : data_endpoint
  path = use_ingest_endpoint ? '/v1/rest/mgmt' : '/v1/rest/query'
  uri = URI("#{endpoint}#{path}")

  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = true
  # Add timeouts to prevent hanging connections
  http.open_timeout = 10
  http.read_timeout = 30
  http.write_timeout = 10

  headers = {
    'Authorization' => "Bearer #{access_token}",
    'Content-Type' => 'application/json',
    'Accept' => 'application/json',
    'x-ms-client-version' => "Kusto.FluentD:#{Fluent::Plugin::Kusto::VERSION}",
    'x-ms-app' => 'Kusto.FluentD',
    'x-ms-user' => 'Kusto.FluentD'
  }

  body_hash = { csl: query }
  body_hash[:db] = database_name if database_name
  body = body_hash.to_json

  request = Net::HTTP::Post.new(uri.request_uri, headers)
  request.body = body

  response = http.request(request)
  unless response.code.to_i.between?(200, 299)
    # Print error details if query fails
    puts "Kusto query failed with status #{response.code}:"
    puts response.body
    begin
      error_handler = defined?(KustoErrorHandler) ? KustoErrorHandler.new(response.body) : nil
      puts "Permanent Kusto error: #{error_handler.message}" if error_handler&.permanent_error?
    rescue StandardError => e
      puts "Failed to parse error response with KustoErrorHandler: #{e.message}"
    end
    return response
  end

  begin
    # Parse and return rows from response JSON
    response_json = JSON.parse(response.body)
    tables = response_json['Tables']
    rows = tables && tables[0] && tables[0]['Rows']
    rows || []
  rescue JSON::ParserError => e
    puts "Failed to parse JSON: #{e}"
    puts response.body
    response
  end
end

#to_ingest_endpoint(data_endpoint) ⇒ Object 



13
14
15
16
 
# File 'lib/fluent/plugin/kusto_query.rb', line 13

def to_ingest_endpoint(data_endpoint)
  # Convert a Kusto data endpoint to its corresponding ingest endpoint
  data_endpoint.sub(%r{^https://}, 'https://ingest-')
end