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 extracted from headers in smtp_server_message_received
    meta = { 'from', 'subject' },
    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[0], 'xxxx')
        -- record.recipient = 'xxxx@' + parts[1]
        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,
}

Optimizing Default Shaping and ddding dedicated Traffic Shaping for webhook delivery

To ensure that the max_batch_latency value will be used, it has to be equal or below the global KumoMTA idle_timeout value from the “default” shaping :

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

To optimize webhook batching and delivery, KumoMTA recommends using a dedicated traffic shaping rule such as :

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

The idea is to maximise batch size. The higher the connection_limit value, the smaller the batches will be. KumoMTA splits batches to use as many connections as possible to speed up event delivery.

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' }
  ...
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', ... },