Azure Event Hubs plugin

The plugin uses the connection string to access Azure Events Hubs. Find the connection string here: Azure Portal-> Event Hub -> Shared access polices. The event_hub_connections option passes the Event Hub connection strings for the basic configuration.

Sample connection string:

Endpoint=sb://logstash.servicebus.windows.net/;SharedAccessKeyName=activity-log-read-only;SharedAccessKey=mm6AbDcEfj8lk7sjsbzoTJ10qAkiSaG663YykEAG2eg=;EntityPath=insights-operational-logs

Azure Blob Storage account is an essential part of Azure-to-Logstash configuration. A Blob Storage account is a central location that enables multiple instances of Logstash to work together to process events. It records the offset (location) of processed events. On restart, Logstash resumes processing exactly where it left off.

Configuration notes:

Sample Blob Storage connection string:

DefaultEndpointsProtocol=https;AccountName=logstash;AccountKey=ETOPnkd/hDAWidkEpPZDiXffQPku/SZdXhPSLnfqdRTalssdEuPkZwIcouzXjCLb/xPZjzhmHfwRCGo0SBSw==;EndpointSuffix=core.windows.net

Find the connection string to Blob Storage here: Azure Portal-> Blob Storage account -> Access keys.

Here are some guidelines to help you avoid data conflicts that can cause lost events.

Basic configuration is the default and supports consuming from multiple Event Hubs. All Events Hubs, except for the connection string, share the same configuration.

You supply a list of Event Hub connection strings, complete with the Event Hub EntityPath that defines the Event Hub name. All other configuration settings are shared.

input {
   azure_event_hubs {
      event_hub_connections => ["Endpoint=sb://example1...EntityPath=insights-logs-errors", "Endpoint=sb://example2...EntityPath=insights-metrics-pt1m"]
      threads => 8
      decorate_events => true
      consumer_group => "logstash"
      storage_connection => "DefaultEndpointsProtocol=https;AccountName=example...."
   }
}

The advanced configuration model accommodates deployments where different Event Hubs require different configurations. Options can be configured per Event Hub. You provide a list of Event Hub names through the event_hubs option. Under each name, specify the configuration for that Event Hub. Options can be defined globally or expressed per Event Hub.

If the same configuration option appears in both the global and event_hub sections, the more specific (event_hub) setting takes precedence.

input {
   azure_event_hubs {
     config_mode => "advanced"
     threads => 8
     decorate_events => true
     storage_connection => "DefaultEndpointsProtocol=https;AccountName=example...."
     event_hubs => [
        {"insights-operational-logs" => {
         event_hub_connection => "Endpoint=sb://example1..."
         initial_position => "beginning"
         consumer_group => "iam_team"
        }},
      {"insights-metrics-pt1m" => {
         event_hub_connection => "Endpoint=sb://example2..."
         initial_position => "end"
         consumer_group => "db_team"
       }}
     ]
   }
}

In this example, storage_connection and decorate_events are applied globally. The two Event Hubs each have their own settings for consumer_groups and initial_position.

Sets configuration to either Basic configuration (default) or Advanced configuration.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"  , "Endpoint=sb://example2...;EntityPath=event_hub_name2"  ]
}

Defines the Event Hubs to be read. An array of hashes where each entry is a hash of the Event Hub name and its configuration options.

azure_event_hubs {
  config_mode => "advanced"
  event_hubs => [
      { "event_hub_name1" => {
          event_hub_connection => "Endpoint=sb://example1..."
      }},
      { "event_hub_name2" => {
          event_hub_connection => "Endpoint=sb://example2..."
          storage_connection => "DefaultEndpointsProtocol=https;AccountName=example...."
          storage_container => "my_container"
     }}
   ]
   consumer_group => "logstash" # shared across all Event Hubs
}

List of connection strings that identifies the Event Hubs to be read. Connection strings include the EntityPath for the Event Hub.

The event_hub_connections option is defined per Event Hub. All other configuration options are shared among Event Hubs.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"  , "Endpoint=sb://example2...;EntityPath=event_hub_name2"  ]
}

Connection string that identifies the Event Hub to be read. Advanced configuration options can be set per Event Hub. This option modifies event_hub_name, and should be nested under it. (See sample.) This option accepts only one connection string.

azure_event_hubs {
   config_mode => "advanced"
   event_hubs => [
     { "event_hub_name1" => {
        event_hub_connection => "Endpoint=sb://example1...;EntityPath=event_hub_name1"
     }}
   ]
}

Interval in seconds to write checkpoints during batch processing. Checkpoints tell Logstash where to resume processing after a restart. Checkpoints are automatically written at the end of each batch, regardless of this setting.

Writing checkpoints too frequently can slow down processing unnecessarily.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"]
   checkpoint_interval => 5
}

Consumer group used to read the Event Hub(s). Create a consumer group specifically for Logstash. Then ensure that all instances of Logstash use that consumer group so that they can work together properly.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"]
   consumer_group => "logstash"
}

Adds metadata about the Event Hub, including Event Hub name, consumer_group, processor_host, partition, offset, sequence, timestamp, and event_size.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"]
   decorate_events => true
}

When first reading from an Event Hub, start from this position:

Note: If storage_connection is set, the initial_position value is used only the first time Logstash reads from the Event Hub.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"]
   initial_position => "beginning"
}

Number of seconds to look back to find the initial position for pre-existing events. This option is used only if initial_position is set to look_back. If storage_connection is set, this configuration applies only the first time Logstash reads from the Event Hub.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"]
   initial_position => "look_back"
   initial_position_look_back => 86400
}

Maximum number of events retrieved and processed together. A checkpoint is created after each batch. Increasing this value may help with performance, but requires more memory.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"]
   max_batch_size => 125
}

Connection string for blob account storage. Blob account storage persists the offsets between restarts, and ensures that multiple instances of Logstash process different partitions. When this value is set, restarts resume where processing left off. When this value is not set, the initial_position value is used on every restart.

We strongly recommend that you define this value for production environments.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"]
   storage_connection => "DefaultEndpointsProtocol=https;AccountName=example...."
}

Name of the storage container used to persist offsets and allow multiple instances of Logstash to work together.

azure_event_hubs {
   event_hub_connections => ["Endpoint=sb://example1...;EntityPath=event_hub_name1"]
   storage_connection => "DefaultEndpointsProtocol=https;AccountName=example...."
   storage_container => "my_container"
}

To avoid overwriting offsets, you can use different storage containers. This is particularly important if you are monitoring two Event Hubs with the same name. You can use the advanced configuration model to configure different storage containers.

azure_event_hubs {
     config_mode => "advanced"
     consumer_group => "logstash"
     storage_connection => "DefaultEndpointsProtocol=https;AccountName=example...."
     event_hubs => [
        {"insights-operational-logs" => {
         event_hub_connection => "Endpoint=sb://example1..."
         storage_container => "insights-operational-logs-1"
        }},
        {"insights-operational-logs" => {
         event_hub_connection => "Endpoint=sb://example2..."
         storage_container => "insights-operational-logs-2"
        }}
     ]
   }

Total number of threads used to process events. The value you set here applies to all Event Hubs. Even with advanced configuration, this value is a global setting, and can’t be set per event hub.

azure_event_hubs {
   threads => 16
}

The number of threads should be the number of Event Hubs plus one or more. See Best practices for more information.

Add a field to an event

The codec used for input data. Input codecs are a convenient method for decoding your data before it enters the input, without needing a separate filter in your Logstash pipeline.

Disable or enable metric logging for this specific plugin instance by default we record all the metrics we can, but you can disable metrics collection for a specific plugin.

Add a unique ID to the plugin configuration. If no ID is specified, Logstash will generate one. It is strongly recommended to set this ID in your configuration. This is particularly useful when you have two or more plugins of the same type, for example, if you have 2 azure_event_hubs inputs. Adding a named ID in this case will help in monitoring Logstash when using the monitoring APIs.

input {
  azure_event_hubs {
    id => "my_plugin_id"
  }
}

Add any number of arbitrary tags to your event.

This can help with processing later.

Add a type field to all events handled by this input.

Types are used mainly for filter activation.

The type is stored as part of the event itself, so you can also use the type to search for it in Kibana.

If you try to set a type on an event that already has one (for example when you send an event from a shipper to an indexer) then a new input will not override the existing type. A type set at the shipper stays with that event for its life even when sent to another Logstash server.