SymbX/systemd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #32758 from yuwata/journal-ratelimit

Browse Source 
journal: modernize journal ratelimiting
This commit is contained in:
Luca Boccassi
2024-05-13 15:16:40 +02:00
committed by GitHub
parent 2b214821a2 0e2e3fa35a
commit 74cf904aac
13 changed files with 177 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic/log.c
+4 -4
View File
@@ -396,7 +396,7 @@ void log_forget_fds(void) {
}
void log_set_max_level(int level) {
assert(level == LOG_NULL || (level & LOG_PRIMASK) == level);
assert(level == LOG_NULL || LOG_PRI(level) == level);
log_max_level = level;
@@ -788,7 +788,7 @@ int log_dispatch_internal(
return -ERRNO_VALUE(error);
/* Patch in LOG_DAEMON facility if necessary */
if ((level & LOG_FACMASK) == 0)
if (LOG_FAC(level) == 0)
level |= log_facility;
if (open_when_needed)
@@ -1077,7 +1077,7 @@ int log_struct_internal(
log_target == LOG_TARGET_NULL)
return -ERRNO_VALUE(error);
if ((level & LOG_FACMASK) == 0)
if (LOG_FAC(level) == 0)
level |= log_facility;
if (IN_SET(log_target,
@@ -1180,7 +1180,7 @@ int log_struct_iovec_internal(
log_target == LOG_TARGET_NULL)
return -ERRNO_VALUE(error);
if ((level & LOG_FACMASK) == 0)
if (LOG_FAC(level) == 0)
level |= log_facility;
if (IN_SET(log_target, LOG_TARGET_AUTO,
src/journal/journald-kmsg.c
+1 -1
View File
@@ -260,7 +260,7 @@ void dev_kmsg_record(Server *s, char *p, size_t l) {
iovec[n++] = IOVEC_MAKE_STRING("_TRANSPORT=kernel");
char syslog_priority[STRLEN("PRIORITY=") + DECIMAL_STR_MAX(int)];
xsprintf(syslog_priority, "PRIORITY=%i", priority & LOG_PRIMASK);
xsprintf(syslog_priority, "PRIORITY=%i", LOG_PRI(priority));
iovec[n++] = IOVEC_MAKE_STRING(syslog_priority);
char syslog_facility[STRLEN("SYSLOG_FACILITY=") + DECIMAL_STR_MAX(int)];
src/journal/journald-native.c
+2 -2
View File
@@ -53,13 +53,13 @@ static void server_process_entry_meta(
else if (l == 17 &&
startswith(p, "SYSLOG_FACILITY=") &&
p[16] >= '0' && p[16] <= '9')
*priority = (*priority & LOG_PRIMASK) | ((p[16] - '0') << 3);
*priority = LOG_PRI(*priority) | ((p[16] - '0') << 3);
else if (l == 18 &&
startswith(p, "SYSLOG_FACILITY=") &&
p[16] >= '0' && p[16] <= '9' &&
p[17] >= '0' && p[17] <= '9')
*priority = (*priority & LOG_PRIMASK) | (((p[16] - '0')*10 + (p[17] - '0')) << 3);
*priority = LOG_PRI(*priority) | (((p[16] - '0')*10 + (p[17] - '0')) << 3);
else if (l >= 19 &&
startswith(p, "SYSLOG_IDENTIFIER=")) {
src/journal/journald-rate-limit.c
+97 -116
View File
@@ -1,18 +1,13 @@
/* SPDX-License-Identifier: LGPL-2.1-or-later */
#include <errno.h>
#include "alloc-util.h"
#include "hashmap.h"
#include "journald-rate-limit.h"
#include "list.h"
#include "logarithm.h"
#include "random-util.h"
#include "string-util.h"
#include "time-util.h"
#define POOLS_MAX 5
#define BUCKETS_MAX 127
#define GROUPS_MAX 2047
static const int priority_map[] = {
@@ -23,20 +18,17 @@ static const int priority_map[] = {
[LOG_WARNING] = 2,
[LOG_NOTICE] = 3,
[LOG_INFO] = 3,
[LOG_DEBUG] = 4
[LOG_DEBUG] = 4,
};
typedef struct JournalRateLimitPool JournalRateLimitPool;
typedef struct JournalRateLimitGroup JournalRateLimitGroup;
struct JournalRateLimitPool {
typedef struct JournalRateLimitPool {
usec_t begin;
unsigned num;
unsigned suppressed;
};
} JournalRateLimitPool;
struct JournalRateLimitGroup {
JournalRateLimit *parent;
typedef struct JournalRateLimitGroup {
OrderedHashmap *groups_by_id;
char *id;
@@ -44,116 +36,112 @@ struct JournalRateLimitGroup {
usec_t interval;
JournalRateLimitPool pools[POOLS_MAX];
uint64_t hash;
} JournalRateLimitGroup;
LIST_FIELDS(JournalRateLimitGroup, bucket);
LIST_FIELDS(JournalRateLimitGroup, lru);
};
struct JournalRateLimit {
JournalRateLimitGroup* buckets[BUCKETS_MAX];
JournalRateLimitGroup *lru, *lru_tail;
unsigned n_groups;
uint8_t hash_key[16];
};
JournalRateLimit *journal_ratelimit_new(void) {
JournalRateLimit *r;
r = new0(JournalRateLimit, 1);
if (!r)
static JournalRateLimitGroup* journal_ratelimit_group_free(JournalRateLimitGroup *g) {
if (!g)
return NULL;
random_bytes(r->hash_key, sizeof(r->hash_key));
return r;
}
static void journal_ratelimit_group_free(JournalRateLimitGroup *g) {
assert(g);
if (g->parent) {
assert(g->parent->n_groups > 0);
if (g->parent->lru_tail == g)
g->parent->lru_tail = g->lru_prev;
LIST_REMOVE(lru, g->parent->lru, g);
LIST_REMOVE(bucket, g->parent->buckets[g->hash % BUCKETS_MAX], g);
g->parent->n_groups--;
}
if (g->groups_by_id && g->id)
/* The group is already removed from the hashmap when this is called from the
* destructor of the hashmap. Hence, do not check the return value here. */
ordered_hashmap_remove_value(g->groups_by_id, g->id, g);
free(g->id);
free(g);
return mfree(g);
}
void journal_ratelimit_free(JournalRateLimit *r) {
assert(r);
DEFINE_TRIVIAL_CLEANUP_FUNC(JournalRateLimitGroup*, journal_ratelimit_group_free);
while (r->lru)
journal_ratelimit_group_free(r->lru);
free(r);
}
DEFINE_PRIVATE_HASH_OPS_WITH_VALUE_DESTRUCTOR(
journal_ratelimit_group_hash_ops,
char,
string_hash_func,
string_compare_func,
JournalRateLimitGroup,
journal_ratelimit_group_free);
static bool journal_ratelimit_group_expired(JournalRateLimitGroup *g, usec_t ts) {
unsigned i;
assert(g);
for (i = 0; i < POOLS_MAX; i++)
if (g->pools[i].begin + g->interval >= ts)
FOREACH_ELEMENT(p, g->pools)
if (usec_add(p->begin, g->interval) >= ts)
return false;
return true;
}
static void journal_ratelimit_vacuum(JournalRateLimit *r, usec_t ts) {
assert(r);
static void journal_ratelimit_vacuum(OrderedHashmap *groups_by_id, usec_t ts) {
/* Makes room for at least one new item, but drop all expired items too. */
while (r->n_groups >= GROUPS_MAX ||
(r->lru_tail && journal_ratelimit_group_expired(r->lru_tail, ts)))
journal_ratelimit_group_free(r->lru_tail);
while (ordered_hashmap_size(groups_by_id) >= GROUPS_MAX)
journal_ratelimit_group_free(ordered_hashmap_first(groups_by_id));
JournalRateLimitGroup *g;
while ((g = ordered_hashmap_first(groups_by_id)) && journal_ratelimit_group_expired(g, ts))
journal_ratelimit_group_free(g);
}
static JournalRateLimitGroup* journal_ratelimit_group_new(JournalRateLimit *r, const char *id, usec_t interval, usec_t ts) {
static int journal_ratelimit_group_new(
OrderedHashmap **groups_by_id,
const char *id,
usec_t interval,
usec_t ts,
JournalRateLimitGroup **ret) {
_cleanup_(journal_ratelimit_group_freep) JournalRateLimitGroup *g = NULL;
int r;
assert(groups_by_id);
assert(id);
assert(ret);
g = new(JournalRateLimitGroup, 1);
if (!g)
return -ENOMEM;
*g = (JournalRateLimitGroup) {
.id = strdup(id),
.interval = interval,
};
if (!g->id)
return -ENOMEM;
journal_ratelimit_vacuum(*groups_by_id, ts);
r = ordered_hashmap_ensure_put(groups_by_id, &journal_ratelimit_group_hash_ops, g->id, g);
if (r < 0)
return r;
assert(r > 0);
g->groups_by_id = *groups_by_id;
*ret = TAKE_PTR(g);
return 0;
}
static int journal_ratelimit_group_acquire(
OrderedHashmap **groups_by_id,
const char *id,
usec_t interval,
usec_t ts,
JournalRateLimitGroup **ret) {
JournalRateLimitGroup *g;
assert(r);
assert(groups_by_id);
assert(id);
assert(ret);
g = new0(JournalRateLimitGroup, 1);
g = ordered_hashmap_get(*groups_by_id, id);
if (!g)
return NULL;
g->id = strdup(id);
if (!g->id)
goto fail;
g->hash = siphash24_string(g->id, r->hash_key);
return journal_ratelimit_group_new(groups_by_id, id, interval, ts, ret);
g->interval = interval;
journal_ratelimit_vacuum(r, ts);
LIST_PREPEND(bucket, r->buckets[g->hash % BUCKETS_MAX], g);
LIST_PREPEND(lru, r->lru, g);
if (!g->lru_next)
r->lru_tail = g;
r->n_groups++;
g->parent = r;
return g;
fail:
journal_ratelimit_group_free(g);
return NULL;
*ret = g;
return 0;
}
static unsigned burst_modulate(unsigned burst, uint64_t available) {
@@ -184,13 +172,21 @@ static unsigned burst_modulate(unsigned burst, uint64_t available) {
return burst;
}
int journal_ratelimit_test(JournalRateLimit *r, const char *id, usec_t rl_interval, unsigned rl_burst, int priority, uint64_t available) {
JournalRateLimitGroup *g, *found = NULL;
int journal_ratelimit_test(
OrderedHashmap **groups_by_id,
const char *id,
usec_t rl_interval,
unsigned rl_burst,
int priority,
uint64_t available) {
JournalRateLimitGroup *g;
JournalRateLimitPool *p;
unsigned burst;
uint64_t h;
usec_t ts;
int r;
assert(groups_by_id);
assert(id);
/* Returns:
@@ -200,33 +196,18 @@ int journal_ratelimit_test(JournalRateLimit *r, const char *id, usec_t rl_interv
* < 0 → error
*/
if (!r)
return 1;
ts = now(CLOCK_MONOTONIC);
h = siphash24_string(id, r->hash_key);
g = r->buckets[h % BUCKETS_MAX];
LIST_FOREACH(bucket, i, g)
if (streq(i->id, id)) {
found = i;
break;
}
if (!found) {
found = journal_ratelimit_group_new(r, id, rl_interval, ts);
if (!found)
return -ENOMEM;
} else
found->interval = rl_interval;
r = journal_ratelimit_group_acquire(groups_by_id, id, rl_interval, ts, &g);
if (r < 0)
return r;
if (rl_interval == 0 || rl_burst == 0)
return 1;
burst = burst_modulate(rl_burst, available);
p = &found->pools[priority_map[priority]];
p = &g->pools[priority_map[priority]];
if (p->begin <= 0) {
p->suppressed = 0;
@@ -235,7 +216,7 @@ int journal_ratelimit_test(JournalRateLimit *r, const char *id, usec_t rl_interv
return 1;
}
if (p->begin + rl_interval < ts) {
if (usec_add(p->begin, rl_interval) < ts) {
unsigned s;
s = p->suppressed;
src/journal/journald-rate-limit.h
+10 -5
View File
@@ -1,10 +1,15 @@
/* SPDX-License-Identifier: LGPL-2.1-or-later */
#pragma once
#include <inttypes.h>
#include "hashmap.h"
#include "time-util.h"
typedef struct JournalRateLimit JournalRateLimit;
JournalRateLimit *journal_ratelimit_new(void);
void journal_ratelimit_free(JournalRateLimit *r);
int journal_ratelimit_test(JournalRateLimit *r, const char *id, usec_t rl_interval, unsigned rl_burst, int priority, uint64_t available);
int journal_ratelimit_test(
OrderedHashmap **groups_by_id,
const char *id,
usec_t rl_interval,
unsigned rl_burst,
int priority,
uint64_t available);
src/journal/journald-server.c
+8 -7
View File
@@ -1273,7 +1273,13 @@ void server_dispatch_message(
if (c && c->unit) {
(void) server_determine_space(s, &available, /* limit= */ NULL);
rl = journal_ratelimit_test(s->ratelimit, c->unit, c->log_ratelimit_interval, c->log_ratelimit_burst, priority & LOG_PRIMASK, available);
rl = journal_ratelimit_test(
&s->ratelimit_groups_by_id,
c->unit,
c->log_ratelimit_interval,
c->log_ratelimit_burst,
LOG_PRI(priority),
available);
if (rl == 0)
return;
@@ -2803,10 +2809,6 @@ int server_init(Server *s, const char *namespace) {
if (r < 0)
return r;
s->ratelimit = journal_ratelimit_new();
if (!s->ratelimit)
return log_oom();
r = cg_get_root_path(&s->cgroup_root);
if (r < 0)
return log_error_errno(r, "Failed to acquire cgroup root path: %m");
@@ -2907,8 +2909,7 @@ Server* server_free(Server *s) {
safe_close(s->notify_fd);
safe_close(s->forward_socket_fd);
if (s->ratelimit)
journal_ratelimit_free(s->ratelimit);
ordered_hashmap_free(s->ratelimit_groups_by_id);
server_unmap_seqnum_file(s->seqnum, sizeof(*s->seqnum));
server_unmap_seqnum_file(s->kernel_seqnum, sizeof(*s->kernel_seqnum));
src/journal/journald-server.h
+1 -2
View File
@@ -14,7 +14,6 @@ typedef struct Server Server;
#include "hashmap.h"
#include "journal-file.h"
#include "journald-context.h"
#include "journald-rate-limit.h"
#include "journald-stream.h"
#include "list.h"
#include "prioq.h"
@@ -108,7 +107,7 @@ struct Server {
char *buffer;
JournalRateLimit *ratelimit;
OrderedHashmap *ratelimit_groups_by_id;
usec_t sync_interval_usec;
usec_t ratelimit_interval;
unsigned ratelimit_burst;
src/journal/journald-stream.c
+1 -1
View File
@@ -310,7 +310,7 @@ static int stdout_stream_log(
syslog_priority[STRLEN("PRIORITY=")] = '0' + LOG_PRI(priority);
iovec[n++] = IOVEC_MAKE_STRING(syslog_priority);
if (priority & LOG_FACMASK) {
if (LOG_FAC(priority) != 0) {
xsprintf(syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority));
iovec[n++] = IOVEC_MAKE_STRING(syslog_facility);
}
src/journal/journald-syslog.c
+4 -4
View File
@@ -177,8 +177,8 @@ void server_forward_syslog(Server *s, int priority, const char *identifier, cons
int syslog_fixup_facility(int priority) {
if ((priority & LOG_FACMASK) == 0)
return (priority & LOG_PRIMASK) | LOG_USER;
if (LOG_FAC(priority) == 0)
return LOG_PRI(priority) | LOG_USER;
return priority;
}
@@ -403,10 +403,10 @@ void server_process_syslog_message(
iovec[n++] = IOVEC_MAKE_STRING("_TRANSPORT=syslog");
xsprintf(syslog_priority, "PRIORITY=%i", priority & LOG_PRIMASK);
xsprintf(syslog_priority, "PRIORITY=%i", LOG_PRI(priority));
iovec[n++] = IOVEC_MAKE_STRING(syslog_priority);
if (priority & LOG_FACMASK) {
if (LOG_FAC(priority) != 0) {
xsprintf(syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority));
iovec[n++] = IOVEC_MAKE_STRING(syslog_facility);
}
src/journal/meson.build
+6
View File
@@ -122,6 +122,12 @@ executables += [
libxz_cflags,
],
},
test_template + {
'sources' : files(
'test-journald-rate-limit.c',
'journald-rate-limit.c',
),
},
journal_test_template + {
'sources' : files('test-journald-syslog.c'),
'dependencies' : [
src/journal/test-journald-rate-limit.c
+40
View File
@@ -0,0 +1,40 @@
/* SPDX-License-Identifier: LGPL-2.1-or-later */
#include "journald-rate-limit.h"
#include "tests.h"
TEST(journal_ratelimit_test) {
_cleanup_ordered_hashmap_free_ OrderedHashmap *rl = NULL;
int r;
for (unsigned i = 0; i < 20; i++) {
r = journal_ratelimit_test(&rl, "hoge", USEC_PER_SEC, 10, LOG_DEBUG, 0);
assert_se(r == (i < 10 ? 1 : 0));
r = journal_ratelimit_test(&rl, "foo", 10 * USEC_PER_SEC, 10, LOG_DEBUG, 0);
assert_se(r == (i < 10 ? 1 : 0));
}
/* Different priority group with the same ID is not ratelimited. */
assert_se(journal_ratelimit_test(&rl, "hoge", USEC_PER_SEC, 10, LOG_INFO, 0) == 1);
assert_se(journal_ratelimit_test(&rl, "foo", 10 * USEC_PER_SEC, 10, LOG_INFO, 0) == 1);
/* Still LOG_DEBUG is ratelimited. */
assert_se(journal_ratelimit_test(&rl, "hoge", USEC_PER_SEC, 10, LOG_DEBUG, 0) == 0);
assert_se(journal_ratelimit_test(&rl, "foo", 10 * USEC_PER_SEC, 10, LOG_DEBUG, 0) == 0);
/* Different ID is not ratelimited. */
assert_se(journal_ratelimit_test(&rl, "quux", USEC_PER_SEC, 10, LOG_DEBUG, 0) == 1);
usleep_safe(USEC_PER_SEC);
/* The ratelimit is now expired (11 trials are suppressed, so the return value should be 12). */
assert_se(journal_ratelimit_test(&rl, "hoge", USEC_PER_SEC, 10, LOG_DEBUG, 0) == 1 + 11);
/* foo is still ratelimited. */
assert_se(journal_ratelimit_test(&rl, "foo", 10 * USEC_PER_SEC, 10, LOG_DEBUG, 0) == 0);
/* Still other priority and/or other IDs are not ratelimited. */
assert_se(journal_ratelimit_test(&rl, "hoge", USEC_PER_SEC, 10, LOG_INFO, 0) == 1);
assert_se(journal_ratelimit_test(&rl, "foo", 10 * USEC_PER_SEC, 10, LOG_INFO, 0) == 1);
assert_se(journal_ratelimit_test(&rl, "quux", USEC_PER_SEC, 10, LOG_DEBUG, 0) == 1);
}
DEFINE_TEST_MAIN(LOG_INFO);
src/libsystemd/sd-journal/journal-send.c
+2 -2
View File
@@ -121,7 +121,7 @@ _public_ int sd_journal_printv(int priority, const char *format, va_list ap) {
assert_return(priority <= 7, -EINVAL);
assert_return(format, -EINVAL);
xsprintf(p, "PRIORITY=%i", priority & LOG_PRIMASK);
xsprintf(p, "PRIORITY=%i", LOG_PRI(priority));
va_copy(aq, ap);
len = vsnprintf(buffer + 8, LINE_MAX, format, aq);
@@ -485,7 +485,7 @@ _public_ int sd_journal_printv_with_location(int priority, const char *file, con
assert_return(priority <= 7, -EINVAL);
assert_return(format, -EINVAL);
xsprintf(p, "PRIORITY=%i", priority & LOG_PRIMASK);
xsprintf(p, "PRIORITY=%i", LOG_PRI(priority));
va_copy(aq, ap);
len = vsnprintf(buffer + 8, LINE_MAX, format, aq);
src/udev/udev-manager.c
+1 -1
View File
@@ -885,7 +885,7 @@ static int on_ctrl_msg(UdevCtrl *uctrl, UdevCtrlMessageType type, const UdevCtrl
switch (type) {
case UDEV_CTRL_SET_LOG_LEVEL:
if ((value->intval & LOG_PRIMASK) != value->intval) {
if (LOG_PRI(value->intval) != value->intval) {
log_debug("Received invalid udev control message (SET_LOG_LEVEL, %i), ignoring.", value->intval);
break;
}