Fluentd output plugin for ingesting logs and data into Azure Data Explorer (Kusto).

This plugin allows you to send data from Fluentd to Azure Data Explorer (Kusto) using Azure Blob Storage and Queue for scalable, reliable ingestion. It supports both buffered and non-buffered modes, handles authentication via Azure AD or Managed Identity, and provides robust error handling and logging.

• Ruby 2.5 or later
  Check version (Windows/Linux):

  ruby --version

  Install on Ubuntu/Linux:

  sudo apt-get install ruby-full

  Install on Windows (using RubyInstaller): Download RubyInstaller Official Ruby installation guide

• Fluentd v1.0 or later
  Check version (Windows/Linux):

  fluentd --version

  Install on Ubuntu/Linux:

  gem install fluentd

  Install on Windows (in Command Prompt after Ruby is installed):

  gem install fluentd

  Official Fluentd installation guide

$ gem install fluent-plugin-kusto 

Add the following line to your Gemfile:

gem "fluent-plugin-kusto", "~> 1.0.0"

And then execute:

$ bundle

The Kusto output plugin lets you ingest your logs into an Azure Data Explorer cluster, using the Queued Ingestion mechanism.

Create an Azure Data Explorer cluster in one of the following ways:

• Create a free-tier cluster
• Create a fully featured cluster

FluentD uses the Azure application's credentials to ingest data into your cluster.

• Register an application
• Add a client secret
• Authorize the app in your database

• System-assigned Managed Identity:

  • Enable system-assigned managed identity

• User-assigned Managed Identity:

  • Create and assign user-assigned managed identity

• Grant Permissions:

  • Assign permissions to managed identity for Azure Data Explorer

• Follow workload_identity.md

To create a new database in your Azure Data Explorer cluster, use the following KQL command:

.create database <database_name>

Fluent Bit ingests the event data into Kusto in a JSON format. By default, the table includes 3 properties:

• record - the actual event payload.
• tag - the event tag.
• timestamp - the event timestamp.

A table with the expected schema must exist in order for data to be ingested properly.

.create table <table_name> (tag:string, timestamp:datetime, record:dynamic)

This plugin supports four authentication methods for connecting to Azure Data Explorer:

Traditional client credentials flow using Azure AD app registration. Best for CI/CD pipelines and traditional applications.

Required Parameters:

• auth_type: aad
• tenant_id: Your Azure AD tenant ID
• client_id: The Azure AD application client ID
• client_secret: The Azure AD application client secret

Uses the system-assigned managed identity of Azure resources (VMs, App Services, AKS nodes). No secrets to manage.

Required Parameters:

• auth_type: system_managed_identity
• managed_identity_client_id: Set to SYSTEM

Uses a user-assigned managed identity. Allows sharing the same identity across multiple Azure resources.

Required Parameters:

• auth_type: user_managed_identity
• managed_identity_client_id: The client ID (GUID) of the user-assigned managed identity

Modern approach for Kubernetes/AKS workloads. Replaces the legacy Pod Identity system using OIDC federation.

Required Parameters:

• auth_type: workload_identity
• workload_identity_client_id: The client ID for workload identity
• workload_identity_tenant_id: The tenant ID for workload identity
• workload_identity_token_file_path: Path to the workload identity token file (optional, defaults to /var/run/secrets/azure/tokens/azure-identity-token)

The plugin uses a standardized 3-column schema for all ingested data:

You can now use pre-defined ingestion mappings in Kusto to transform data during ingestion by setting the ingestion_mapping_reference parameter. This allows you to:

• Transform the default 3-column format into your desired schema
• Apply data transformations during ingestion for better performance
• Use Kusto's native ingestion mapping capabilities

Example:

<match test.kusto>
  @type kusto
  # ... other configuration ...
  ingestion_mapping_reference my_custom_mapping
</match>

Then create the mapping in Kusto:

.create table MyTable ingestion json mapping "my_custom_mapping" 
@'[
  {"column":"EventTime", "path":"$.timestamp", "datatype":"datetime"},
  {"column":"Source", "path":"$.tag", "datatype":"string"},
  {"column":"Level", "path":"$.record.level", "datatype":"string"},
  {"column":"Message", "path":"$.record.message", "datatype":"string"}
]'

If you prefer not to use ingestion mappings, you can still use this pattern for schema transformation:

-- 1. Create landing table (matches plugin output)
.create table RawLogs (tag:string, timestamp:datetime, record:dynamic)

-- 2. Create your target table with desired schema
.create table ProcessedLogs (
    EventTime: datetime,
    Source: string,
    Level: string,
    Message: string,
    UserId: string,
    Properties: dynamic
)

-- 3. Create update policy to transform data
.alter table ProcessedLogs policy update
@'[{
    "IsEnabled": true,
    "Source": "RawLogs",
    "Query": "RawLogs | extend EventTime=timestamp, Source=tag, Level=tostring(record.level), Message=tostring(record.message), UserId=tostring(record.userId), Properties=record.properties | project EventTime, Source, Level, Message, UserId, Properties",
    "IsTransactional": true,
    "PropagateIngestionProperties": false
}]'

This approach provides flexibility to transform the generic 3-column format into any schema you need.

<match test.kusto>
  @type kusto
  @log_level debug
  
  # Authentication - Azure AD
  auth_type aad
  tenant_id 12345678-1234-1234-1234-123456789abc
  client_id 87654321-4321-4321-4321-abcdef123456
  client_secret your-app-secret-here
  
  # Kusto connection
  endpoint https://mycluster.eastus.kusto.windows.net
  database_name MyDatabase
  table_name MyLogs
  
  # Optional settings
  azure_cloud AzureCloud
  compression_enabled true
  buffered true
  delayed false
  
  <buffer>
    @type memory
    timekey 1m
    timekey_wait 30s
    flush_interval 10s
  </buffer>
</match>

<match test.kusto>
  @type kusto
  @log_level debug
  
  # Authentication - System Managed Identity
  auth_type system_managed_identity
  managed_identity_client_id SYSTEM
  
  # Kusto connection
  endpoint https://mycluster.eastus.kusto.windows.net
  database_name MyDatabase
  table_name MyLogs
  
  # Optional settings
  azure_cloud AzureCloud
  compression_enabled true
  buffered true
  delayed false
  
  <buffer>
    @type memory
    timekey 1m
    timekey_wait 30s
    flush_interval 10s
  </buffer>
</match>

<match test.kusto>
  @type kusto
  @log_level debug
  
  # Authentication - User Managed Identity
  auth_type user_managed_identity
  managed_identity_client_id 11111111-2222-3333-4444-555555555555
  
  # Kusto connection
  endpoint https://mycluster.eastus.kusto.windows.net
  database_name MyDatabase
  table_name MyLogs
  
  # Optional settings
  azure_cloud AzureCloud
  compression_enabled true
  buffered true
  delayed false
  
  <buffer>
    @type memory
    timekey 1m
    timekey_wait 30s
    flush_interval 10s
  </buffer>
</match>

<match test.kusto>
  @type kusto
  @log_level debug
  
  # Authentication - Workload Identity
  auth_type workload_identity
  workload_identity_client_id 99999999-8888-7777-6666-555555555555
  workload_identity_tenant_id 12345678-1234-1234-1234-123456789abc
  workload_identity_token_file_path /var/run/secrets/azure/tokens/azure-identity-token
  
  # Kusto connection
  endpoint https://mycluster.eastus.kusto.windows.net
  database_name MyDatabase
  table_name MyLogs
  
  # Optional settings
  azure_cloud AzureCloud
  compression_enabled true
  buffered true
  delayed false
  
  <buffer>
    @type memory
    timekey 1m
    timekey_wait 30s
    flush_interval 10s
  </buffer>
</match>

This diagram shows the main components and data flow for the plugin, including configuration, error handling, token management, and Azure resource interactions.

• Fixed critical authentication initialization bugs - Resolved NameError in ManagedIdentityTokenProvider and WorkloadIdentityTokenProvider
• Added comprehensive unit test coverage - New test suites for authentication providers with 14 test cases and 45+ assertions
• Improved E2E test reliability - Enhanced timeout configurations to handle Azure Kusto ingestion delays (480s-600s timeouts)
• Enhanced authentication stability - All authentication methods now properly validated: AAD, System/User Managed Identity, Workload Identity, Azure CLI

• License: Apache License, Version 2.0