From 8ad5df6121c46753a6d12fafa5ab9da309ddb721 Mon Sep 17 00:00:00 2001 From: Pablo Neira Ayuso Date: Wed, 21 Oct 2009 01:43:07 +0200 Subject: conntrackd: add `DisableInternalCache' clause This patch adds the clause `DisableInternalCache' that allows you to bypass the internal cache. This clause can only be used with the notrack synchronization mode. Signed-off-by: Pablo Neira Ayuso --- include/Makefile.am | 2 +- include/conntrackd.h | 12 +-- include/internal.h | 39 +++++++++ src/Makefile.am | 1 + src/internal_bypass.c | 165 +++++++++++++++++++++++++++++++++++++ src/internal_cache.c | 220 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/read_config_lex.l | 1 + src/read_config_yy.y | 13 ++- src/run.c | 62 ++++++++------ src/stats-mode.c | 24 +++--- src/sync-alarm.c | 5 +- src/sync-ftfw.c | 19 +++-- src/sync-mode.c | 190 ++++--------------------------------------- src/sync-notrack.c | 53 ++++++++++-- 14 files changed, 572 insertions(+), 234 deletions(-) create mode 100644 include/internal.h create mode 100644 src/internal_bypass.c create mode 100644 src/internal_cache.c diff --git a/include/Makefile.am b/include/Makefile.am index a89490e..cbbca6b 100644 --- a/include/Makefile.am +++ b/include/Makefile.am @@ -4,5 +4,5 @@ noinst_HEADERS = alarm.h jhash.h cache.h linux_list.h linux_rbtree.h \ debug.h log.h hash.h mcast.h conntrack.h \ network.h filter.h queue.h vector.h cidr.h \ traffic_stats.h netlink.h fds.h event.h bitops.h channel.h \ - process.h origin.h external.h date.h + process.h origin.h internal.h external.h date.h diff --git a/include/conntrackd.h b/include/conntrackd.h index 7737532..c7f33f0 100644 --- a/include/conntrackd.h +++ b/include/conntrackd.h @@ -6,6 +6,7 @@ #include "alarm.h" #include "filter.h" #include "channel.h" +#include "internal.h" #include #include @@ -99,6 +100,7 @@ struct ct_conf { int error_queue_length; } channelc; struct { + int internal_cache_disable; int external_cache_disable; } sync; struct { @@ -177,7 +179,6 @@ struct ct_general_state { #define STATE_SYNC(x) state.sync->x struct ct_sync_state { - struct cache *internal; /* internal events cache (netlink) */ struct external_handler *external; struct multichannel *channel; @@ -239,18 +240,11 @@ extern union ct_state state; extern struct ct_general_state st; struct ct_mode { + struct internal_handler *internal; int (*init)(void); void (*run)(fd_set *readfds); int (*local)(int fd, int type, void *data); void (*kill)(void); - void (*dump)(struct nf_conntrack *ct); - int (*resync)(enum nf_conntrack_msg_type type, - struct nf_conntrack *ct, - void *data); - int (*purge)(void); - void (*event_new)(struct nf_conntrack *ct, int origin); - void (*event_upd)(struct nf_conntrack *ct, int origin); - int (*event_dst)(struct nf_conntrack *ct, int origin); }; /* conntrackd modes */ diff --git a/include/internal.h b/include/internal.h new file mode 100644 index 0000000..1f11340 --- /dev/null +++ b/include/internal.h @@ -0,0 +1,39 @@ +#ifndef _INTERNAL_H_ +#define _INTERNAL_H_ + +#include + +struct nf_conntrack; + +enum { + INTERNAL_F_POPULATE = (1 << 0), + INTERNAL_F_RESYNC = (1 << 1), + INTERNAL_F_MAX = (1 << 2) +}; + +struct internal_handler { + void *data; + unsigned int flags; + + int (*init)(void); + void (*close)(void); + + void (*new)(struct nf_conntrack *ct, int origin_type); + void (*update)(struct nf_conntrack *ct, int origin_type); + int (*destroy)(struct nf_conntrack *ct, int origin_type); + + void (*dump)(int fd, int type); + void (*populate)(struct nf_conntrack *ct); + void (*purge)(void); + int (*resync)(enum nf_conntrack_msg_type type, + struct nf_conntrack *ct, void *data); + void (*flush)(void); + + void (*stats)(int fd); + void (*stats_ext)(int fd); +}; + +extern struct internal_handler internal_cache; +extern struct internal_handler internal_bypass; + +#endif diff --git a/src/Makefile.am b/src/Makefile.am index 8b36642..76f0e73 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -22,6 +22,7 @@ conntrackd_SOURCES = alarm.c main.c run.c hash.c queue.c rbtree.c \ channel.c multichannel.c channel_mcast.c channel_udp.c \ tcp.c channel_tcp.c \ external_cache.c external_inject.c \ + internal_cache.c internal_bypass.c \ read_config_yy.y read_config_lex.l # yacc and lex generate dirty code diff --git a/src/internal_bypass.c b/src/internal_bypass.c new file mode 100644 index 0000000..4caaf4f --- /dev/null +++ b/src/internal_bypass.c @@ -0,0 +1,165 @@ +/* + * (C) 2009 by Pablo Neira Ayuso + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * This feature has been sponsored by 6WIND . + */ +#include "conntrackd.h" +#include "sync.h" +#include "log.h" +#include "cache.h" +#include "netlink.h" +#include "network.h" +#include "origin.h" + +static int _init(void) +{ + return 0; +} + +static void _close(void) +{ +} + +static int dump_cb(enum nf_conntrack_msg_type type, + struct nf_conntrack *ct, void *data) +{ + char buf[1024]; + int size, *fd = data; + + size = nfct_snprintf(buf, 1024, ct, NFCT_T_UNKNOWN, NFCT_O_DEFAULT, 0); + if (size < 1024) { + buf[size] = '\n'; + size++; + } + send(*fd, buf, size, 0); + + return NFCT_CB_CONTINUE; +} + +static void dump(int fd, int type) +{ + struct nfct_handle *h; + u_int32_t family = AF_UNSPEC; + int ret; + + h = nfct_open(CONNTRACK, 0); + if (h == NULL) { + dlog(LOG_ERR, "can't allocate memory for the internal cache"); + return; + } + nfct_callback_register(h, NFCT_T_ALL, dump_cb, &fd); + ret = nfct_query(h, NFCT_Q_DUMP, &family); + if (ret == -1) { + dlog(LOG_ERR, "can't dump kernel table"); + } + nfct_close(h); +} + +static void flush(void) +{ + nl_flush_conntrack_table(STATE(flush)); +} + +struct { + uint32_t new; + uint32_t upd; + uint32_t del; +} internal_bypass_stats; + +static void stats(int fd) +{ + char buf[512]; + int size; + + size = sprintf(buf, "internal bypass:\n" + "connections new:\t\t%12u\n" + "connections updated:\t\t%12u\n" + "connections destroyed:\t\t%12u\n\n", + internal_bypass_stats.new, + internal_bypass_stats.upd, + internal_bypass_stats.del); + + send(fd, buf, size, 0); +} + +/* unused, INTERNAL_F_POPULATE is unset. No cache, nothing to populate. */ +static void populate(struct nf_conntrack *ct) +{ +} + +/* unused, INTERNAL_F_RESYNC is unset. */ +static void purge(void) +{ +} + +/* unused, INTERNAL_F_RESYNC is unset. Nothing to resync, we have no cache. */ +static int resync(enum nf_conntrack_msg_type type, + struct nf_conntrack *ct, + void *data) +{ + return NFCT_CB_CONTINUE; +} + +static void +event_new_sync(struct nf_conntrack *ct, int origin) +{ + struct nethdr *net; + + /* this event has been triggered by me, skip */ + if (origin != CTD_ORIGIN_NOT_ME) + return; + + net = BUILD_NETMSG(ct, NET_T_STATE_NEW); + multichannel_send(STATE_SYNC(channel), net); + internal_bypass_stats.new++; +} + +static void +event_update_sync(struct nf_conntrack *ct, int origin) +{ + struct nethdr *net; + + /* this event has been triggered by me, skip */ + if (origin != CTD_ORIGIN_NOT_ME) + return; + + net = BUILD_NETMSG(ct, NET_T_STATE_UPD); + multichannel_send(STATE_SYNC(channel), net); + internal_bypass_stats.upd++; +} + +static int +event_destroy_sync(struct nf_conntrack *ct, int origin) +{ + struct nethdr *net; + + /* this event has been triggered by me, skip */ + if (origin != CTD_ORIGIN_NOT_ME) + return 1; + + net = BUILD_NETMSG(ct, NET_T_STATE_DEL); + multichannel_send(STATE_SYNC(channel), net); + internal_bypass_stats.del++; + + return 1; +} + +struct internal_handler internal_bypass = { + .init = _init, + .close = _close, + .dump = dump, + .flush = flush, + .stats = stats, + .stats_ext = stats, + .populate = populate, + .purge = purge, + .resync = resync, + .new = event_new_sync, + .update = event_update_sync, + .destroy = event_destroy_sync, +}; diff --git a/src/internal_cache.c b/src/internal_cache.c new file mode 100644 index 0000000..daadfd6 --- /dev/null +++ b/src/internal_cache.c @@ -0,0 +1,220 @@ +/* + * (C) 2009 by Pablo Neira Ayuso + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + */ +#include "conntrackd.h" +#include "sync.h" +#include "log.h" +#include "cache.h" +#include "netlink.h" +#include "network.h" +#include "origin.h" + +static inline void sync_send(struct cache_object *obj, int query) +{ + STATE_SYNC(sync)->enqueue(obj, query); +} + +static int _init(void) +{ + STATE(mode)->internal->data = + cache_create("internal", + STATE_SYNC(sync)->internal_cache_flags, + STATE_SYNC(sync)->internal_cache_extra); + + if (!STATE(mode)->internal->data) { + dlog(LOG_ERR, "can't allocate memory for the internal cache"); + return -1; + } + return 0; +} + +static void _close(void) +{ + cache_destroy(STATE(mode)->internal->data); +} + +static void dump(int fd, int type) +{ + cache_dump(STATE(mode)->internal->data, fd, NFCT_O_PLAIN); +} + +static void flush(void) +{ + cache_flush(STATE(mode)->internal->data); +} + +static void stats(int fd) +{ + cache_stats(STATE(mode)->internal->data, fd); +} + +static void stats_ext(int fd) +{ + cache_stats_extended(STATE(mode)->internal->data, fd); +} + +static void populate(struct nf_conntrack *ct) +{ + /* This is required by kernels < 2.6.20 */ + nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES); + nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS); + nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES); + nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS); + nfct_attr_unset(ct, ATTR_USE); + + cache_update_force(STATE(mode)->internal->data, ct); +} + +static int purge_step(void *data1, void *data2) +{ + struct cache_object *obj = data2; + + STATE(get_retval) = 0; + nl_get_conntrack(STATE(get), obj->ct); /* modifies STATE(get_reval) */ + if (!STATE(get_retval)) { + if (obj->status != C_OBJ_DEAD) { + cache_object_set_status(obj, C_OBJ_DEAD); + sync_send(obj, NET_T_STATE_DEL); + cache_object_put(obj); + } + } + + return 0; +} + +static void purge(void) +{ + cache_iterate(STATE(mode)->internal->data, NULL, purge_step); +} + +static int resync(enum nf_conntrack_msg_type type, + struct nf_conntrack *ct, + void *data) +{ + struct cache_object *obj; + + if (ct_filter_conntrack(ct, 1)) + return NFCT_CB_CONTINUE; + + /* This is required by kernels < 2.6.20 */ + nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES); + nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS); + nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES); + nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS); + nfct_attr_unset(ct, ATTR_USE); + + obj = cache_update_force(STATE(mode)->internal->data, ct); + if (obj == NULL) + return NFCT_CB_CONTINUE; + + switch (obj->status) { + case C_OBJ_NEW: + sync_send(obj, NET_T_STATE_NEW); + break; + case C_OBJ_ALIVE: + sync_send(obj, NET_T_STATE_UPD); + break; + } + return NFCT_CB_CONTINUE; +} + +static void +event_new_sync(struct nf_conntrack *ct, int origin) +{ + struct cache_object *obj; + int id; + + /* this event has been triggered by a direct inject, skip */ + if (origin == CTD_ORIGIN_INJECT) + return; + + /* required by linux kernel <= 2.6.20 */ + nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES); + nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS); + nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES); + nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS); + + obj = cache_find(STATE(mode)->internal->data, ct, &id); + if (obj == NULL) { +retry: + obj = cache_object_new(STATE(mode)->internal->data, ct); + if (obj == NULL) + return; + if (cache_add(STATE(mode)->internal->data, obj, id) == -1) { + cache_object_free(obj); + return; + } + /* only synchronize events that have been triggered by other + * processes or the kernel, but don't propagate events that + * have been triggered by conntrackd itself, eg. commits. */ + if (origin == CTD_ORIGIN_NOT_ME) + sync_send(obj, NET_T_STATE_NEW); + } else { + cache_del(STATE(mode)->internal->data, obj); + cache_object_free(obj); + goto retry; + } +} + +static void +event_update_sync(struct nf_conntrack *ct, int origin) +{ + struct cache_object *obj; + + /* this event has been triggered by a direct inject, skip */ + if (origin == CTD_ORIGIN_INJECT) + return; + + obj = cache_update_force(STATE(mode)->internal->data, ct); + if (obj == NULL) + return; + + if (origin == CTD_ORIGIN_NOT_ME) + sync_send(obj, NET_T_STATE_UPD); +} + +static int +event_destroy_sync(struct nf_conntrack *ct, int origin) +{ + struct cache_object *obj; + int id; + + /* this event has been triggered by a direct inject, skip */ + if (origin == CTD_ORIGIN_INJECT) + return 0; + + /* we don't synchronize events for objects that are not in the cache */ + obj = cache_find(STATE(mode)->internal->data, ct, &id); + if (obj == NULL) + return 0; + + if (obj->status != C_OBJ_DEAD) { + cache_object_set_status(obj, C_OBJ_DEAD); + if (origin == CTD_ORIGIN_NOT_ME) { + sync_send(obj, NET_T_STATE_DEL); + } + cache_object_put(obj); + } + return 1; +} + +struct internal_handler internal_cache = { + .flags = INTERNAL_F_POPULATE | INTERNAL_F_RESYNC, + .init = _init, + .close = _close, + .dump = dump, + .flush = flush, + .stats = stats, + .stats_ext = stats_ext, + .populate = populate, + .purge = purge, + .resync = resync, + .new = event_new_sync, + .update = event_update_sync, + .destroy = event_destroy_sync, +}; diff --git a/src/read_config_lex.l b/src/read_config_lex.l index b4be6f0..b2d4bdb 100644 --- a/src/read_config_lex.l +++ b/src/read_config_lex.l @@ -136,6 +136,7 @@ notrack [N|n][O|o][T|t][R|r][A|a][C|c][K|k] "Type" { return T_TYPE; } "Priority" { return T_PRIO; } "NetlinkEventsReliable" { return T_NETLINK_EVENTS_RELIABLE; } +"DisableInternalCache" { return T_DISABLE_INTERNAL_CACHE; } "DisableExternalCache" { return T_DISABLE_EXTERNAL_CACHE; } "ErrorQueueLength" { return T_ERROR_QUEUE_LENGTH; } diff --git a/src/read_config_yy.y b/src/read_config_yy.y index 5075cf0..157e945 100644 --- a/src/read_config_yy.y +++ b/src/read_config_yy.y @@ -72,7 +72,7 @@ static void __max_dedicated_links_reached(void); %token T_FROM T_USERSPACE T_KERNELSPACE T_EVENT_ITER_LIMIT T_DEFAULT %token T_NETLINK_OVERRUN_RESYNC T_NICE T_IPV4_DEST_ADDR T_IPV6_DEST_ADDR %token T_SCHEDULER T_TYPE T_PRIO T_NETLINK_EVENTS_RELIABLE -%token T_DISABLE_EXTERNAL_CACHE T_ERROR_QUEUE_LENGTH +%token T_DISABLE_INTERNAL_CACHE T_DISABLE_EXTERNAL_CACHE T_ERROR_QUEUE_LENGTH %token T_IP T_PATH_VAL %token T_NUMBER @@ -852,9 +852,20 @@ sync_mode_notrack_list: sync_mode_notrack_line: timeout | purge + | disable_internal_cache | disable_external_cache ; +disable_internal_cache: T_DISABLE_INTERNAL_CACHE T_ON +{ + conf.sync.internal_cache_disable = 1; +}; + +disable_internal_cache: T_DISABLE_INTERNAL_CACHE T_OFF +{ + conf.sync.internal_cache_disable = 0; +}; + disable_external_cache: T_DISABLE_EXTERNAL_CACHE T_ON { conf.sync.external_cache_disable = 1; diff --git a/src/run.c b/src/run.c index 54ab1a5..803bbcc 100644 --- a/src/run.c +++ b/src/run.c @@ -28,6 +28,7 @@ #include "process.h" #include "origin.h" #include "date.h" +#include "internal.h" #include #include @@ -56,7 +57,9 @@ void killer(int foo) local_server_destroy(&STATE(local)); STATE(mode)->kill(); - nfct_close(STATE(dump)); /* cache_wt needs this here */ + if (STATE(mode)->internal->flags & INTERNAL_F_POPULATE) { + nfct_close(STATE(dump)); + } destroy_fds(STATE(fds)); unlink(CONFIG(lockfile)); @@ -210,9 +213,13 @@ static int local_handler(int fd, void *data) } break; case RESYNC_MASTER: - STATE(stats).nl_kernel_table_resync++; - dlog(LOG_NOTICE, "resync with master table"); - nl_dump_conntrack_table(STATE(dump)); + if (STATE(mode)->internal->flags & INTERNAL_F_POPULATE) { + STATE(stats).nl_kernel_table_resync++; + dlog(LOG_NOTICE, "resync with master table"); + nl_dump_conntrack_table(STATE(dump)); + } else { + dlog(LOG_NOTICE, "resync is unsupported in this mode"); + } break; case STATS_RUNTIME: dump_stats_runtime(fd); @@ -238,8 +245,8 @@ static void do_overrun_resync_alarm(struct alarm_block *a, void *data) static void do_polling_alarm(struct alarm_block *a, void *data) { - if (STATE(mode)->purge) - STATE(mode)->purge(); + if (STATE(mode)->internal->purge) + STATE(mode)->internal->purge(); nl_send_resync(STATE(resync)); add_alarm(&STATE(polling_alarm), CONFIG(poll_kernel_secs), 0); @@ -264,13 +271,13 @@ static int event_handler(const struct nlmsghdr *nlh, switch(type) { case NFCT_T_NEW: - STATE(mode)->event_new(ct, origin_type); + STATE(mode)->internal->new(ct, origin_type); break; case NFCT_T_UPDATE: - STATE(mode)->event_upd(ct, origin_type); + STATE(mode)->internal->update(ct, origin_type); break; case NFCT_T_DESTROY: - if (STATE(mode)->event_dst(ct, origin_type)) + if (STATE(mode)->internal->destroy(ct, origin_type)) update_traffic_stats(ct); break; default: @@ -295,7 +302,7 @@ static int dump_handler(enum nf_conntrack_msg_type type, switch(type) { case NFCT_T_UPDATE: - STATE(mode)->dump(ct); + STATE(mode)->internal->populate(ct); break; default: STATE(stats).nl_dump_unknown_type++; @@ -371,23 +378,26 @@ init(void) } nfct_callback_register(STATE(resync), NFCT_T_ALL, - STATE(mode)->resync, + STATE(mode)->internal->resync, NULL); register_fd(nfct_fd(STATE(resync)), STATE(fds)); fcntl(nfct_fd(STATE(resync)), F_SETFL, O_NONBLOCK); - STATE(dump) = nfct_open(CONNTRACK, 0); - if (STATE(dump) == NULL) { - dlog(LOG_ERR, "can't open netlink handler: %s", - strerror(errno)); - dlog(LOG_ERR, "no ctnetlink kernel support?"); - return -1; - } - nfct_callback_register(STATE(dump), NFCT_T_ALL, dump_handler, NULL); + if (STATE(mode)->internal->flags & INTERNAL_F_POPULATE) { + STATE(dump) = nfct_open(CONNTRACK, 0); + if (STATE(dump) == NULL) { + dlog(LOG_ERR, "can't open netlink handler: %s", + strerror(errno)); + dlog(LOG_ERR, "no ctnetlink kernel support?"); + return -1; + } + nfct_callback_register(STATE(dump), NFCT_T_ALL, + dump_handler, NULL); - if (nl_dump_conntrack_table(STATE(dump)) == -1) { - dlog(LOG_ERR, "can't get kernel conntrack table"); - return -1; + if (nl_dump_conntrack_table(STATE(dump)) == -1) { + dlog(LOG_ERR, "can't get kernel conntrack table"); + return -1; + } } STATE(get) = nfct_open(CONNTRACK, 0); @@ -499,7 +509,9 @@ static void __run(struct timeval *next_alarm) * we resync ourselves. */ nl_resize_socket_buffer(STATE(event)); - if (CONFIG(nl_overrun_resync) > 0) { + if (CONFIG(nl_overrun_resync) > 0 && + STATE(mode)->internal->flags & + INTERNAL_F_RESYNC) { add_alarm(&STATE(resync_alarm), CONFIG(nl_overrun_resync),0); } @@ -523,8 +535,8 @@ static void __run(struct timeval *next_alarm) } if (FD_ISSET(nfct_fd(STATE(resync)), &readfds)) { nfct_catch(STATE(resync)); - if (STATE(mode)->purge) - STATE(mode)->purge(); + if (STATE(mode)->internal->purge) + STATE(mode)->internal->purge(); } } else { /* using polling mode */ diff --git a/src/stats-mode.c b/src/stats-mode.c index 5cfb638..0403ce2 100644 --- a/src/stats-mode.c +++ b/src/stats-mode.c @@ -21,6 +21,7 @@ #include "cache.h" #include "log.h" #include "conntrackd.h" +#include "internal.h" #include #include @@ -87,7 +88,7 @@ static int local_handler_stats(int fd, int type, void *data) return ret; } -static void dump_stats(struct nf_conntrack *ct) +static void populate_stats(struct nf_conntrack *ct) { nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES); nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS); @@ -134,11 +135,9 @@ static int purge_step(void *data1, void *data2) return 0; } -static int purge_stats(void) +static void purge_stats(void) { cache_iterate(STATE_STATS(cache), NULL, purge_step); - - return 0; } static void @@ -188,15 +187,20 @@ event_destroy_stats(struct nf_conntrack *ct, int origin) return 0; } +static struct internal_handler internal_cache_stats = { + .flags = INTERNAL_F_POPULATE | INTERNAL_F_RESYNC, + .populate = populate_stats, + .resync = resync_stats, + .purge = purge_stats, + .new = event_new_stats, + .update = event_update_stats, + .destroy = event_destroy_stats +}; + struct ct_mode stats_mode = { .init = init_stats, .run = NULL, .local = local_handler_stats, .kill = kill_stats, - .dump = dump_stats, - .resync = resync_stats, - .purge = purge_stats, - .event_new = event_new_stats, - .event_upd = event_update_stats, - .event_dst = event_destroy_stats + .internal = &internal_cache_stats, }; diff --git a/src/sync-alarm.c b/src/sync-alarm.c index 4757026..0fc7943 100644 --- a/src/sync-alarm.c +++ b/src/sync-alarm.c @@ -109,7 +109,8 @@ static int alarm_recv(const struct nethdr *net) static void alarm_enqueue(struct cache_object *obj, int query) { - struct cache_alarm *ca = cache_get_extra(STATE_SYNC(internal), obj); + struct cache_alarm *ca = + cache_get_extra(STATE(mode)->internal->data, obj); if (queue_add(STATE_SYNC(tx_queue), &ca->qnode)) cache_object_get(obj); } @@ -134,7 +135,7 @@ static int tx_queue_xmit(struct queue_node *n, const void *data) int type; ca = (struct cache_alarm *)n; - obj = cache_data_get_object(STATE_SYNC(internal), ca); + obj = cache_data_get_object(STATE(mode)->internal->data, ca); type = object_status_to_network_type(obj->status); net = BUILD_NETMSG(obj->ct, type); multichannel_send(STATE_SYNC(channel), net); diff --git a/src/sync-ftfw.c b/src/sync-ftfw.c index 0d31e17..86edeab 100644 --- a/src/sync-ftfw.c +++ b/src/sync-ftfw.c @@ -166,7 +166,8 @@ static void ftfw_kill(void) static int do_cache_to_tx(void *data1, void *data2) { struct cache_object *obj = data2; - struct cache_ftfw *cn = cache_get_extra(STATE_SYNC(internal), obj); + struct cache_ftfw *cn = + cache_get_extra(STATE(mode)->internal->data, obj); if (queue_in(rs_queue, &cn->qnode)) { queue_del(&cn->qnode); @@ -224,7 +225,8 @@ static int ftfw_local(int fd, int type, void *data) break; case SEND_BULK: dlog(LOG_NOTICE, "sending bulk update"); - cache_iterate(STATE_SYNC(internal), NULL, do_cache_to_tx); + cache_iterate(STATE(mode)->internal->data, + NULL, do_cache_to_tx); break; case STATS_RSQUEUE: ftfw_local_queue(fd); @@ -303,7 +305,7 @@ static int rs_queue_empty(struct queue_node *n, const void *data) cn = (struct cache_ftfw *) n; if (h == NULL) { queue_del(n); - obj = cache_data_get_object(STATE_SYNC(internal), cn); + obj = cache_data_get_object(STATE(mode)->internal->data, cn); cache_object_put(obj); return 0; } @@ -314,7 +316,7 @@ static int rs_queue_empty(struct queue_node *n, const void *data) dp("queue: deleting from queue (seq=%u)\n", cn->seq); queue_del(n); - obj = cache_data_get_object(STATE_SYNC(internal), cn); + obj = cache_data_get_object(STATE(mode)->internal->data, cn); cache_object_put(obj); break; } @@ -347,7 +349,7 @@ static int digest_msg(const struct nethdr *net) } else if (IS_RESYNC(net)) { dp("RESYNC ALL\n"); - cache_iterate(STATE_SYNC(internal), NULL, do_cache_to_tx); + cache_iterate(STATE(mode)->internal->data, NULL, do_cache_to_tx); return MSG_CTL; } else if (IS_ALIVE(net)) @@ -464,7 +466,7 @@ static void rs_queue_purge_full(void) struct cache_object *obj; cn = (struct cache_ftfw *)n; - obj = cache_data_get_object(STATE_SYNC(internal), cn); + obj = cache_data_get_object(STATE(mode)->internal->data, cn); cache_object_put(obj); break; } @@ -512,7 +514,7 @@ static int tx_queue_xmit(struct queue_node *n, const void *data) struct nethdr *net; cn = (struct cache_ftfw *)n; - obj = cache_data_get_object(STATE_SYNC(internal), cn); + obj = cache_data_get_object(STATE(mode)->internal->data, cn); type = object_status_to_network_type(obj->status); net = BUILD_NETMSG(obj->ct, type); nethdr_set_hello(net); @@ -546,7 +548,8 @@ static void ftfw_xmit(void) static void ftfw_enqueue(struct cache_object *obj, int type) { - struct cache_ftfw *cn = cache_get_extra(STATE_SYNC(internal), obj); + struct cache_ftfw *cn = + cache_get_extra(STATE(mode)->internal->data, obj); if (queue_in(rs_queue, &cn->qnode)) { queue_del(&cn->qnode); queue_add(STATE_SYNC(tx_queue), &cn->qnode); diff --git a/src/sync-mode.c b/src/sync-mode.c index 63fae68..ecc2f0d 100644 --- a/src/sync-mode.c +++ b/src/sync-mode.c @@ -28,6 +28,7 @@ #include "queue.h" #include "process.h" #include "origin.h" +#include "internal.h" #include "external.h" #include @@ -246,7 +247,7 @@ static void do_reset_cache_alarm(struct alarm_block *a, void *data) exit(EXIT_SUCCESS); } /* this is not required if events don't get lost */ - cache_flush(STATE_SYNC(internal)); + STATE(mode)->internal->flush(); } static int init_sync(void) @@ -276,15 +277,15 @@ static int init_sync(void) if (STATE_SYNC(sync)->init) STATE_SYNC(sync)->init(); - STATE_SYNC(internal) = - cache_create("internal", - STATE_SYNC(sync)->internal_cache_flags, - STATE_SYNC(sync)->internal_cache_extra); + if (CONFIG(sync).internal_cache_disable == 0) { + STATE(mode)->internal = &internal_cache; + } else { + STATE(mode)->internal = &internal_bypass; + dlog(LOG_NOTICE, "disabling internal cache"); - if (!STATE_SYNC(internal)) { - dlog(LOG_ERR, "can't allocate memory for the internal cache"); - return -1; } + if (STATE(mode)->internal->init() == -1) + return -1; if (CONFIG(sync).external_cache_disable == 0) { STATE_SYNC(external) = &external_cache; @@ -389,7 +390,7 @@ static void run_sync(fd_set *readfds) static void kill_sync(void) { - cache_destroy(STATE_SYNC(internal)); + STATE(mode)->internal->close(); STATE_SYNC(external)->close(); multichannel_close(STATE_SYNC(channel)); @@ -466,7 +467,7 @@ static int local_handler_sync(int fd, int type, void *data) case DUMP_INTERNAL: ret = fork_process_new(CTD_PROC_ANY, 0, NULL, NULL); if (ret == 0) { - cache_dump(STATE_SYNC(internal), fd, NFCT_O_PLAIN); + STATE(mode)->internal->dump(fd, NFCT_O_PLAIN); exit(EXIT_SUCCESS); } break; @@ -480,7 +481,7 @@ static int local_handler_sync(int fd, int type, void *data) case DUMP_INT_XML: ret = fork_process_new(CTD_PROC_ANY, 0, NULL, NULL); if (ret == 0) { - cache_dump(STATE_SYNC(internal), fd, NFCT_O_XML); + STATE(mode)->internal->dump(fd, NFCT_O_XML); exit(EXIT_SUCCESS); } break; @@ -512,14 +513,14 @@ static int local_handler_sync(int fd, int type, void *data) /* inmediate flush, remove pending flush scheduled if any */ del_alarm(&STATE_SYNC(reset_cache_alarm)); dlog(LOG_NOTICE, "flushing caches"); - cache_flush(STATE_SYNC(internal)); + STATE(mode)->internal->flush(); STATE_SYNC(external)->flush(); break; case FLUSH_INT_CACHE: /* inmediate flush, remove pending flush scheduled if any */ del_alarm(&STATE_SYNC(reset_cache_alarm)); dlog(LOG_NOTICE, "flushing internal cache"); - cache_flush(STATE_SYNC(internal)); + STATE(mode)->internal->flush(); break; case FLUSH_EXT_CACHE: dlog(LOG_NOTICE, "flushing external cache"); @@ -529,7 +530,7 @@ static int local_handler_sync(int fd, int type, void *data) killer(0); break; case STATS: - cache_stats(STATE_SYNC(internal), fd); + STATE(mode)->internal->stats(fd); STATE_SYNC(external)->stats(fd); dump_traffic_stats(fd); multichannel_stats(STATE_SYNC(channel), fd); @@ -540,7 +541,7 @@ static int local_handler_sync(int fd, int type, void *data) multichannel_stats(STATE_SYNC(channel), fd); break; case STATS_CACHE: - cache_stats_extended(STATE_SYNC(internal), fd); + STATE(mode)->internal->stats_ext(fd); STATE_SYNC(external)->stats_ext(fd); break; case STATS_LINK: @@ -559,167 +560,10 @@ static int local_handler_sync(int fd, int type, void *data) return ret; } -static void sync_send(struct cache_object *obj, int query) -{ - STATE_SYNC(sync)->enqueue(obj, query); -} - -static void dump_sync(struct nf_conntrack *ct) -{ - /* This is required by kernels < 2.6.20 */ - nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES); - nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS); - nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES); - nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS); - nfct_attr_unset(ct, ATTR_USE); - - cache_update_force(STATE_SYNC(internal), ct); -} - -static int purge_step(void *data1, void *data2) -{ - struct cache_object *obj = data2; - - STATE(get_retval) = 0; - nl_get_conntrack(STATE(get), obj->ct); /* modifies STATE(get_reval) */ - if (!STATE(get_retval)) { - if (obj->status != C_OBJ_DEAD) { - cache_object_set_status(obj, C_OBJ_DEAD); - sync_send(obj, NET_T_STATE_DEL); - cache_object_put(obj); - } - } - - return 0; -} - -static int purge_sync(void) -{ - cache_iterate(STATE_SYNC(internal), NULL, purge_step); - - return 0; -} - -static int resync_sync(enum nf_conntrack_msg_type type, - struct nf_conntrack *ct, - void *data) -{ - struct cache_object *obj; - - if (ct_filter_conntrack(ct, 1)) - return NFCT_CB_CONTINUE; - - /* This is required by kernels < 2.6.20 */ - nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES); - nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS); - nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES); - nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS); - nfct_attr_unset(ct, ATTR_USE); - - obj = cache_update_force(STATE_SYNC(internal), ct); - if (obj == NULL) - return NFCT_CB_CONTINUE; - - switch (obj->status) { - case C_OBJ_NEW: - sync_send(obj, NET_T_STATE_NEW); - break; - case C_OBJ_ALIVE: - sync_send(obj, NET_T_STATE_UPD); - break; - } - return NFCT_CB_CONTINUE; -} - -static void -event_new_sync(struct nf_conntrack *ct, int origin) -{ - struct cache_object *obj; - int id; - - /* this event has been triggered by a direct inject, skip */ - if (origin == CTD_ORIGIN_INJECT) - return; - - /* required by linux kernel <= 2.6.20 */ - nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES); - nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS); - nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES); - nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS); - - obj = cache_find(STATE_SYNC(internal), ct, &id); - if (obj == NULL) { -retry: - obj = cache_object_new(STATE_SYNC(internal), ct); - if (obj == NULL) - return; - if (cache_add(STATE_SYNC(internal), obj, id) == -1) { - cache_object_free(obj); - return; - } - /* only synchronize events that have been triggered by other - * processes or the kernel, but don't propagate events that - * have been triggered by conntrackd itself, eg. commits. */ - if (origin == CTD_ORIGIN_NOT_ME) - sync_send(obj, NET_T_STATE_NEW); - } else { - cache_del(STATE_SYNC(internal), obj); - cache_object_free(obj); - goto retry; - } -} - -static void -event_update_sync(struct nf_conntrack *ct, int origin) -{ - struct cache_object *obj; - - /* this event has been triggered by a direct inject, skip */ - if (origin == CTD_ORIGIN_INJECT) - return; - - obj = cache_update_force(STATE_SYNC(internal), ct); - if (obj == NULL) - return; - - if (origin == CTD_ORIGIN_NOT_ME) - sync_send(obj, NET_T_STATE_UPD); -} - -static int -event_destroy_sync(struct nf_conntrack *ct, int origin) -{ - struct cache_object *obj; - int id; - - /* this event has been triggered by a direct inject, skip */ - if (origin == CTD_ORIGIN_INJECT) - return 0; - - /* we don't synchronize events for objects that are not in the cache */ - obj = cache_find(STATE_SYNC(internal), ct, &id); - if (obj == NULL) - return 0; - - if (obj->status != C_OBJ_DEAD) { - cache_object_set_status(obj, C_OBJ_DEAD); - if (origin == CTD_ORIGIN_NOT_ME) { - sync_send(obj, NET_T_STATE_DEL); - } - cache_object_put(obj); - } - return 1; -} - struct ct_mode sync_mode = { .init = init_sync, .run = run_sync, .local = local_handler_sync, .kill = kill_sync, - .dump = dump_sync, - .resync = resync_sync, - .purge = purge_sync, - .event_new = event_new_sync, - .event_upd = event_update_sync, - .event_dst = event_destroy_sync + /* the internal handler is set in run-time. */ }; diff --git a/src/sync-notrack.c b/src/sync-notrack.c index d9f273e..c4ad941 100644 --- a/src/sync-notrack.c +++ b/src/sync-notrack.c @@ -74,12 +74,44 @@ static void tx_queue_add_ctlmsg(uint32_t flags, uint32_t from, uint32_t to) static int do_cache_to_tx(void *data1, void *data2) { struct cache_object *obj = data2; - struct cache_notrack *cn = cache_get_extra(STATE_SYNC(internal), obj); + struct cache_notrack *cn = + cache_get_extra(STATE(mode)->internal->data, obj); if (queue_add(STATE_SYNC(tx_queue), &cn->qnode)) cache_object_get(obj); return 0; } +static int kernel_resync_cb(enum nf_conntrack_msg_type type, + struct nf_conntrack *ct, void *data) +{ + struct nethdr *net; + + net = BUILD_NETMSG(ct, NET_T_STATE_NEW); + multichannel_send(STATE_SYNC(channel), net); + + return NFCT_CB_CONTINUE; +} + +/* Only used if the internal cache is disabled. */ +static void kernel_resync(void) +{ + struct nfct_handle *h; + u_int32_t family = AF_UNSPEC; + int ret; + + h = nfct_open(CONNTRACK, 0); + if (h == NULL) { + dlog(LOG_ERR, "can't allocate memory for the internal cache"); + return; + } + nfct_callback_register(h, NFCT_T_ALL, kernel_resync_cb, NULL); + ret = nfct_query(h, NFCT_Q_DUMP, &family); + if (ret == -1) { + dlog(LOG_ERR, "can't dump kernel table"); + } + nfct_close(h); +} + static int notrack_local(int fd, int type, void *data) { int ret = LOCAL_RET_OK; @@ -91,7 +123,12 @@ static int notrack_local(int fd, int type, void *data) break; case SEND_BULK: dlog(LOG_NOTICE, "sending bulk update"); - cache_iterate(STATE_SYNC(internal), NULL, do_cache_to_tx); + if (CONFIG(sync).internal_cache_disable) { + kernel_resync(); + } else { + cache_iterate(STATE(mode)->internal->data, + NULL, do_cache_to_tx); + } break; default: ret = 0; @@ -107,7 +144,12 @@ static int digest_msg(const struct nethdr *net) return MSG_DATA; if (IS_RESYNC(net)) { - cache_iterate(STATE_SYNC(internal), NULL, do_cache_to_tx); + if (CONFIG(sync).internal_cache_disable) { + kernel_resync(); + } else { + cache_iterate(STATE(mode)->internal->data, + NULL, do_cache_to_tx); + } return MSG_CTL; } @@ -154,7 +196,7 @@ static int tx_queue_xmit(struct queue_node *n, const void *data2) struct nethdr *net; cn = (struct cache_ftfw *)n; - obj = cache_data_get_object(STATE_SYNC(internal), cn); + obj = cache_data_get_object(STATE(mode)->internal->data, cn); type = object_status_to_network_type(obj->status);; net = BUILD_NETMSG(obj->ct, type); @@ -175,7 +217,8 @@ static void notrack_xmit(void) static void notrack_enqueue(struct cache_object *obj, int query) { - struct cache_notrack *cn = cache_get_extra(STATE_SYNC(internal), obj); + struct cache_notrack *cn = + cache_get_extra(STATE(mode)->internal->data, obj); if (queue_add(STATE_SYNC(tx_queue), &cn->qnode)) cache_object_get(obj); } -- cgit v1.2.3