The reported behavior was that messages seemd to be retried much faster
than the retry schedule, and at a quick glance it looked like the
nxdomain code path didn't respest the backoff, but from hooking
up this test and making the durations longer, it really doesn't seem
like the issue was that simple:
refs: https://github.com/KumoCorp/kumomta/issues/271
This really is adding batching support to custom lua delivery
protocol handlers, but the main use case for these today is
to implement log hooks.
The way that it works is that you can specify a `batch_size`
as part of setting up the lua protocol handler.
Then, when it is time to send messages, if the batch_size is
the default of 1, the lua delivery logic will invoke the `send` method
on the connection object returned from the constructor. This
is the same as the behavior from before this commit.
However, if the batch_size is greater than 1, then the lua delivery
logic will instead attempt to collect up to batch_size messages
that are immediately available from the ready queue, and then pass
those to a new `send_batch` method.
The send_batch method accepts an array of messages; that array will
always have at least one message, and up to batch_size messages,
depending on the throughput and queue size.
If the send_batch method's return value applies equally to all
messages in the batch, so if it indicates that something failed,
that disposition will apply to all messages.
One of the reasons that I'd avoided implementing batching thus far
was that it makes it awkward to resolve persistent/recurring issues
that are due to a single message in that batch. If the batch is
always retried together then there is a good chance that it will
always fail together.
There's no explicit mitigation for that issue here, but it may
be probablistically mitigated by the jitter that is applied to
messages that transiently fail. If a batch transiently fails,
each message in that batch will be subject to its own random
jitter which should cause an offending message to be retried
with a different subset of messages next time around.
The integration test included here demonstrates the batching
working with an http log hook implementation.
Previously we were using only our timeq module, which is built on top of
hashed hierarchical timer wheels.
Timer wheels have O(1) insertion and removal which are excellent
properties for larger delayed queues.
However, they do not know how to answer the question "when is the
next item due", but only "what is due in the next tick".
The underlying timer wheel implementation assumes a 1ms granularity
which is a little bit at-odds with our pragmatic view of the scheduled
queue, which is "if it's in there, precision timing isn't important, and
it's generally fine to consider once per minute", requiring that we
either aggressively scheduled a maintainer task to wake up every 1ms per
scheduled queue (untenable!) or have it wake up no more than once per
second but ideally closer to once per minute to then tick however
milliseconds are necessary to advance the wheel to the current slot.
For small numbers of scheduled queues with sufficiently large retry
intervals this hasn't bubbled up as an issue so far, but it bothers
me that it isn't as efficient as it could be because we have to wake up
reasonably frequently to keep things ticking over, and that introduces
higher continual CPU utilization. It's small, but I worry about
the aggregate cost spread over very large numbers of scheduled queues.
What I really want here is a a timer wheel that I can tick with
arbitrary granularity and with that in mind I took a look at adapting
the handful of existing implementations and found that we're already
using the cleanest implementation, and it would take some effort that I
didn't really want to spend right now.
I opted for a reasonably simple alternative option, which is to adopt a
skiplist for the queue. This has O(log n) insertion to maintain ordering
with O(1) removal and can answer "when is the next item due". What this
means is that we pay a slightly higher insertion cost one-time in
exchange for being able to put the maintainer for the queue asleep until
we need it, and not have to keep waking up between times, which should
scale better.
What this means in practice is that we now wake up the maintainer either
when the next message is due, or once per minute to re-evaluate the
queue configuration hook, so we're slightly better off, but totally
where I'd like to be.
I've introduced a reap_interval (default 10 minutes) and a
refresh_interval (default 1 minute) as parameters in get_queue_config so
that you can increase that 1m interval for reloading.
What I'd like to do in a follow up commit is introduce a way to define
the refresh policy. For example, it would be neat to say "watch my
policy directory and refresh when it changes", which would make things
the most efficient for many users. For those that are loading their
config from a remote datasource, we'd need to consider some other
mechanism for this; maybe some kind of long-poll or pubsub, but will
obviously still be able to support the current interval based polling.
Now, with all of that said: I didn't want to switch the product
default over and hope for best, so what I did was add a strategy
option to allow this to be adopted on a per-queue basis.
Since I was in here adding some options, I also added an option
that allows explicitly setting the interval used for timerwheel
ticks, so you now have a lot more opportunities for tuning this
stuff.
We've been trying to run down a weird problem with the lapin
client where the system appears to get bogged down around lapin
and the submission rate goes through the floor. There no obvious
signs of a problem elsewhere in the system or in the logging,
so we're trying out an optional alternative client implementation.
This commit adds a "one-shot" publish function:
```
log_hooks:new {
name = 'amqp',
constructor = function(domain, tenant, campaign)
local sender = {}
local host, port = table.unpack(kumo.string.split(AMQP_HOST_PORT, ':'))
function sender:send(msg)
kumo.amqp.basic_publish {
routing_key = 'woot',
payload = msg:get_data(),
connection = {
host = host,
port = tonumber(port),
},
}
return '250 ok'
end
return sender
end,
}
```
behind the scenes we maintain a connection+channel pool for this.
This first pass doesn't support confirms, to the publish is
fire-and-forget; we assume that if there was no error while trying to
acquire the channel or returned immediately by the underlying
basic_publish function, that the server took the message and we're
done.
I was going to use mobc for the connection pooling here, since I just
implemented it for the recent redis client changes, but it triggered a
non-sensical panic around mobc's internal use of the metrics crate.
Upon further inspection, mobc appears to clobber certain metrics when
multiple Pools are in used, which makes it unsuitable for real world
use. That was the cause of the panic; I couldn't figure out the cause of
the panic. I switched to using the deadpool connection pooling crate
which has a similar API and doesn't panic. I will replace the use of
mobc in mod-redis with deadpool in a later commit.
I've introduced a place to stash the "global" tokio runtime that
is configured by `main` in `kumod` and `tsa-daemon`. That allows
us to bounce onto that runtime, rather than spawning an ad-hoc
single threaded runtime.
Since we're still preserving the existing client, this commit
introduces a new and slightly different integration test to
verify that publishing is operational from here.
Docs still need to be written up for this, but I want to put
this through its paces before doing that.
I had kumomta-dev installed, so requiring the helpers succeeded.
It failed in CI because we need to set the path to find the local
assets dir like we do in TSA.
So do that!
The back story is that we've been trying to run down a situation
where the AMQP connections seemd to get busy, running close to 100%,
and that appears to impact some other processing.
This commit dedicates a thread per AMQP connection to manage its
connection state. This is likely not totally necessary, but it
is desirable to move it off of the current tokio context, which
is one of our lua localset threads, in order to reduce that
contention.
I don't really like this as a solution, it's just giving us
an option to play with while we zero-in on the underlying issue.
As part of this, we now have an explicit timeout around the connect
operation.
This commit also adds a publish_with_timeout method on the amqp
connection, which combines the publish and the wait for confirmation
and puts a timeout around that operation.
I'm expecting CI to not like this, as we're using the testcontainers
crate which spawns docker images to set up isolated server instances.
This is not directly compatible with our current CI environment.
So these tests are only enabled when KUMOD_TESTCONTAINERS=1 is
set in the environment.
The test sets up a rabbit instance, declares a queue, sets up simple log
hook in kumo, injects a message and then waits for it to process
through, then checks to receive a reception and delivery record.
To facilitate this, some adjustments needed to be made to the time
calculation in the maintainer because we previously didn't consider it
to be valid to have a retry_interval below 1 minute, but in order for
this integration test to be viable to run as part of the CI it needs to
run in significantly less time than 1 minute.
The approach taken here is to avoid considering 1 minute as the
baseline, but rather take 1/20th of the retry_interval. In the default
configuration, the numbers work out the same as previously, but they
will scale down as the retry_interval is reduced.
Care is taken to avoid a couple of borderline busy wait scenarios where
we might otherwise have woken up at unrealistically small intervals:
timeq can suggest 1ms in a few scenarios, and we just round those up to
the next second to avoid that.
It's worth noting that we do not consider the scheduled queue to be a
realtime, high granularity queue: anything that lands there is
considered to be bulk/batch and will be handled later: it isn't worth
prioritizing with high granularity because messages that land there are
generally not going to be delivered quickly.
When `log_arf` or `log_oob` are set to true with `relay_to=false`, we
now return a 550 error response for messages that are not ARF or OOB
reports. Previously, we would return a 250 response and silently drop
the message in this case, which gave the false impression that it was
accepted for relaying.
Expand integration test to explicitly assert that the right things
are allowed/denied/relayed/parsed.
As best as I can tell, this is a casualty of a last moment
code format/copy-pasta. All of the logic works correctly,
but the Reception log record didn't include the relay
disposition so the log_oob or log_arf flag didn't make it
to the logging layer.
This code isn't currently reachable (defaults to false), but allows
for a runtime selection between rustls and openssl-based tls.
refs: https://github.com/KumoCorp/kumomta/issues/8
Previously, you would do either:
`msg:set_meta('queue', 'smart.host.domain')`
or
`msg:set_meta('queue', '[10.0.0.1]')`
to override the effective domain for a message and cause it to be routed
to somewhere other than the recipient domain.
That was OK for basic smart hosting, but limiting when you wanted to use
multiple candidate hosts.
This commit expands the queue config `protocol` field to support
specifying an explicit list of MX hosts that should be used instead.
The integration tests have been migrated away from the old style to this
new style.
While adding plumbing for this, I uncovered an inconsistency between the
queue name generated for the ready queue and the name used by suspension
handling. The inconsistency was introduced in
0842a0fc8b and related work. This commit
resolves it.
The name is passed through to should_enqueue_log_record as an additonal
parameter to make it possible to reason about whether a given record
should get queued for a specific log hook instance.
This is a breaking change, but it can be easily resolved by adding
the name parameter to the `configure_log_hook` call.
These are two different groups of queues, so there are two different
sets of things to manage them.
kcli now has `suspend(-list|cancel)?` and
`suspend-ready-q(-list|cancel)?` subcommands for establishing a
suspension, listing the suspensions and cancelling a suspension
in the scheduled-q and ready-q namespaces respectively.
The names of the ready queues can be derived from the metrics API:
```console
$ curl -s 'http://localhost:8000/metrics.json' | jq .
...
"ready_count": {
"help": "number of messages in the ready queue",
"type": "gauge",
"value": {
"service": {
"smtp_client:source2->(in1-smtp|in2-smtp).messagingengine.com": 0.0
}
}
},
...
```
From the above, `source2->(in1-smtp|in2-smtp).messagingengine.com` is
the name of the underlying ready queue.
We can and probably should add something to `kcli` to make that slightly
easier to review and manage for the operator.
refs: https://github.com/KumoCorp/kumomta/issues/51
This is relatively basic in that it can catch general lua lints,
but doesn't know about the set of functions available to kumo.
It doesn't appear as though selene is able to be extended to
know about those yet; that is tracked by
https://github.com/Kampfkarren/selene/issues/520
I've fixed the couple of lints in our policy and test files
as part of this commit.
You can install and run it like this:
```console
$ cargo install selene
```
Then:
```console
$ selene .
```
This has our first end-to-end integration test that validates
smtp -> source mta -> sink mta -> maildir
and confirms that the message has the right bits inside it.
The end to end test will get refactored into more easily usable
pieces in follow-up commit(s).
Wez Furlong