# Azure Data Explorer The *Kusto* output plugin lets you ingest your logs into an [Azure Data Explorer](https://azure.microsoft.com/en-us/products/data-explorer/) cluster, using the [Queued Ingestion](https://learn.microsoft.com/en-us/kusto/api/netfx/about-kusto-ingest?view=azure-data-explorer\&preserve-view=true\&tabs=csharp#queued-ingestion) mechanism. This output plugin can also be used to ingest logs into an [Eventhouse](https://blog.fabric.microsoft.com/en-us/blog/eventhouse-overview-handling-real-time-data-with-microsoft-fabric/) cluster in Microsoft Fabric Real Time Analytics. ## Authentication Methods Fluent-Bit can use various authentication methods to connect to your Azure Data Explorer cluster: ### Service Principal Authentication (Default) For service principal authentication, you'll need to create an Azure AD application: * [Register an Application](https://docs.microsoft.com/en-us/azure/active-directory/develop/quickstart-register-app#register-an-application) * [Add a client secret](https://docs.microsoft.com/en-us/azure/active-directory/develop/quickstart-register-app#add-a-client-secret) * [Authorize the app in your database](https://docs.microsoft.com/en-us/azure/data-explorer/kusto/management/access-control/principals-and-identity-providers#azure-ad-tenants) Configure Fluent Bit with your application's `tenant_id`, `client_id`, and `client_secret`. ### Managed Identity Authentication When running on Azure services that support Managed Identities (such as Azure VMs, AKS, or App Service): 1. [Assign the managed identity appropriate permissions to your Kusto database](https://learn.microsoft.com/en-us/azure/data-explorer/configure-managed-identities-cluster) 2. Configure Fluent Bit with `auth_type` set to `managed_identity` 3. For system-assigned identity, set `client_id` to `system` 4. For user-assigned identity, set `client_id` to the managed identity's client ID (GUID) ### Workload Identity Authentication For Kubernetes environments using Azure Workload Identity: 1. [Set up Azure Workload Identity in your Kubernetes cluster](https://learn.microsoft.com/en-us/azure/aks/workload-identity-deploy-cluster) 2. Configure your pod to use a service account with Workload Identity Federation 3. Configure Fluent Bit with: * `auth_type` set to `workload_identity` * `tenant_id` and `client_id` of your Azure AD application * `workload_identity_token_file` pointing to your token file path (typically `/var/run/secrets/azure/tokens/azure-identity-token`) ## For ingesting into Azure Data Explorer: Creating a Kusto Cluster and Database Create an Azure Data Explorer cluster in one of the following ways: * [Create a free-tier cluster](https://dataexplorer.azure.com/freecluster) * [Create a fully featured cluster](https://learn.microsoft.com/en-us/azure/data-explorer/create-cluster-and-database?tabs=free) ## Ingest into Microsoft Fabric real time analytics: create an Eventhouse cluster and KQL database Create an Eventhouse cluster and a KQL database using the following steps: * [Create an Eventhouse cluster](https://learn.microsoft.com/en-us/training/modules/query-data-kql-database-microsoft-fabric/) * [Create a KQL database](https://learn.microsoft.com/en-us/training/modules/query-data-kql-database-microsoft-fabric/) ## Create an Azure registered application Fluent Bit uses the application's credentials to ingest data into your cluster. * [Register an application](https://docs.microsoft.com/en-us/azure/active-directory/develop/quickstart-register-app#register-an-application) * [Add a client secret](https://docs.microsoft.com/en-us/azure/active-directory/develop/quickstart-register-app#add-a-client-secret) * [Authorize the app in your database](https://learn.microsoft.com/en-us/azure/app-service/overview-authentication-authorization) ## Create a table Fluent Bit ingests the event data into Kusto in a JSON format. By default, the table includes 3 properties: * `log` - 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. ```shell .create table FluentBit (log:dynamic, tag:string, timestamp:datetime) ``` ## Optional - create an ingestion mapping By default, Kusto will insert incoming ingestion data into a table by inferring the mapped table columns, from the payload properties. However, this mapping can be customized by creating a [JSON ingestion mapping](https://docs.microsoft.com/en-us/azure/data-explorer/kusto/management/mappings#json-mapping). The plugin can be configured to use an ingestion mapping using the `ingestion_mapping_reference` configuration key. ## Configuration parameters | Key | Description | Default | | -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------ | | `alias` | Sets an alias, use to define multiple instances of the same output plugin. | *none* | | `auth_type` | Set the authentication type: `service_principal`, `managed_identity`, or `workload_identity`. For `managed_identity`, use `system` as `client_id` for `system-assigned identity`, or specify the managed identity's client ID. | `service_principal` | | `azure_kusto_buffer_key` | When buffering is `true`, set the Azure Kusto buffer key which must be specified when using multiple instances of Azure Kusto output plugin and buffering is enabled. | `key` | | `blob_uri_length` | Set the length of generated blob URI before ingesting to Kusto. | `64` | | `buffer_dir` | When buffering is `true`, specifies the location of directory where the buffered data will be stored. | `/tmp/fluent-bit/azure-kusto/` | | `buffer_file_delete_early` | When buffering is `true`, whether to delete the buffered file early after successful blob creation. | `false` | | `buffering_enabled` | Enable buffering into disk before ingesting into Azure Kusto. | `false` | | `client_id` | Required if `managed_identity_client_id` isn't set. The client ID of the AAD registered application. | *none* | | `client_secret` | Set the client secret (Application Password) of the AAD application used for authentication. | *none* | | `compression_enabled` | If enabled, sends compressed HTTP payload (gzip) to Kusto. | `true` | | `database_name` | The database name. | *none* | | `delete_on_max_upload_error` | When buffering is `true`, whether to delete the buffer file on maximum upload errors. | `false` | | `host` | IP address or hostname of the target HTTP server. | `127.0.0.1` | | `include_tag_key` | If enabled, a tag is appended to output. The key name is used `tag_key` property. | `true` | | `include_time_key` | If enabled, a timestamp is appended to output. The key name is used `time_key` property. | `true` | | `ingestion_endpoint` | The cluster's ingestion endpoint, usually in the form `https://ingest-cluster_name.region.kusto.windows.net`. | *none* | | `ingestion_endpoint_connect_timeout` | The connection timeout of various Kusto endpoints in seconds. | `60` | | `ingestion_mapping_reference` | The name of a [JSON ingestion mapping](https://docs.microsoft.com/en-us/azure/data-explorer/kusto/management/mappings#json-mapping) that will be used to map the ingested payload into the table columns. | *none* | | `ingestion_resources_refresh_interval` | Set the Azure Kusto ingestion resources refresh interval. | `3600` | | `io_timeout` | Configure the HTTP IO timeout for uploads. | `60s` | | `log_key` | Key name of the log content. | `log` | | `log_level` | Specifies the log level for output plugin. If not set here, plugin uses global log level in `service` section. | `info` | | `log_supress_interval` | Suppresses log messages from output plugin that appear similar within a specified time interval. `0` no suppression. | `0` | | `match` | Set a tag pattern to match records that output should process. Exact matches or wildcards (for example `*`). | *none* | | `match_regex` | Set a regular expression to match tags for output routing. This allows more flexible matching compared to wildcards. | *none* | | `net.connect_timeout` | Set maximum time allowed to establish a connection, this time includes the TLS handshake. | `10s` | | `net.connect_timeout_log_error` | On connection timeout, specify if it should log an error. When disabled, the timeout is logged as a debug message. | `true` | | `net.dns.mode` | Select the primary DNS connection type (TCP or UDP). | *none* | | `net.dns.prefer_ipv4` | Select the primary DNS resolver type (LEGACY or ASYNC). | `false` | | `net.dns.prefer_ipv6` | Prioritize IPv6 DNS results when trying to establish a connection. | `false` | | `net.dns.resolver` | Select the primary DNS resolver type (LEGACY or ASYNC). | *none* | | `net.io_timeout` | Set maximum time a connection can stay idle while assigned. | `0s` | | `net.keepalive` | Enable or disable `Keepalive` support. | `false` | | `net.keepalive_idle_timeout` | Set maximum time allowed for an idle `Keepalive` connection.. | `false` | | `net.keepalive_max_recycle` | Set maximum number of times a keepalive connection can be used before it's retried. | `2000` | | `net.max_worker_connections` | Set the maximum number of active TCP connections that can be used per worker thread. | `0` | | `net.proxy_env_ignore` | Ignore the environment variables `HTTP_PROXY`, `HTTPS_PROXY` and `NO_PROXY` when set. | `false` | | `net.source_address` | Specify network address to bind for data traffic. | *none* | | `net.tcp_keepalive` | Enable or disable Keepalive support. | `off` | | `net.tcp_keepalive_interval` | Interval between TCP keepalive probes when no response is received on a `keepidle` probe. | `-1` | | `net.tcp_keepalive_probes` | Number of unacknowledged probes to consider a connection dead. | `-1` | | `net.tcp_keepalive_time` | Interval between the last data packet sent and the first TCP keepalive probe. | `-1` | | `port` | TCP port of the target HTTP server. | `0` | | `retry_limit` | Set retry limit for output plugin when delivery fails. Integer, `no_limits`, `false`, or `off` to disable, or `no_retries` to disable retries entirely. | `1` | | `scheduler_max_retries` | Optional. When buffering is `On`, set the maximum number of retries for ingestion using the scheduler. | `3` | | `store_dir_limit_size` | When buffering is `true`, set the max size of the buffer directory. | `8G` | | `table_name` | The table name. | *none* | | `tag_key` | The key name of tag. If `include_tag_key` is false, this property is ignored. | `tag` | | `tenant_id` | Required if `managed_identity_client_id` isn't set. The tenant/domain ID of the AAD registered application. | *none* | | `time_key` | The key name of time. If `include_time_key` is false, this property is ignored. | `timestamp` | | `tls` | Enable or disable TLS/SSL support. | `off` | | `tls.ca_file` | Absolute path to CA certificate file. | *none* | | `tls.ca_path` | Absolute path to scan for certificate files. | *none* | | `tls.ciphers` | Specify TLS ciphers up to TLSv1.2. | *none* | | `tls.crt_file` | Absolute path to Certificate file. | *none* | | `tls.debug` | Set TLS debug level. Accepts `0` (No debug), `1`(Error), `2` (State change), `3` (Informational) and `4` (Verbose). | `1` | | `tls.key_file` | Absolute path to private Key file. | *none* | | `tls.key_passwd` | Optional password for tls.key\_file file. | *none* | | `tls.max_version` | Specify the maximum version of TLS. | *none* | | `tls.min_version` | Specify the minimum version of TLS. | *none* | | `tls.verify` | Force certificate validation. | `on` | | `tls.verify_hostname` | Enable or disable to verify hostname. | `off` | | `tls.vhost` | Hostname to be used for TLS SNI extension. | *none* | | `tls.windows.certstore_name` | Sets the `certstore` name on an output (Windows). | *none* | | `tls.windows.use_enterprise_store` | Sets whether using enterprise certificate store or not on an output (Windows). | *none* | | `unify_tag` | This creates a single buffer file when the buffering mode is `true`. | `true` | | `upload_file_size` | Specifies the size of files to be uploaded in megabytes. | `200MB` | | `upload_timeout` | Optionally specify a timeout for uploads. Fluent Bit will start ingesting buffer files which have been created more than x minutes and haven't reached `upload_file_size` limit yet. | `30m` | | `workload_identity_token_file` | Set the token path for workload identity authentication. | *none* | ### Configuration file Get started with this configuration file: {% tabs %} {% tab title="fluent-bit.yaml" %} ```yaml service: flush: 1 log_level: info pipeline: inputs: - name: dummy dummy: '{"name": "Fluent Bit", "year": 2020}' samples: 1 tag: var.log.containers.app-default-96cbdef2340.log outputs: - name: azure_kusto match: '*' tenant_id: client_id: client_secret: ingestion_endpoint: https://ingest-..kusto.windows.net database_name: table_name: ingestion_mapping_reference: ingestion_endpoint_connect_timeout: compression_enabled: ingestion_resources_refresh_interval: buffering_enabled: on upload_timeout: 2m upload_file_size: 125M azure_kusto_buffer_key: kusto1 buffer_file_delete_early: off unify_tag: on buffer_dir: /var/log/ store_dir_limit_size: 16GB blob_uri_length: 128 scheduler_max_retries: 3 delete_on_max_upload_error: off io_timeout: 60s ``` {% endtab %} {% tab title="fluent-bit.conf" %} ``` [OUTPUT] Name azure_kusto Match * Tenant_Id Client_Id Client_Secret Ingestion_Endpoint https://ingest-..kusto.windows.net Database_Name Table_Name Ingestion_Mapping_Reference Ingestion_Endpoint_Connect_Timeout Compression_Enabled Ingestion_Resources_Refresh_Interval Buffering_Enabled On Upload_Timeout 2m Upload_File_Size 125M Azure_Kusto_Buffer_Key kusto1 Buffer_File_Delete_Early Off Unify_Tag On Buffer_Dir /var/log/ Store_Dir_Limit_Size 16GB Blob_Uri_Length 128 Scheduler_Max_Retries 3 Delete_On_Max_Upload_Error Off Io_Timeout 60s ``` **Managed Identity Authentication** ``` [OUTPUT] Match * Name azure_kusto Auth_Type managed_identity Client_Id # Use 'system' for system-assigned managed identity Ingestion_Endpoint https://ingest-..kusto.windows.net Database_Name Table_Name # Additional parameters as needed ``` **Workload Identity Authentication** ``` [OUTPUT] Match * Name azure_kust Auth_Type workload_identity Tenant_Id Client_Id Workload_Identity_Token_File /var/run/secrets/azure/tokens/azure-identity-token Ingestion_Endpoint https://ingest-..kusto.windows.net Database_Name Table_Name # Additional parameters as needed ``` {% endtab %} {% endtabs %} ## Troubleshooting ### `403 Forbidden` If you get a `403 Forbidden` error response, make sure that: * You provided the correct AAD registered application credentials. * You authorized the application to ingest into your database or table. --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://docs.fluentbit.io/manual/data-pipeline/outputs/azure_kusto.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.