Writing a new ManageIQ OpenStack Event Monitor

This article focuses on writing an Event Monitor class in ManageIQ project for a new arbitrary event service.

Let's imagine a situation that none of supported event-providing services (AMQP, Ceilometer, Panko) is suitable to given use case and a new one needs to be implemented. In this article, it will be called http_monitor which should fit to most of HTTP-based event providing services.

Where to start

Source code responsible for event monitoring feature is located at https://github.com/ManageIQ/manageiq-providers-openstack/tree/master/lib/manageiq/providers/openstack/legacy/events

There are several things which need to be done:

  1. Write new event monitor class (for capturing events from a HTTP service to ManageIQ)
  2. Prepare Conversion class for captured events to modify it to format expected by ManageIQ internals
  3. Update OpenstackEvent monitor to use the new event monitor class

All of this lives in ManageIQ project, so loading, process spawning, etc. is already solved.

New OpenStackEventMonitor class

There are existing event monitor classes already (files ending “_monitor.rb”) which can be used for better understanding how things work. For simplicity, assume following code as a minimal skeleton of a new monitor.

# file: openstack_http_event_monitor.rb
require 'manageiq/providers/openstack/legacy/openstack_event_monitor'
require 'manageiq/providers/openstack/legacy/events/openstack_event'

class OpenstackHTTPEventMonitor < OpenstackEventMonitor

  def self.available?(options = {})
    options[:ems].connect(:service => "Event")
    return true
  rescue => ex
    $log.debug("OpenstackHTTPEventMonitor availability check failed with #{ex}.") if $log

  def self.plugin_priority

  def initialize(options = {})
    @options = options
    @ems = options[:ems]

  def start
    @since          = nil
    @monitor_events = true

  def stop
    @monitor_events = false

  def each_batch
    while @monitor_events
      $log.info("Querying HTTP for events newer than #{latest_event_timestamp}...") if $log
      events = list_events(query_options).sort_by(&:generated)

      @since = events.last.generated unless events.empty?

      converted_events = events.map do |event|
        converted_event = OpenstackCeilometerHTTPConverter.new(event)
        $log.debug("Processing a new OpenStack event: #{event.inspect}") if $log
        openstack_event(nil, converted_event.metadata, converted_event.payload)

      yield converted_events

  def each
    each_batch do |events|
      events.each { |e| yield e }


  def query_options
      'time_from' => latest_event_timestamp || ''
      # time-based queries can be changed e.g. to order by Event ID (if it is numeric)..
      # add result size limit, etc.

  def list_events(query_options)
    # Get events from remote service API, e.g. using Excon and return an Array of hashes or objects
    # Excon.get(:query => query_options) or an API call..

  def latest_event_timestamp
    return @since if @since.present?

    @since = @ems.ems_events.maximum(:timestamp) || @ems.created_on.iso8601 || nil

Event format conversion

The Event payload expects following attributes: message_id, event_type, timestamp and payload. If the format of messages is different to this, a converter needs to be implemented. An example of easy one is openstack_ceilometer_event_converter.rb.

ManageIQ code takes care of taking the Event through Automate, Alerts, etc. until it ends up in event_streams database table. If the event monitor should trigger some new functionality in ManageIQ, the Automate part is the place where the action can be defined.

Adding the new Event Monitor

A class responsible for selecting a service which will be used for the Event Monitor process is https://github.com/ManageIQ/manageiq-providers-openstack/blob/master/lib/manageiq/providers/openstack/legacy/openstack_event_monitor.rb#L76-L99

Lifecycle of the event monitor process consists of few phases:

  1. Test if chosen event monitor is available by calling available? class method (Note, this is a class method, so all credentials, endpoints, etc. must be passed as arguments)
  2. Start the Event monitor (create the class instance and call start method)
  3. Periodically call each method to get new events and process it (using yield)
  4. Stop by calling stop method and terminate the process

Selection of appropriate Event monitor is defined on EMS where options came from.

def self.event_monitor_selected_class(options)
  case options[:events_monitor]
  when :ceilometer
  when :amqp
  # Added following:
  when :http

Options above comes from provider (an ExtManagementSystem subclasses like ManageIQ::Providers::Openstack::CloudManager) where its add/edit form contains input for Event service (currently there are available AMQP and Ceilometer/Panko).

The new event monitor might need update UI to get input from user like service endpoint or credentials. All such data are connected to the provider.


Event monitor can be tested either end-to-end in interaction with the event service with e.g. VCR recording to allow run tests offline. Or at least there should be few unit tests ensuring what queries are sent to the event service and how the returned data are processed. An example, even not perfect, can be found at openstack_ceilometer_event_monitor_spec.rb.


The article went through main pieces of implementation which needs to be done in order to add new Event monitor service to ManageIQ tool. An existing code in https://github.com/ManageIQ/manageiq-providers-openstack can provide more examples and inspiration.

#manageiq #openstack