fluent-plugin-azure-logs-ingestion

Fluentd output plugin that sends records to Log Analytics Workspace tables by using the Azure Monitor Logs Ingestion API.

• Ruby 2.4 or later
• Fluentd 1.x

fluent-gem install fluent-plugin-azure-logs-ingestion

If you use td-agent, use td-agent-gem instead of fluent-gem.

Add the following line to your Gemfile.

gem 'fluent-plugin-azure-logs-ingestion'

Then run bundle install.

Bundler can point directly at the GitHub repository. Specify ref when you want to pin a specific revision.

gem 'fluent-plugin-azure-logs-ingestion', git: 'https://github.com/fukasawah/fluent-plugin-azure-logs-ingestion.git', ref: '90782c8aad34a1101566909162d8cf02aa40c11a'

Then run bundle install.

<match azure.logs>
	@type azure_logs_ingestion
	endpoint https://example.japaneast-1.ingest.monitor.azure.com
	dcr_immutable_id dcr-000a00a000a00000a000000aa000a0aa
	stream_name Custom-MyTable

	tenant_id YOUR_TENANT_ID
	client_id YOUR_CLIENT_ID
	client_secret YOUR_CLIENT_SECRET

	<buffer>
		@type file
		path /var/log/fluent/azure-logs-ingestion-buffer.*.buf
		chunk_limit_size 900KB
	</buffer>
</match>

This plugin changes only the buffer defaults needed for a production-friendly file buffer and a chunk size that is likely to fit within the Logs Ingestion API request size limit.

Service principal credentials can be written directly in the Fluentd configuration or read from environment variables.

Available environment variables:

• AZURE_TENANT_ID
• AZURE_CLIENT_ID
• AZURE_CLIENT_SECRET

When using Managed Identity, specify use_msi true and omit tenant_id and client_secret. When using User-assigned Managed Identity, specify the User-assigned Managed Identity client ID in client_id.

<match azure.logs>
	@type azure_logs_ingestion
	endpoint https://example.japaneast-1.ingest.monitor.azure.com
	dcr_immutable_id dcr-000a00a000a00000a000000aa000a0aa
	stream_name Custom-MyTable
	use_msi true
	client_id YOUR_USER_ASSIGNED_MANAGED_IDENTITY_CLIENT_ID

	<buffer>
		@type file
		path /var/log/fluent/azure-logs-ingestion-buffer.*.buf
	</buffer>
</match>

• chunk_limit_size 900KB: The Logs Ingestion API request size limit is 1 MB. One chunk should fit in one request, and starting around 900 KB is safer because JSON serialization can increase the API request size.
• flush_mode and flush_interval use Fluentd defaults. If you need lower delivery latency, specify them explicitly as normal Fluentd buffer settings.

When sending to the Log Analytics Workspace Auxiliary tier without converting TimeGenerated in a DCR transformation, the range of TimeGenerated values in one request must be less than 30 minutes. To satisfy this limit, treat the original log timestamp as the Fluentd event time, then split chunks by time with <buffer time> and timekey.

For example, if the record field created_at is an ISO8601 string, convert it to event time with the input parser. Specify keep_time_key true when you also want to send created_at to Azure.

<source>
	@type tail
	path /var/log/myapp/app.log
	tag azure.logs

	<parse>
		@type json
		time_key created_at
		time_format %iso8601
		keep_time_key true
	</parse>
</source>

<match azure.logs>
	@type azure_logs_ingestion
	# ...
	<buffer time>
		@type file
		# ...
		timekey 20m
	</buffer>
</match>

If you need to replace the event time after a record has already been ingested, you can use renew_time_key in a filter. The field specified in renew_time_key must be a Unix timestamp.

<filter azure.logs>
	@type record_transformer
	renew_time_key created_at
</filter>


<match azure.logs>
	@type azure_logs_ingestion
	# ...
	<buffer time>
		@type file
		# ...
		timekey 20m
	</buffer>
</match>

The time in <buffer time> is the Fluentd event time, not a time field inside the record. Merely leaving created_at or TimeGenerated in the payload does not make it available for time-based chunking.

• This plugin does not rewrite TimeGenerated. If the payload has an original timestamp field such as time, prefer creating it in the DCR transformation, for example extend TimeGenerated = todatetime(['time']).
• HTTP 400, 401, 403, and 413 are treated as unrecoverable. 429 and 5xx are retried by Fluentd.

• Currently, when the Log Analytics Workspace SKU is Auxiliary tier and the DCR transformation is not used, TimeGenerated in one request must stay within less than 30 minutes.

  • This limit only applies when ingesting to Auxiliary log tables. If the source entries for TimeGenerated are ingested without being transformed, the range of entries must be less than 30 minutes.

    https://learn.microsoft.com/en-us/azure/azure-monitor/fundamentals/service-limits#logs-ingestion-api

• Logs Ingestion API request size must be kept to 1 MB or less.

  • Maximum size of API call | 1 MB

    https://learn.microsoft.com/en-us/azure/azure-monitor/fundamentals/service-limits#logs-ingestion-api

• Log Analytics Workspace has no deduplication mechanism. If Azure accepts a request but Fluentd cannot confirm the response successfully, retrying can create duplicate records.

• Azure Monitor Logs Ingestion API overview: https://learn.microsoft.com/azure/azure-monitor/logs/logs-ingestion-api-overview
• Create data collection rules (DCRs) using JSON: https://learn.microsoft.com/azure/azure-monitor/data-collection/data-collection-rule-create-edit
• Azure DCR structure: https://learn.microsoft.com/azure/azure-monitor/data-collection/data-collection-rule-structure
• Azure custom tables and _CL suffix: https://learn.microsoft.com/azure/azure-monitor/logs/create-custom-table
• Managed identity on Azure VM: https://learn.microsoft.com/entra/identity/managed-identities-azure-resources/how-to-use-vm-token
• Managed identity on App Service / Functions: https://learn.microsoft.com/azure/app-service/overview-managed-identity
• Fluentd output plugin API: https://docs.fluentd.org/plugin-development/api-plugin-output

bundle install
bundle exec rake test

This project uses a single Gemfile. Use rbenv and FLUENTD_VERSION when checking older Ruby / Fluentd combinations locally.

rbenv init
source ~/.bashrc

(
	export RBENV_VERSION=2.4.10 FLUENTD_VERSION=1.15.3
	unset GEM_HOME GEM_PATH MY_RUBY_HOME
	rbenv install "$RBENV_VERSION" -s
	rbenv exec gem install bundler -v 2.3.27 --no-document
	rbenv exec bundle _2.3.27_ install
	rbenv exec bundle _2.3.27_ exec rake test
)

(
	export RBENV_VERSION=2.7.8 FLUENTD_VERSION=1.18.0
	unset GEM_HOME GEM_PATH MY_RUBY_HOME
	rbenv install "$RBENV_VERSION" -s
	rbenv exec gem install bundler -v 2.3.27 --no-document
	rbenv exec bundle _2.3.27_ install
	rbenv exec bundle _2.3.27_ exec rake test
)

Publish one RubyGems artifact. Compatibility is checked with tests; build the distributable gem once with the current Ruby.

bundle install
bundle exec rake test
bundle exec gem build fluent-plugin-azure-logs-ingestion.gemspec --strict
gem push fluent-plugin-azure-logs-ingestion-*.gem