KumoMTA event webhook (with log_hook helper)

This documents describes how to setup integration with KumoMTA and Postmastery Console using the log_hooks.lua helper.

This configuration uses batching, which requires version 2024.11.08-d383b033 or higher.

First make sure that the log_hooks helper is loaded in init.lua:

local log_hooks = require 'policy-extras.log_hooks'

Configure a new log_hook to deliver event data to Postmastery over HTTP. If your Postmastery dataset is stored in US use app.postmastery.com, if your data is stored in EU use app.postmastery.eu in the url below. Replace MY_DATASET_ID and MY_ACCESS_KEY with the values provided by Postmastery:

-- Configure Postmastery log hook
log_hooks:new {
  name = 'postmastery-webhook',
  batch_size = 500,
  min_batch_size = 100,
  max_batch_latency = "60s",

  -- log_parameters are combined with the name and
  -- passed through to kumo.configure_log_hook
  log_parameters = {
    -- Metadata added in smtp_server_message_received hook
    meta = { 'from','subject','message_id' },
    
    per_record = {
      -- SMTP and HTTP Incoming Monitoring, use for debugging only
      Reception = { enable = false },
      -- Rejected Incoming SMTP Command, use for debugging only
      Rejection = { enable = false },
      -- Internal Delayed messages, use for debugging only
      Delayed = { enable = false },
      Any = { enable = true },
    },
  },

  -- The constructor is called when kumod needs to initiate
  -- a new connection to the log target. It must return
  -- a connection object
  constructor = function(domain, tenant, campaign)
    -- Define the connection object
    local connection = {}

    -- Create an HTTP client
    local client = kumo.http.build_client {}

    -- The send method is called for each log event
    -- This method must be named send_batch when batch_size > 1
    function connection:send_batch(messages)
      local payload = {}
      for _, msg in ipairs(messages) do
        -- Rather than collecting the pre-templated record as
        -- a string, get it as an object.  This makes it easier
        -- to compose it as an array and json encode than doing
        -- the string manipulation by-hand.
        local record = msg:get_meta 'log_record'
        -- Uncomment the following code to mask the local-part of the recipient
        -- local parts = kumo.string.split(record.recipient, '@')
        -- record.response.content = record.response.content:gsub(parts[1], 'xxxx')
        -- record.recipient = 'xxxx@' .. parts[2]
        table.insert(payload, record)
      end

      -- Encode the array of objects as json
      local data = kumo.serde.json_encode(payload)

      local response = client
        :post('https://app.postmastery.eu/v1/kumomta/MY_DATASET_ID')
        :header('Authorization', 'MY_ACCESS_KEY')
        :header('Content-Type', 'application/json')
        :body(data)
        :send()

      local disposition = string.format(
        '%d %s: %s',
        response:status_code(),
        response:status_reason(),
        response:text()
      )

      if response:status_is_success() then
        return disposition
      end
      
      -- Retry in case of server error (HTTP status 500-599)
      if response:status_is_server_error() then
        kumo.reject(400, disposition)
      end
      -- Bounce in case of other error
      kumo.reject(500, disposition)
    end

    -- The close method is called when the connection needs
    -- to be closed
    function connection:close()
      client:close()
    end

    return connection
  end,
}

Optimising webhook delivery

To ensure that the max_batch_latency value will be used, the KumoMTA idle_timeout value from the “default” shaping rule must be equal or higher:

["default"]
...
idle_timeout = 60s
...

For webhook delivery efficiency is preferred over low latency. To optimize batching and reduce overhead we recommend using a traffic shaping rule for the webhook:

["postmastery-webhook.log_hook"]
mx_rollup = false
connection_limit = 1
max_deliveries_per_connection = 100000
max_connection_rate = "1000/s"

The connection_limit is set to 1 to prevent batches from being split across multiple connections when traffic is low. A higher connection_limit may be needed on high traffic systems to increase throughput. Increase it when data is queued for more than, say 5 minutes.

Adding headers as metadata

The From and Subject headers are used for aggregation in Delivery Analytics reports. For performance reasons it is recommended to import headers as metadata instead of retrieving headers in the logger:

kumo.on('smtp_server_message_received', function(msg)
  ...
  msg:import_x_headers { 'from','subject','message-id' }
  ...
end)

Custom headers can be also used for aggregation and shown in Delivery Analytics as report tables:

msg:import_x_headers { 'from','subject', 'x-template-id' }

When importing headers the name is converted to lowercase and ‘-' is converted to '_'. So header X-Template-ID is stored as x_template_id in meta.

Note that campaign is a predefined metadata value and will be used to construct the queue name. The header for the campaign can be set using the queues helper. It will be automatically added as meta variable.

Make sure to retrieve additional imported headers as meta in log_parameters:

  log_parameters = {
    ...
    meta = { 'from', 'subject', 'campaign', ... },