Posted on by admin

Flood control

There are situations where the same recipient gets too many messages within a short time, when a spammer sends the same message to lots of recipients, and so on. To prevent these, EMG, as of version 8.0.7, provides a quite flexible mechanism. First you define a new section describing one or more policies. They should look as below, which should be put in the server.cfg file.

THROTTLE policyName <
KEYS=USERNAME,DESTADDR
LIMIT=3
TIMESPAN=15M
MODE=REJECT
>

The KEYS option contains the list of MGP options to base the decision on. You can use just USERNAME to limit what each user can send, MESSAGE to prevent identical messages from being sent over and over even if by different users or to different recipients, etc. You can also list multiple options separated by comma, as shown above. In this case the policy applies to messages sent to the same recipient by the same EMG user.

The LIMIT option says how many messages with the same values for the MGP options given above, that can be sent within the same timespan (see below). This value can be between 1 and 65536. For values up to 256, the limit is always respected exactly. For higher values we only store every nth timestamp, where n is as low as possible, but at least limit/256. This means that the timestamp comparison will not be exact, but it makes it possible to use much higher limits without using significantly more CPU, RAM, and disk resources. We also try to find an n that divides the limit as exactly as possible. This means all multiples of 256 will be exact, as well as multiples of lower powers of 2 for lower limits. There is of course also an error margin of n in these cases. Some examples are shown below.

  • LIMIT = 257 will store every timestamp, for an effective limit of 256.
  • LIMIT = 258 will store every 2nd timestamp, for an effective limit of 258.
  • LIMIT = 259 will store every 3rd timestamp, for an effective limit of 259.
  • LIMIT = 1000 will store every 4th timestamp, for an effective limit of 1000.
  • LIMIT = 1001 will store every 7th timestamp, for an effective limit of 1001.
  • LIMIT = 1009 will store every 4th timestamp, for an effective limit of 1008.

The TIMESPAN option contains the amount of time that must pass before sending message number “LIMIT + 1” with the same option values. The value is the number of seconds, and you can use M, H, D, and W as suffixes to set the value in minutes, hours, days, and weeks, respectively. The minimum value is 1 minute, and the maximum value is 1 week.

The MODE option is optional, and can be set to LOG, DELAY or REJECT. If not set, the mode depends on how the policy is used, as described below. LOG can be used when adding a new policy, to verify that it is triggered as expected.

Next, you activate this policy for incoming traffic by setting the global option THROTTLE_IN=policyName, once for every policy you want to be followed for all messages. You can also set the same option on specific connectors, again once or multiple times, to activate additional policies. Finally you can set the database column emguser.throttle to the name of a throttle policy. This column was added in database schema 47, in EMG version 8.0.7. For earlier database schemas, this column is ignored. The default mode for incoming traffic is REJECT.

You can control outgoing traffic by setting the global option THROTTLE_OUT, which also can be set on specific connectors. Both these options can be repeated. The default mode here is DELAY, causing the messages to simply be sent a bit later as necessary. This way all messages are delivered, but without violating the rules set by the downstreams operator.

When there are multiple policies used for a message, there are two rules:

  1. If any of the policies reject the message, the message is rejected immediately, and any remaining policies are ignored.
  2. If multiple policies want the message to be delayed, the longest delay is used.

The timestamps for all affected messages are stored in policy specific directories named using the format $EMGDIR/spool/leveldb/throttle-policyName. This way the throttle policies survive application restarts. If you remove a policy or want to reset it, simply remove the corresponding directory when emgd is not running.

All options can be updated at runtime, and are activated by running “emgd -reload” as usual. For limits above 256 some precision is lost, but EMG tries hard to retain as much information it can about timestamps for the previous messages.

Logging

When a message is affected by a throttle policy, the result is logged in a new log file named on the format log/throttle.policyName. Each line will contain a timestamp, a thread id, one of the strings “Delaying message”, “Rejecting message”, and “Triggered by”, plus some parameters from the message. The thread id can be used to find the connector instance, by looking for it in the pdu.* log files.

At startup and when the configuration has been reloaded, the current values of the parameters are logged. For policies without their own MODE value, the mode is logged as “DEALERS CHOICE“.

Please note that the format of these files may change in the future.

Mechanism

The central mechanism here is the same one as used for the throughput management. It is described below for those who want to understand it in more detail.

  • For a limit L we create an array of timestamps with L entries.
  • For every processed message, we append the current time to this array. After reaching the end of the array, the next entry will be written to the first position again. With this circular construction we always have the most L recent timestamps in the array.
  • If the previous timestamp at this position in the array was closer in time than the configured timespan, the new message is either logged, delayed, or rejected, according to the configuration.

Afterwards, the array is persisted to disk. This way it can be read back after a restart as needed. When enough time has passed to make the entire array irrelevant, the entry is removed both from memory and disk.

As we need to persist the entire array each time, its size is limited to 256 entries. For larger limits, we use an additional mechanism. Let’s say the given limit is 1024.

  • For the first message, we store its timestamp in the array described above and set a counter to 1.
  • For the next message, we ignore its timestamp and just increase the counter.
  • When the counter reaches 4 (calculated as 1024/256), we reset the counter to 0. This makes this secondary sequence start over from the beginning.

In order to maintain some sort of precision for the timestamps, this counter is also limited to 256.