diff options
Diffstat (limited to 'proto/aggregator')
| -rw-r--r-- | proto/aggregator/Doc | 1 | ||||
| -rw-r--r-- | proto/aggregator/Makefile | 6 | ||||
| -rw-r--r-- | proto/aggregator/aggregator.c | 731 | ||||
| -rw-r--r-- | proto/aggregator/aggregator.h | 86 | ||||
| -rw-r--r-- | proto/aggregator/config.Y | 134 | ||||
| -rw-r--r-- | proto/aggregator/test.conf | 116 |
6 files changed, 1074 insertions, 0 deletions
diff --git a/proto/aggregator/Doc b/proto/aggregator/Doc new file mode 100644 index 00000000..6111f2f1 --- /dev/null +++ b/proto/aggregator/Doc @@ -0,0 +1 @@ +S aggregator.c diff --git a/proto/aggregator/Makefile b/proto/aggregator/Makefile new file mode 100644 index 00000000..d1dae8dd --- /dev/null +++ b/proto/aggregator/Makefile @@ -0,0 +1,6 @@ +src := aggregator.c +obj := $(src-o-files) +$(all-daemon) +$(cf-local) + +tests_objs := $(tests_objs) $(src-o-files) diff --git a/proto/aggregator/aggregator.c b/proto/aggregator/aggregator.c new file mode 100644 index 00000000..9d95c7db --- /dev/null +++ b/proto/aggregator/aggregator.c @@ -0,0 +1,731 @@ +/* + * BIRD Internet Routing Daemon -- Route aggregation + * + * (c) 2023--2023 Igor Putovny <igor.putovny@nic.cz> + * (c) 2023 CZ.NIC, z.s.p.o. + * + * Can be freely distributed and used under the terms of the GNU GPL. + */ + +/** + * DOC: Route aggregation + * + * This is an implementation of route aggregation functionality. + * It enables user to specify a set of route attributes in the configuarion file + * and then, for a given destination (net), aggregate routes with the same + * values of these attributes into a single multi-path route. + * + * Structure &channel contains pointer to aggregation list which is represented + * by &aggr_list_linearized. In rt_notify_aggregated(), attributes from this + * list are evaluated for every route of a given net and results are stored + * in &rte_val_list which contains pointer to this route and array of &f_val. + * Array of pointers to &rte_val_list entries is sorted using + * sort_rte_val_list(). For comparison of &f_val structures, val_compare() + * is used. Comparator function is written so that sorting is stable. If all + * attributes have the same values, routes are compared by their global IDs. + * + * After sorting, &rte_val_list entries containing equivalent routes will be + * adjacent to each other. Function process_rte_list() iterates through these + * entries to identify sequences of equivalent routes. New route will be + * created for each such sequence, even if only from a single route. + * Only attributes from the aggreagation list will be set for the new route. + * New &rta is created and prepare_rta() is used to copy static and dynamic + * attributes to new &rta from &rta of the original route. New route is created + * by create_merged_rte() from new &rta and exported to the routing table. + */ + +#undef LOCAL_DEBUG + +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif + +#include "nest/bird.h" +#include "nest/iface.h" +#include "filter/filter.h" +#include "aggregator.h" + +#include <stdlib.h> +/* +#include "nest/route.h" +#include "nest/iface.h" +#include "lib/resource.h" +#include "lib/event.h" +#include "lib/timer.h" +#include "lib/string.h" +#include "conf/conf.h" +#include "filter/filter.h" +#include "filter/data.h" +#include "lib/hash.h" +#include "lib/string.h" +#include "lib/alloca.h" +#include "lib/flowspec.h" +*/ + +/* Context of &f_val comparison. */ +struct cmp_ctx { + const struct aggregator_proto *p; + const struct network *net; + const int val_count; + u32 failed:1; +}; + +extern linpool *rte_update_pool; + +/* + * Set static attribute in @rta from static attribute in @old according to @sa. + */ +static void +rta_set_static_attr(struct rta *rta, const struct rta *old, struct f_static_attr sa) +{ + switch (sa.sa_code) + { + case SA_NET: + break; + + case SA_FROM: + rta->from = old->from; + break; + + case SA_GW: + rta->dest = RTD_UNICAST; + rta->nh.gw = old->nh.gw; + rta->nh.iface = old->nh.iface; + rta->nh.next = NULL; + rta->hostentry = NULL; + rta->nh.labels = 0; + break; + + case SA_SCOPE: + rta->scope = old->scope; + break; + + case SA_DEST: + rta->dest = old->dest; + rta->nh.gw = IPA_NONE; + rta->nh.iface = NULL; + rta->nh.next = NULL; + rta->hostentry = NULL; + rta->nh.labels = 0; + break; + + case SA_IFNAME: + rta->dest = RTD_UNICAST; + rta->nh.gw = IPA_NONE; + rta->nh.iface = old->nh.iface; + rta->nh.next = NULL; + rta->hostentry = NULL; + rta->nh.labels = 0; + break; + + case SA_GW_MPLS: + rta->nh.labels = old->nh.labels; + memcpy(&rta->nh.label, &old->nh.label, sizeof(u32) * old->nh.labels); + break; + + case SA_WEIGHT: + rta->nh.weight = old->nh.weight; + break; + + case SA_PREF: + rta->pref = old->pref; + break; + + default: + bug("Invalid static attribute access (%u/%u)", sa.f_type, sa.sa_code); + } +} + +/* + * Compare list of &f_val entries. + * @count: number of &f_val entries + */ +static int +same_val_list(const struct f_val *v1, const struct f_val *v2, uint len) +{ + for (uint i = 0; i < len; i++) + if (!val_same(&v1[i], &v2[i])) + return 0; + + return 1; +} + +/* + * Create and export new merged route. + * @old: first route in a sequence of equivalent routes that are to be merged + * @rte_val: first element in a sequence of equivalent rte_val_list entries + * @length: number of equivalent routes that are to be merged (at least 1) + * @ail: aggregation list + */ +static void +aggregator_bucket_update(struct aggregator_proto *p, struct aggregator_bucket *bucket, struct network *net) +{ + /* Empty bucket */ + if (!bucket->rte) + { + rte_update2(p->dst, net->n.addr, NULL, bucket->last_src); + bucket->last_src = NULL; + return; + } + + /* Allocate RTA and EA list */ + struct rta *rta = allocz(rta_size(bucket->rte->attrs)); + rta->dest = RTD_UNREACHABLE; + rta->source = RTS_AGGREGATED; + rta->scope = SCOPE_UNIVERSE; + + struct ea_list *eal = allocz(sizeof(struct ea_list) + sizeof(struct eattr) * p->aggr_on_da_count); + eal->next = NULL; + eal->count = 0; + rta->eattrs = eal; + + /* Seed the attributes from aggregator rule */ + for (uint i = 0; i < p->aggr_on_count; i++) + { + if (p->aggr_on[i].type == AGGR_ITEM_DYNAMIC_ATTR) + { + u32 ea_code = p->aggr_on[i].da.ea_code; + const struct eattr *e = ea_find(bucket->rte->attrs->eattrs, ea_code); + + if (e) + eal->attrs[eal->count++] = *e; + } + else if (p->aggr_on[i].type == AGGR_ITEM_STATIC_ATTR) + rta_set_static_attr(rta, bucket->rte->attrs, p->aggr_on[i].sa); + } + + struct rte *new = rte_get_temp(rta, bucket->rte->src); + new->net = net; + + /* + log("=============== CREATE MERGED ROUTE ==============="); + log("New route created: id = %d, protocol: %s", new->src->global_id, new->src->proto->name); + log("==================================================="); + */ + + /* merge filter needs one argument called "routes" */ + struct f_val val = { + .type = T_ROUTES_BLOCK, + .val.rte = bucket->rte, + }; + + /* Actually run the filter */ + enum filter_return fret = f_eval_rte(p->merge_by, &new, rte_update_pool, 1, &val, 0); + + /* Src must be stored now, rte_update2() may return new */ + struct rte_src *new_src = new ? new->src : NULL; + + /* Finally import the route */ + switch (fret) + { + /* Pass the route to the protocol */ + case F_ACCEPT: + rte_update2(p->dst, net->n.addr, new, bucket->last_src ?: new->src); + break; + + /* Something bad happened */ + default: + ASSERT_DIE(fret == F_ERROR); + /* fall through */ + + /* We actually don't want this route */ + case F_REJECT: + if (bucket->last_src) + rte_update2(p->dst, net->n.addr, NULL, bucket->last_src); + break; + } + + /* Switch source lock for bucket->last_src */ + if (bucket->last_src != new_src) + { + if (new_src) + rt_lock_source(new_src); + if (bucket->last_src) + rt_unlock_source(bucket->last_src); + + bucket->last_src = new_src; + } +} + +/* + * Reload all the buckets on reconfiguration if merge filter has changed. + * TODO: make this splitted + */ +static void +aggregator_reload_buckets(void *data) +{ + struct aggregator_proto *p = data; + + HASH_WALK(p->buckets, next_hash, b) + if (b->rte) + aggregator_bucket_update(p, b, b->rte->net); + HASH_WALK_END; +} + + +/* + * Evaluate static attribute of @rt1 according to @sa + * and store result in @pos. + */ +static void +eval_static_attr(const struct rte *rt1, struct f_static_attr sa, struct f_val *pos) +{ + const struct rta *rta = rt1->attrs; + +#define RESULT(_type, value, result) \ + do { \ + pos->type = _type; \ + pos->val.value = result; \ + } while (0) + + switch (sa.sa_code) + { + case SA_NET: RESULT(sa.f_type, net, rt1->net->n.addr); break; + case SA_FROM: RESULT(sa.f_type, ip, rta->from); break; + case SA_GW: RESULT(sa.f_type, ip, rta->nh.gw); break; + case SA_PROTO: RESULT(sa.f_type, s, rt1->src->proto->name); break; + case SA_SOURCE: RESULT(sa.f_type, i, rta->source); break; + case SA_SCOPE: RESULT(sa.f_type, i, rta->scope); break; + case SA_DEST: RESULT(sa.f_type, i, rta->dest); break; + case SA_IFNAME: RESULT(sa.f_type, s, rta->nh.iface ? rta->nh.iface->name : ""); break; + case SA_IFINDEX: RESULT(sa.f_type, i, rta->nh.iface ? rta->nh.iface->index : 0); break; + case SA_WEIGHT: RESULT(sa.f_type, i, rta->nh.weight + 1); break; + case SA_PREF: RESULT(sa.f_type, i, rta->pref); break; + case SA_GW_MPLS: RESULT(sa.f_type, i, rta->nh.labels ? rta->nh.label[0] : MPLS_NULL); break; + default: + bug("Invalid static attribute access (%u/%u)", sa.f_type, sa.sa_code); + } + +#undef RESULT +} + +/* + * Evaluate dynamic attribute of @rt1 according to @da + * and store result in @pos. + */ +static void +eval_dynamic_attr(const struct rte *rt1, struct f_dynamic_attr da, struct f_val *pos) +{ + const struct rta *rta = rt1->attrs; + const struct eattr *e = ea_find(rta->eattrs, da.ea_code); + +#define RESULT(_type, value, result) \ + do { \ + pos->type = _type; \ + pos->val.value = result; \ + } while (0) + +#define RESULT_VOID \ + do { \ + pos->type = T_VOID; \ + } while (0) + + if (!e) + { + /* A special case: undefined as_path looks like empty as_path */ + if (da.type == EAF_TYPE_AS_PATH) + { + RESULT(T_PATH, ad, &null_adata); + return; + } + + /* The same special case for int_set */ + if (da.type == EAF_TYPE_INT_SET) + { + RESULT(T_CLIST, ad, &null_adata); + return; + } + + /* The same special case for ec_set */ + if (da.type == EAF_TYPE_EC_SET) + { + RESULT(T_ECLIST, ad, &null_adata); + return; + } + + /* The same special case for lc_set */ + if (da.type == EAF_TYPE_LC_SET) + { + RESULT(T_LCLIST, ad, &null_adata); + return; + } + + /* Undefined value */ + RESULT_VOID; + return; + } + + switch (e->type & EAF_TYPE_MASK) + { + case EAF_TYPE_INT: + RESULT(da.f_type, i, e->u.data); + break; + case EAF_TYPE_ROUTER_ID: + RESULT(T_QUAD, i, e->u.data); + break; + case EAF_TYPE_OPAQUE: + RESULT(T_ENUM_EMPTY, i, 0); + break; + case EAF_TYPE_IP_ADDRESS: + RESULT(T_IP, ip, *((ip_addr *) e->u.ptr->data)); + break; + case EAF_TYPE_AS_PATH: + RESULT(T_PATH, ad, e->u.ptr); + break; + case EAF_TYPE_BITFIELD: + RESULT(T_BOOL, i, !!(e->u.data & (1u << da.bit))); + break; + case EAF_TYPE_INT_SET: + RESULT(T_CLIST, ad, e->u.ptr); + break; + case EAF_TYPE_EC_SET: + RESULT(T_ECLIST, ad, e->u.ptr); + break; + case EAF_TYPE_LC_SET: + RESULT(T_LCLIST, ad, e->u.ptr); + break; + default: + bug("Unknown dynamic attribute type"); + } + +#undef RESULT +#undef RESULT_VOID +} + +static inline u32 aggr_route_hash(const rte *e) +{ + struct { + net *net; + struct rte_src *src; + } obj = { + .net = e->net, + .src = e->src, + }; + + return mem_hash(&obj, sizeof obj); +} + +#define AGGR_RTE_KEY(n) (&(n)->rte) +#define AGGR_RTE_NEXT(n) ((n)->next_hash) +#define AGGR_RTE_EQ(a,b) (((a)->src == (b)->src) && ((a)->net == (b)->net)) +#define AGGR_RTE_FN(_n) aggr_route_hash(_n) +#define AGGR_RTE_ORDER 4 /* Initial */ + +#define AGGR_RTE_REHASH aggr_rte_rehash +#define AGGR_RTE_PARAMS /8, *2, 2, 2, 4, 24 + +HASH_DEFINE_REHASH_FN(AGGR_RTE, struct aggregator_route); + + +#define AGGR_BUCK_KEY(n) (n) +#define AGGR_BUCK_NEXT(n) ((n)->next_hash) +#define AGGR_BUCK_EQ(a,b) (((a)->hash == (b)->hash) && (same_val_list((a)->aggr_data, (b)->aggr_data, p->aggr_on_count))) +#define AGGR_BUCK_FN(n) ((n)->hash) +#define AGGR_BUCK_ORDER 4 /* Initial */ + +#define AGGR_BUCK_REHASH aggr_buck_rehash +#define AGGR_BUCK_PARAMS /8, *2, 2, 2, 4, 24 + +HASH_DEFINE_REHASH_FN(AGGR_BUCK, struct aggregator_bucket); + + +#define AGGR_DATA_MEMSIZE (sizeof(struct f_val) * p->aggr_on_count) + +static void +aggregator_rt_notify(struct proto *P, struct channel *src_ch, net *net, rte *new, rte *old) +{ + struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P); + ASSERT_DIE(src_ch == p->src); + struct aggregator_bucket *new_bucket = NULL, *old_bucket = NULL; + struct aggregator_route *old_route = NULL; + + /* Find the objects for the old route */ + if (old) + old_route = HASH_FIND(p->routes, AGGR_RTE, old); + + if (old_route) + old_bucket = old_route->bucket; + + /* Find the bucket for the new route */ + if (new) + { + /* Routes are identical, do nothing */ + if (old_route && rte_same(&old_route->rte, new)) + return; + + /* Evaluate route attributes. */ + struct aggregator_bucket *tmp_bucket = sl_allocz(p->bucket_slab); + + for (uint val_idx = 0; val_idx < p->aggr_on_count; val_idx++) + { + int type = p->aggr_on[val_idx].type; + + switch (type) + { + case AGGR_ITEM_TERM: { + const struct f_line *line = p->aggr_on[val_idx].line; + struct rte *rt1 = new; + enum filter_return fret = f_eval_rte(line, &new, rte_update_pool, 0, NULL, &tmp_bucket->aggr_data[val_idx]); + + if (rt1 != new) + { + rte_free(rt1); + log(L_WARN "Aggregator rule modifies the route, reverting"); + } + + if (fret > F_RETURN) + log(L_WARN "%s.%s: Wrong number of items left on stack after evaluation of aggregation list", rt1->src->proto->name, rt1->sender); + + break; + } + + case AGGR_ITEM_STATIC_ATTR: { + struct f_val *pos = &tmp_bucket->aggr_data[val_idx]; + eval_static_attr(new, p->aggr_on[val_idx].sa, pos); + break; + } + + case AGGR_ITEM_DYNAMIC_ATTR: { + struct f_val *pos = &tmp_bucket->aggr_data[val_idx]; + eval_dynamic_attr(new, p->aggr_on[val_idx].da, pos); + break; + } + + default: + break; + } + } + + /* Compute the hash */ + tmp_bucket->hash = mem_hash(tmp_bucket->aggr_data, AGGR_DATA_MEMSIZE); + + /* Find the existing bucket */ + if (new_bucket = HASH_FIND(p->buckets, AGGR_BUCK, tmp_bucket)) + sl_free(tmp_bucket); + else + { + new_bucket = tmp_bucket; + HASH_INSERT2(p->buckets, AGGR_BUCK, p->p.pool, new_bucket); + } + + /* Store the route attributes */ + if (rta_is_cached(new->attrs)) + rta_clone(new->attrs); + else + new->attrs = rta_lookup(new->attrs); + + /* Insert the new route into the bucket */ + struct aggregator_route *arte = sl_alloc(p->route_slab); + *arte = (struct aggregator_route) { + .bucket = new_bucket, + .rte = *new, + }; + arte->rte.next = new_bucket->rte, + new_bucket->rte = &arte->rte; + new_bucket->count++; + HASH_INSERT2(p->routes, AGGR_RTE, p->p.pool, arte); + } + + /* Remove the old route from its bucket */ + if (old_bucket) + { + for (struct rte **k = &old_bucket->rte; *k; k = &(*k)->next) + if (*k == &old_route->rte) + { + *k = (*k)->next; + break; + } + + old_bucket->count--; + HASH_REMOVE2(p->routes, AGGR_RTE, p->p.pool, old_route); + rta_free(old_route->rte.attrs); + sl_free(old_route); + } + + /* Announce changes */ + if (old_bucket) + aggregator_bucket_update(p, old_bucket, net); + + if (new_bucket && (new_bucket != old_bucket)) + aggregator_bucket_update(p, new_bucket, net); + + /* Cleanup the old bucket if empty */ + if (old_bucket && (!old_bucket->rte || !old_bucket->count)) + { + ASSERT_DIE(!old_bucket->rte && !old_bucket->count); + HASH_REMOVE2(p->buckets, AGGR_BUCK, p->p.pool, old_bucket); + sl_free(old_bucket); + } +} + +static int +aggregator_preexport(struct channel *C, struct rte *new) +{ + struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, C->proto); + /* Reject our own routes */ + if (new->sender == p->dst) + return -1; + + /* Disallow aggregating already aggregated routes */ + if (new->attrs->source == RTS_AGGREGATED) + { + log(L_ERR "Multiple aggregations of the same route not supported in BIRD 2."); + return -1; + } + + return 0; +} + +static void +aggregator_postconfig(struct proto_config *CF) +{ + struct aggregator_config *cf = SKIP_BACK(struct aggregator_config, c, CF); + + if (!cf->dst->table) + cf_error("Source table not specified"); + + if (!cf->src->table) + cf_error("Destination table not specified"); + + if (cf->dst->table->addr_type != cf->src->table->addr_type) + cf_error("Both tables must be of the same type"); + + cf->dst->in_filter = cf->src->in_filter; + + cf->src->in_filter = FILTER_REJECT; + cf->dst->out_filter = FILTER_REJECT; + + cf->dst->debug = cf->src->debug; +} + +static struct proto * +aggregator_init(struct proto_config *CF) +{ + struct proto *P = proto_new(CF); + struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P); + struct aggregator_config *cf = SKIP_BACK(struct aggregator_config, c, CF); + + proto_configure_channel(P, &p->src, cf->src); + proto_configure_channel(P, &p->dst, cf->dst); + + p->aggr_on_count = cf->aggr_on_count; + p->aggr_on_da_count = cf->aggr_on_da_count; + p->aggr_on = cf->aggr_on; + p->merge_by = cf->merge_by; + + P->rt_notify = aggregator_rt_notify; + P->preexport = aggregator_preexport; + + return P; +} + +static int +aggregator_start(struct proto *P) +{ + struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P); + + p->bucket_slab = sl_new(P->pool, sizeof(struct aggregator_bucket) + AGGR_DATA_MEMSIZE); + HASH_INIT(p->buckets, P->pool, AGGR_BUCK_ORDER); + + p->route_slab = sl_new(P->pool, sizeof(struct aggregator_route)); + HASH_INIT(p->routes, P->pool, AGGR_RTE_ORDER); + + p->reload_buckets = (event) { + .hook = aggregator_reload_buckets, + .data = p, + }; + + return PS_UP; +} + +static int +aggregator_shutdown(struct proto *P) +{ + struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P); + + HASH_WALK_DELSAFE(p->buckets, next_hash, b) + { + while (b->rte) + { + struct aggregator_route *arte = SKIP_BACK(struct aggregator_route, rte, b->rte); + b->rte = arte->rte.next; + b->count--; + HASH_REMOVE(p->routes, AGGR_RTE, arte); + rta_free(arte->rte.attrs); + sl_free(arte); + } + + ASSERT_DIE(b->count == 0); + HASH_REMOVE(p->buckets, AGGR_BUCK, b); + sl_free(b); + } + HASH_WALK_END; + + return PS_DOWN; +} + +static int +aggregator_reconfigure(struct proto *P, struct proto_config *CF) +{ + struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P); + struct aggregator_config *cf = SKIP_BACK(struct aggregator_config, c, CF); + + TRACE(D_EVENTS, "Reconfiguring"); + + /* Compare numeric values (shortcut) */ + if (cf->aggr_on_count != p->aggr_on_count) + return 0; + + if (cf->aggr_on_da_count != p->aggr_on_da_count) + return 0; + + /* Compare aggregator rule */ + for (uint i = 0; i < p->aggr_on_count; i++) + switch (cf->aggr_on[i].type) + { + case AGGR_ITEM_TERM: + if (!f_same(cf->aggr_on[i].line, p->aggr_on[i].line)) + return 0; + break; + case AGGR_ITEM_STATIC_ATTR: + if (memcmp(&cf->aggr_on[i].sa, &p->aggr_on[i].sa, sizeof(struct f_static_attr)) != 0) + return 0; + break; + case AGGR_ITEM_DYNAMIC_ATTR: + if (memcmp(&cf->aggr_on[i].da, &p->aggr_on[i].da, sizeof(struct f_dynamic_attr)) != 0) + return 0; + break; + default: + bug("Broken aggregator rule"); + } + + /* Compare merge filter */ + if (!f_same(cf->merge_by, p->merge_by)) + ev_schedule(&p->reload_buckets); + + p->aggr_on = cf->aggr_on; + p->merge_by = cf->merge_by; + + return 1; +} + +struct protocol proto_aggregator = { + .name = "Aggregator", + .template = "aggregator%d", + .class = PROTOCOL_AGGREGATOR, + .preference = 1, + .channel_mask = NB_ANY, + .proto_size = sizeof(struct aggregator_proto), + .config_size = sizeof(struct aggregator_config), + .postconfig = aggregator_postconfig, + .init = aggregator_init, + .start = aggregator_start, + .shutdown = aggregator_shutdown, + .reconfigure = aggregator_reconfigure, +}; + +void +aggregator_build(void) +{ + proto_build(&proto_aggregator); +} diff --git a/proto/aggregator/aggregator.h b/proto/aggregator/aggregator.h new file mode 100644 index 00000000..19459b1d --- /dev/null +++ b/proto/aggregator/aggregator.h @@ -0,0 +1,86 @@ +/* + * BIRD -- Aggregator Pseudoprotocol + * + * (c) 2023 Igor Putovny <igor.putovny@nic.cz> + * (c) 2023 Maria Matejka <mq@ucw.cz> + * (c) 2023 CZ.NIC z.s.p.o. + * + * Can be freely distributed and used under the terms of the GNU GPL. + * + * This file contains the data structures used by Babel. + */ + +#ifndef _BIRD_AGGREGATOR_H_ +#define _BIRD_AGGREGATOR_H_ + +#include "nest/bird.h" +#include "nest/protocol.h" +#include "lib/hash.h" + +struct aggregator_config { + struct proto_config c; + struct channel_config *src, *dst; + uint aggr_on_count; + uint aggr_on_da_count; + struct aggr_item *aggr_on; + const struct f_line *merge_by; +}; + +struct aggregator_route { + struct aggregator_route *next_hash; + struct aggregator_bucket *bucket; + struct rte rte; +}; + +struct aggregator_bucket { + struct aggregator_bucket *next_hash; + struct rte *rte; /* Pointer to struct aggregator_route.rte */ + struct rte_src *last_src; /* Which src we announced the bucket last with */ + u32 count; + u32 hash; + struct f_val aggr_data[0]; +}; + +struct aggregator_proto { + struct proto p; + struct channel *src, *dst; + + /* Buckets by aggregator rule */ + HASH(struct aggregator_bucket) buckets; + slab *bucket_slab; + + /* Routes by net and src */ + HASH(struct aggregator_route) routes; + slab *route_slab; + + /* Aggregator rule */ + uint aggr_on_count; + uint aggr_on_da_count; + struct aggr_item *aggr_on; + + /* Merge filter */ + const struct f_line *merge_by; + event reload_buckets; +}; + +enum aggr_item_type { + AGGR_ITEM_TERM, + AGGR_ITEM_STATIC_ATTR, + AGGR_ITEM_DYNAMIC_ATTR, +}; + +struct aggr_item { + enum aggr_item_type type; + union { + struct f_static_attr sa; + struct f_dynamic_attr da; + const struct f_line *line; + }; +}; + +struct aggr_item_node { + const struct aggr_item_node *next; + struct aggr_item i; +}; + +#endif diff --git a/proto/aggregator/config.Y b/proto/aggregator/config.Y new file mode 100644 index 00000000..44b7752f --- /dev/null +++ b/proto/aggregator/config.Y @@ -0,0 +1,134 @@ +/* + * BIRD -- Aggregator configuration + * + * (c) 2023 Igor Putovny <igor.putovny@nic.cz> + * (c) 2023 Maria Matejka <mq@ucw.cz> + * (c) 2023 CZ.NIC z.s.p.o. + * + * Can be freely distributed and used under the terms of the GNU GPL. + */ + +CF_HDR + +#include "proto/aggregator/aggregator.h" + +CF_DEFINES + +#define AGGREGATOR_CFG ((struct aggregator_config *) this_proto) +#define AGGR_ITEM_ALLOC ((struct aggr_item_node *) cfg_allocz(sizeof(struct aggr_item_node))) + + +CF_DECLS + +CF_KEYWORDS(AGGREGATOR, AGGREGATE, ON, MERGE, BY) + +%type <ai> aggr_item aggr_list + +CF_GRAMMAR + +proto: aggregator_proto ; + +aggregator_proto_start: proto_start AGGREGATOR +{ + this_proto = proto_config_new(&proto_aggregator, $1); + this_channel = AGGREGATOR_CFG->src = channel_config_new(NULL, "source", 0, this_proto); + AGGREGATOR_CFG->dst = channel_config_new(NULL, "destination", 0, this_proto); + + AGGREGATOR_CFG->src->ra_mode = AGGREGATOR_CFG->dst->ra_mode = RA_ANY; +}; + +aggregator_proto_item: + proto_item + | channel_item_ + | PEER TABLE rtable { AGGREGATOR_CFG->dst->table = $3; } + | AGGREGATE ON aggr_list { + if (AGGREGATOR_CFG->aggr_on) + cf_error("Only one aggregate on clause allowed"); + + _Bool net_present = 0; + int count = 0; + + for (const struct aggr_item_node *item = $3; item; item = item->next) { +// log(L_WARN "type %d sacode %d", item->i.type, item->i.sa.sa_code); + if (item->i.type == AGGR_ITEM_STATIC_ATTR && item->i.sa.sa_code == SA_NET) + net_present = 1; + + count++; + } + + if (!net_present) + cf_error("'NET' must be present"); + + AGGREGATOR_CFG->aggr_on = cfg_alloc(sizeof(struct aggr_item) * count); + + int pos = 0; + for (const struct aggr_item_node *item = $3; item; item = item->next) { + if (item->i.type == AGGR_ITEM_DYNAMIC_ATTR) + AGGREGATOR_CFG->aggr_on_da_count++; + + AGGREGATOR_CFG->aggr_on[pos++] = item->i; + } + + AGGREGATOR_CFG->aggr_on_count = pos; + } + | MERGE BY { + cf_push_block_scope(new_config); + cf_create_symbol(new_config, "routes", SYM_VARIABLE | T_ROUTES_BLOCK, offset, f_new_var(sym_->scope)); + } function_body { + cf_pop_block_scope(new_config); + $4->args++; + AGGREGATOR_CFG->merge_by = $4; + } +; + +aggregator_proto_opts: /* empty */ | aggregator_proto_opts aggregator_proto_item ';' ; +aggregator_proto: aggregator_proto_start proto_name '{' aggregator_proto_opts '}' ; + + +aggr_list: + aggr_item + | aggr_list ',' aggr_item { + if ($3 == NULL) { + $$ = $1; + } else { + $$ = $3; + $$->next = $1; + } + } + ; + +aggr_item: + '(' term ')' { + $$ = AGGR_ITEM_ALLOC; + $$->i.type = AGGR_ITEM_TERM; + $$->i.line = f_linearize($2, 1); + } + | CF_SYM_KNOWN { + switch ($1->class) { + case SYM_ATTRIBUTE: + $$ = AGGR_ITEM_ALLOC; + $$->i.type = AGGR_ITEM_DYNAMIC_ATTR; + $$->i.da = *$1->attribute; + break; + case SYM_CONSTANT_RANGE: + $$ = NULL; + break; + default: + cf_error("Can't aggregate on symbol type %s.", cf_symbol_class_name($1)); + } + } + | dynamic_attr { + $$ = AGGR_ITEM_ALLOC; + $$->i.type = AGGR_ITEM_DYNAMIC_ATTR; + $$->i.da = $1; + } + | static_attr { + $$ = AGGR_ITEM_ALLOC; + $$->i.type = AGGR_ITEM_STATIC_ATTR; + $$->i.sa = $1; + } + ; + +CF_CODE + +CF_END diff --git a/proto/aggregator/test.conf b/proto/aggregator/test.conf new file mode 100644 index 00000000..e5e1e267 --- /dev/null +++ b/proto/aggregator/test.conf @@ -0,0 +1,116 @@ +log "bird.log" all; + +protocol device {} + +protocol static { + ipv6; + route 2001:db8:0::/48 unreachable { bgp_path.prepend(65432); bgp_path.prepend(4200000000); }; + route 2001:db8:1::/48 unreachable; + route 2001:db8:2::/48 unreachable; + route 2001:db8:3::/48 unreachable; + route 2001:db8:4::/48 unreachable; + route 2001:db8:5::/48 unreachable; + route 2001:db8:6::/48 unreachable; + route 2001:db8:7::/48 unreachable; + route 2001:db8:8::/48 unreachable; + route 2001:db8:9::/48 unreachable; + route 2001:db8:a::/48 unreachable; + route 2001:db8:b::/48 unreachable; + route 2001:db8:c::/48 unreachable; + route 2001:db8:d::/48 unreachable; + route 2001:db8:e::/48 unreachable; + route 2001:db8:f::/48 unreachable; +} + +protocol static { + ipv6 { + import filter { + bgp_med = 1; + bgp_community = -empty-.add((65533,1)).add((65500,0xe)); + accept; + }; + }; + route 2001:db8:1::/48 unreachable; + route 2001:db8:3::/48 unreachable; + route 2001:db8:5::/48 unreachable; + route 2001:db8:7::/48 unreachable; + route 2001:db8:9::/48 unreachable; + route 2001:db8:b::/48 unreachable; + route 2001:db8:d::/48 unreachable; + route 2001:db8:f::/48 unreachable; +} + +protocol static { + ipv6 { + import filter { + bgp_med = 2; + bgp_community = -empty-.add((65533,2)).add((65500,0xd)); + accept; + }; + }; + route 2001:db8:2::/48 unreachable; + route 2001:db8:3::/48 unreachable; + route 2001:db8:6::/48 unreachable; + route 2001:db8:7::/48 unreachable; + route 2001:db8:a::/48 unreachable; + route 2001:db8:b::/48 unreachable; + route 2001:db8:e::/48 unreachable; + route 2001:db8:f::/48 unreachable; +} + +protocol static { + ipv6 { + import filter { + bgp_med = 4; + bgp_community = -empty-.add((65533,4)).add((65500,0xb)); + accept; + }; + }; + route 2001:db8:4::/48 unreachable; + route 2001:db8:5::/48 unreachable; + route 2001:db8:6::/48 unreachable; + route 2001:db8:7::/48 unreachable; + route 2001:db8:c::/48 unreachable; + route 2001:db8:d::/48 unreachable; + route 2001:db8:e::/48 unreachable; + route 2001:db8:f::/48 unreachable; +} + +protocol static { + ipv6 { + import filter { + bgp_med = 8; + bgp_community = -empty-.add((65533,8)).add((65500,0x7)); + accept; + }; + }; + route 2001:db8:8::/48 unreachable; + route 2001:db8:9::/48 unreachable; + route 2001:db8:a::/48 unreachable; + route 2001:db8:b::/48 unreachable; + route 2001:db8:c::/48 unreachable; + route 2001:db8:d::/48 unreachable; + route 2001:db8:e::/48 unreachable; + route 2001:db8:f::/48 unreachable; +} + +ipv6 table agr_result; + +protocol aggregator { + table master6; + peer table agr_result; + export all; + aggregate on net,(defined(bgp_med)); + merge by { + print "Merging all these: ", routes; + bgp_med = 0; + for route r in routes do { + if ! defined(r.bgp_med) then { unset(bgp_med); accept; } + + print r, " bgp_med: ", r.bgp_med; + bgp_med = bgp_med + r.bgp_med; + bgp_community = bgp_community.add(r.bgp_community); + } + accept; + }; +} |