src/journal/journald-rate-limit.c

   1 /* SPDX-License-Identifier: LGPL-2.1+ */
   2 /***
   3   Copyright 2011 Lennart Poettering
   4 ***/
   5
   6 #include <errno.h>
   7 #include <string.h>
   8
   9 #include "alloc-util.h"
  10 #include "hashmap.h"
  11 #include "journald-rate-limit.h"
  12 #include "list.h"
  13 #include "random-util.h"
  14 #include "string-util.h"
  15 #include "util.h"
  16
  17 #define POOLS_MAX 5
  18 #define BUCKETS_MAX 127
  19 #define GROUPS_MAX 2047
  20
  21 static const int priority_map[] = {
  22         [LOG_EMERG]   = 0,
  23         [LOG_ALERT]   = 0,
  24         [LOG_CRIT]    = 0,
  25         [LOG_ERR]     = 1,
  26         [LOG_WARNING] = 2,
  27         [LOG_NOTICE]  = 3,
  28         [LOG_INFO]    = 3,
  29         [LOG_DEBUG]   = 4
  30 };
  31
  32 typedef struct JournalRateLimitPool JournalRateLimitPool;
  33 typedef struct JournalRateLimitGroup JournalRateLimitGroup;
  34
  35 struct JournalRateLimitPool {
  36         usec_t begin;
  37         unsigned num;
  38         unsigned suppressed;
  39 };
  40
  41 struct JournalRateLimitGroup {
  42         JournalRateLimit *parent;
  43
  44         char *id;
  45         JournalRateLimitPool pools[POOLS_MAX];
  46         uint64_t hash;
  47
  48         LIST_FIELDS(JournalRateLimitGroup, bucket);
  49         LIST_FIELDS(JournalRateLimitGroup, lru);
  50 };
  51
  52 struct JournalRateLimit {
  53         usec_t interval;
  54         unsigned burst;
  55
  56         JournalRateLimitGroup* buckets[BUCKETS_MAX];
  57         JournalRateLimitGroup *lru, *lru_tail;
  58
  59         unsigned n_groups;
  60
  61         uint8_t hash_key[16];
  62 };
  63
  64 JournalRateLimit *journal_rate_limit_new(usec_t interval, unsigned burst) {
  65         JournalRateLimit *r;
  66
  67         assert(interval > 0 || burst == 0);
  68
  69         r = new0(JournalRateLimit, 1);
  70         if (!r)
  71                 return NULL;
  72
  73         r->interval = interval;
  74         r->burst = burst;
  75
  76         random_bytes(r->hash_key, sizeof(r->hash_key));
  77
  78         return r;
  79 }
  80
  81 static void journal_rate_limit_group_free(JournalRateLimitGroup *g) {
  82         assert(g);
  83
  84         if (g->parent) {
  85                 assert(g->parent->n_groups > 0);
  86
  87                 if (g->parent->lru_tail == g)
  88                         g->parent->lru_tail = g->lru_prev;
  89
  90                 LIST_REMOVE(lru, g->parent->lru, g);
  91                 LIST_REMOVE(bucket, g->parent->buckets[g->hash % BUCKETS_MAX], g);
  92
  93                 g->parent->n_groups--;
  94         }
  95
  96         free(g->id);
  97         free(g);
  98 }
  99
 100 void journal_rate_limit_free(JournalRateLimit *r) {
 101         assert(r);
 102
 103         while (r->lru)
 104                 journal_rate_limit_group_free(r->lru);
 105
 106         free(r);
 107 }
 108
 109 _pure_ static bool journal_rate_limit_group_expired(JournalRateLimitGroup *g, usec_t ts) {
 110         unsigned i;
 111
 112         assert(g);
 113
 114         for (i = 0; i < POOLS_MAX; i++)
 115                 if (g->pools[i].begin + g->parent->interval >= ts)
 116                         return false;
 117
 118         return true;
 119 }
 120
 121 static void journal_rate_limit_vacuum(JournalRateLimit *r, usec_t ts) {
 122         assert(r);
 123
 124         /* Makes room for at least one new item, but drop all
 125          * expored items too. */
 126
 127         while (r->n_groups >= GROUPS_MAX ||
 128                (r->lru_tail && journal_rate_limit_group_expired(r->lru_tail, ts)))
 129                 journal_rate_limit_group_free(r->lru_tail);
 130 }
 131
 132 static JournalRateLimitGroup* journal_rate_limit_group_new(JournalRateLimit *r, const char *id, usec_t ts) {
 133         JournalRateLimitGroup *g;
 134         struct siphash state;
 135
 136         assert(r);
 137         assert(id);
 138
 139         g = new0(JournalRateLimitGroup, 1);
 140         if (!g)
 141                 return NULL;
 142
 143         g->id = strdup(id);
 144         if (!g->id)
 145                 goto fail;
 146
 147         siphash24_init(&state, r->hash_key);
 148         string_hash_func(g->id, &state);
 149         g->hash = siphash24_finalize(&state);
 150
 151         journal_rate_limit_vacuum(r, ts);
 152
 153         LIST_PREPEND(bucket, r->buckets[g->hash % BUCKETS_MAX], g);
 154         LIST_PREPEND(lru, r->lru, g);
 155         if (!g->lru_next)
 156                 r->lru_tail = g;
 157         r->n_groups++;
 158
 159         g->parent = r;
 160         return g;
 161
 162 fail:
 163         journal_rate_limit_group_free(g);
 164         return NULL;
 165 }
 166
 167 static unsigned burst_modulate(unsigned burst, uint64_t available) {
 168         unsigned k;
 169
 170         /* Modulates the burst rate a bit with the amount of available
 171          * disk space */
 172
 173         k = u64log2(available);
 174
 175         /* 1MB */
 176         if (k <= 20)
 177                 return burst;
 178
 179         burst = (burst * (k-16)) / 4;
 180
 181         /*
 182          * Example:
 183          *
 184          *      <= 1MB = rate * 1
 185          *        16MB = rate * 2
 186          *       256MB = rate * 3
 187          *         4GB = rate * 4
 188          *        64GB = rate * 5
 189          *         1TB = rate * 6
 190          */
 191
 192         return burst;
 193 }
 194
 195 int journal_rate_limit_test(JournalRateLimit *r, const char *id, int priority, uint64_t available) {
 196         uint64_t h;
 197         JournalRateLimitGroup *g;
 198         JournalRateLimitPool *p;
 199         struct siphash state;
 200         unsigned burst;
 201         usec_t ts;
 202
 203         assert(id);
 204
 205         /* Returns:
 206          *
 207          * 0     → the log message shall be suppressed,
 208          * 1 + n → the log message shall be permitted, and n messages were dropped from the peer before
 209          * < 0   → error
 210          */
 211
 212         if (!r)
 213                 return 1;
 214
 215         if (r->interval == 0 || r->burst == 0)
 216                 return 1;
 217
 218         burst = burst_modulate(r->burst, available);
 219
 220         ts = now(CLOCK_MONOTONIC);
 221
 222         siphash24_init(&state, r->hash_key);
 223         string_hash_func(id, &state);
 224         h = siphash24_finalize(&state);
 225         g = r->buckets[h % BUCKETS_MAX];
 226
 227         LIST_FOREACH(bucket, g, g)
 228                 if (streq(g->id, id))
 229                         break;
 230
 231         if (!g) {
 232                 g = journal_rate_limit_group_new(r, id, ts);
 233                 if (!g)
 234                         return -ENOMEM;
 235         }
 236
 237         p = &g->pools[priority_map[priority]];
 238
 239         if (p->begin <= 0) {
 240                 p->suppressed = 0;
 241                 p->num = 1;
 242                 p->begin = ts;
 243                 return 1;
 244         }
 245
 246         if (p->begin + r->interval < ts) {
 247                 unsigned s;
 248
 249                 s = p->suppressed;
 250                 p->suppressed = 0;
 251                 p->num = 1;
 252                 p->begin = ts;
 253
 254                 return 1 + s;
 255         }
 256
 257         if (p->num < burst) {
 258                 p->num++;
 259                 return 1;
 260         }
 261
 262         p->suppressed++;
 263         return 0;
 264 }