5d7ca7ee697f22f714711733c4ac1fb3393116bb
[distributedratelimiting.git] / drl / ulogd_DRL.c
1 /* See the DRL-LICENSE file for this file's software license. */
2
3 /*
4  * ulogd output target for DRL: GRD and FPS
5  *
6  * Ken Yocum <kyocum@cs.ucsd.edu>
7  *
8  * Original shell of this code from ulogd_NETFLOW
9  * Like that code, we keep track of per-slice data rates 
10  * out of this slice.  Thus we are rate limiting particular slices
11  * across multiple boxes, ensuring that their outbound rate does not
12  * exceed some fixed limit.  
13  *
14  * Fabric:  static mesh
15  *
16  * Enforcer: linux drop percentage.
17  *    
18  * This file reads packets from the netlink socket.  It updates all
19  * the hashmaps which track how much data has arrived per flow.
20  * It starts two threads for this limiter.
21  * One thread handles periodic estimation.
22  * The other thread handles communication with other limiters. 
23  *
24  * 
25  * Code map
26  *
27  * ulogd_DRL: attach to netlink socket, accept packets.  replaces ratelimit.cc
28  * util.c: generic hashing functions, flow comparisons, sundry items. 
29  * gossip.c: Recv gossip, send gossip.
30  * peer_comm.c: Thread to listen for updates from other limiters. 
31  * estimate.c: Thread to calculate the local limits. 
32  * 
33  * 
34  * Ken Yocum <kyocum@cs.ucsd.edu>
35  * 2007 
36  * 
37  * Some code appropriated from ulogd_NETFLOW:
38  *
39  * Mark Huang <mlhuang@cs.princeton.edu>
40  * Copyright (C) 2004-2005 The Trustees of Princeton University
41  *
42  * Based on admindump.pl by Mic Bowman and Paul Brett
43  * Copyright (c) 2002 Intel Corporation
44  *
45  */
46
47 /* Enable GNU glibc extensions */
48 #define _GNU_SOURCE
49
50 #include <stdio.h>
51 #include <stdlib.h>
52
53 /* va_start() and friends */
54 #include <stdarg.h>
55
56 /* ispunct() */
57 #include <ctype.h>
58
59 /* strstr() and friends */
60 #include <string.h>
61
62 /* dirname() and basename() */
63 #include <libgen.h>
64
65 /* fork() and wait() */
66 #include <sys/types.h>
67 #include <unistd.h>
68 #include <sys/wait.h>
69
70 /* errno and assert() */
71 #include <errno.h>
72 #include <assert.h>
73
74 /* getopt_long() */
75 #include <getopt.h>
76
77 /* time() and friends */
78 #include <time.h>
79 #include <sys/time.h>
80
81 /* inet_aton() */
82 #include <sys/socket.h>
83 #include <netinet/in.h>
84 #include <arpa/inet.h>
85
86 /* ICMP definitions */
87 #include <netinet/ip.h>
88 #include <netinet/ip_icmp.h>
89
90 /* stat() */
91 #include <sys/stat.h>
92
93 /* pthread_create() */
94 #include <pthread.h>
95
96 /* flock() */
97 #include <sys/file.h>
98
99 /* Signal definitions - so that we can catch SIGHUP and update config. */
100 #include <signal.h>
101
102 #include <ulogd/ulogd.h>
103 #include <ulogd/conffile.h>
104
105 /* Perhaps useful for files within vservers? */
106 #if !defined(STANDALONE) && HAVE_LIBPROPER
107 #include <proper/prop.h>
108 #endif
109
110 /*
111  * Jenkins hash support
112  * lives in raterouter.h
113  */
114
115 /* DRL specifics */
116 #include "raterouter.h"
117 #include "util.h"
118 #include "calendar.h"
119 #include "ratetypes.h" /* needs util and pthread.h */
120 #include "logging.h"
121
122 /*
123  * /etc/ulogd.conf configuration options
124  * Add the config options for DRL. 
125  */
126
127 static config_entry_t drl_configfile = {
128     .next = NULL,
129     .key = "drl_configfile",
130     .type = CONFIG_TYPE_STRING,
131     .options = CONFIG_OPT_MANDATORY,
132     .u = { .string = "drl.xml" },
133 };
134
135 /** The administrative bandwidth limit (mbps) for the local node.  The node
136  * will not set a limit higher than this, even when distributed capacity is
137  * available.  Set to 0 for no limit. */
138 static config_entry_t nodelimit = {
139     .next = &drl_configfile,
140     .key = "nodelimit",
141     .type = CONFIG_TYPE_INT,
142     .options = CONFIG_OPT_MANDATORY,
143     .u = { .value = 0 },
144 };
145
146 /** Determines the verbosity of logging. */
147 static config_entry_t drl_loglevel = {
148     .next = &nodelimit,
149     .key = "drl_loglevel",
150     .type = CONFIG_TYPE_INT,
151     .options = CONFIG_OPT_MANDATORY,
152     .u = { .value = LOG_WARN },
153 };
154
155 /** The path of the logfile. */
156 static config_entry_t drl_logfile = {
157     .next = &drl_loglevel,
158     .key = "drl_logfile",
159     .type = CONFIG_TYPE_STRING,
160     .options = CONFIG_OPT_MANDATORY,
161     .u = { .string = "drl_logfile.log" },
162 };
163
164 /** The choice of DRL protocol. */
165 static config_entry_t policy = {
166     .next = &drl_logfile,
167     .key = "policy",
168     .type = CONFIG_TYPE_STRING,
169     .options = CONFIG_OPT_MANDATORY,
170     .u = { .string = "GRD" },
171 };
172
173 /** The estimate interval, in milliseconds. */
174 static config_entry_t estintms = {
175     .next = &policy,
176     .key = "estintms",
177     .type = CONFIG_TYPE_INT,
178     .options = CONFIG_OPT_MANDATORY,
179     .u = { .value = 100 },
180 };
181
182 #define config_entries (&estintms)
183
184 /*
185  * Debug functionality
186  */
187
188 #ifdef DMALLOC
189 #include <dmalloc.h>
190 #endif
191
192 #define NIPQUAD(addr) \
193     ((unsigned char *)&addr)[0], \
194     ((unsigned char *)&addr)[1], \
195     ((unsigned char *)&addr)[2], \
196     ((unsigned char *)&addr)[3]
197
198 #define IPQUAD(addr) \
199     ((unsigned char *)&addr)[3], \
200     ((unsigned char *)&addr)[2], \
201     ((unsigned char *)&addr)[1], \
202     ((unsigned char *)&addr)[0]
203
204
205
206 /* Salt for the hash functions */
207 static int salt;
208
209 /*
210  * Hash slice name lookups on context ID.
211  */
212
213 /* Special context IDs */
214 #define UNKNOWN_XID -1
215 #define ROOT_XID 0
216
217 enum {
218     CONNECTION_REFUSED_XID = 65536, /* MAX_S_CONTEXT + 1 */
219     ICMP_ECHOREPLY_XID,
220     ICMP_UNREACH_XID,
221 };
222
223
224 /* globals */
225 pthread_t estimate_thread;
226 pthread_t signal_thread;
227 pthread_t comm_thread;
228 uint32_t local_ip = 0;
229 limiter_t limiter;
230 extern FILE *logfile;
231 extern uint8_t system_loglevel;
232
233 /* functions */
234
235 static inline uint32_t
236 hash_flow(uint8_t protocol, uint32_t src_ip, uint16_t src_port, uint32_t dst_ip, uint16_t dst_port)
237 {
238     unsigned char mybytes[FLOWKEYSIZE];
239     mybytes[0] = protocol;
240     *(uint32_t*)(&(mybytes[1])) = src_ip;
241     *(uint32_t*)(&(mybytes[5])) = dst_ip;
242     *(uint32_t*)(&(mybytes[9])) = (src_port << 16) | dst_port;
243     return jhash(mybytes,FLOWKEYSIZE,salt) & (FLOW_HASH_SIZE - 1);
244 }
245
246 uint32_t sampled_hasher(const key_flow *key) {
247     return hash_flow(key->protocol, key->source_ip, key->source_port, key->dest_ip, key->dest_port);
248 }
249
250 uint32_t standard_hasher(const key_flow *key) {
251     return hash_flow(key->protocol, key->source_ip, key->source_port, key->dest_ip, key->dest_port);
252 }
253
254 struct intr_id {
255     char* name;
256     ulog_iret_t *res;
257 };
258
259 /* Interesting keys */
260 enum {
261     OOB_TIME_SEC = 0,
262     OOB_MARK,
263     IP_SADDR,
264     IP_DADDR,
265     IP_TOTLEN,
266     IP_PROTOCOL,
267     TCP_SPORT,
268     TCP_DPORT,
269     TCP_ACK,
270     TCP_FIN,
271     TCP_SYN,
272     TCP_RST,
273     UDP_SPORT,
274     UDP_DPORT,
275     ICMP_TYPE,
276     ICMP_CODE,
277     GRE_FLAG_KEY,
278     GRE_VERSION,
279     GRE_KEY,
280     PPTP_CALLID,
281 };
282
283 #define INTR_IDS (sizeof(intr_ids)/sizeof(intr_ids[0]))
284 static struct intr_id intr_ids[] = {
285     [OOB_TIME_SEC] = { "oob.time.sec", 0 },
286     [OOB_MARK] = { "oob.mark", 0 },
287     [IP_SADDR] = { "ip.saddr", 0 },
288     [IP_DADDR] = { "ip.daddr", 0 },
289     [IP_TOTLEN] = { "ip.totlen", 0 },
290     [IP_PROTOCOL] = { "ip.protocol", 0 },
291     [TCP_SPORT] = { "tcp.sport", 0 },
292     [TCP_DPORT] { "tcp.dport", 0 },
293     [TCP_ACK] = { "tcp.ack", 0 },
294     [TCP_FIN] = { "tcp.fin", 0 },
295     [TCP_SYN] = { "tcp.syn", 0 },
296     [TCP_RST] = { "tcp.rst", 0 },
297     [UDP_SPORT] = { "udp.sport", 0 },
298     [UDP_DPORT] = { "udp.dport", 0 },
299     [ICMP_TYPE] = { "icmp.type", 0 },
300     [ICMP_CODE] = { "icmp.code", 0 },
301     [GRE_FLAG_KEY] = { "gre.flag.key", 0 },
302     [GRE_VERSION] = { "gre.version", 0 },
303     [GRE_KEY] = { "gre.key", 0 },
304     [PPTP_CALLID] = { "pptp.callid", 0 },
305 };
306
307 #define GET_VALUE(x) intr_ids[x].res->value
308
309 #define DATE(t) ((t) / (24*60*60) * (24*60*60))
310
311 static int _output_drl(ulog_iret_t *res)
312 {
313     int xid;
314     uint32_t src_ip, dst_ip;
315     uint16_t src_port, dst_port;
316     uint8_t protocol;
317
318     key_flow key;
319     identity_t *ident;
320     leaf_t *leaf;
321
322     protocol = GET_VALUE(IP_PROTOCOL).ui8;
323     src_ip = GET_VALUE(IP_SADDR).ui32;
324     dst_ip = GET_VALUE(IP_DADDR).ui32;
325     xid = GET_VALUE(OOB_MARK).ui32;
326
327     switch (protocol) {
328             
329         case IPPROTO_TCP:
330             src_port = GET_VALUE(TCP_SPORT).ui16;
331             dst_port = GET_VALUE(TCP_DPORT).ui16;
332             break;
333
334         case IPPROTO_UDP:
335             /* netflow had an issue with many udp flows and set
336              * src_port=0 to handle it.  We don't. 
337              */
338             src_port = GET_VALUE(UDP_SPORT).ui16;
339
340             /*
341              * traceroutes create a large number of flows in the db
342              * this is a quick hack to catch the most common form
343              * of traceroute (basically we're mapping any UDP packet
344              * in the 33435-33524 range to the "trace" port, 33524 is
345              * 3 packets * nhops (30).
346              */
347             dst_port = GET_VALUE(UDP_DPORT).ui16;
348             if (dst_port >= 33435 && dst_port <= 33524)
349                 dst_port = 33435;
350             break;
351
352         case IPPROTO_ICMP:
353             src_port = GET_VALUE(ICMP_TYPE).ui8;
354             dst_port = GET_VALUE(ICMP_CODE).ui8;
355
356             /*
357              * We special case some of the ICMP traffic that the kernel
358              * always generates. Since this is attributed to root, it 
359              * creates significant "noise" in the output. We want to be
360              * able to quickly see that root is generating traffic.
361              */
362             if (xid == ROOT_XID) {
363                 if (src_port == ICMP_ECHOREPLY)
364                     xid = ICMP_ECHOREPLY_XID;
365                 else if (src_port == ICMP_UNREACH)
366                     xid = ICMP_UNREACH_XID;
367             }
368             break;
369
370         case IPPROTO_GRE:
371             if (GET_VALUE(GRE_FLAG_KEY).b) {
372                 if (GET_VALUE(GRE_VERSION).ui8 == 1) {
373                     /* Get PPTP call ID */
374                     src_port = GET_VALUE(PPTP_CALLID).ui16;
375                 } else {
376                     /* XXX Truncate GRE keys to 16 bits */
377                     src_port = (uint16_t) GET_VALUE(GRE_KEY).ui32;
378                 }
379             } else {
380                 /* No key available */
381                 src_port = 0;
382             }
383             dst_port = 0;
384             break;
385
386         default:
387             /* This is the default key for packets from unsupported protocols */
388             src_port = 0;
389             dst_port = 0;
390             break;
391     }
392
393     key.protocol = protocol;
394     key.source_ip = src_ip;
395     key.dest_ip = dst_ip;
396     key.source_port = src_port;
397     key.dest_port = dst_port;
398     key.packet_size = GET_VALUE(IP_TOTLEN).ui16;
399     key.packet_time = (time_t) GET_VALUE(OOB_TIME_SEC).ui32;
400
401     pthread_rwlock_rdlock(&limiter.limiter_lock); /* CLUNK! */
402
403     leaf = (leaf_t *) map_search(limiter.stable_instance.leaf_map, &xid, sizeof(xid));
404
405     /* Even if the packet doesn't match any specific xid, it should still
406      * count in the machine-type tables.  This catches root (xid == 0) and
407      * unclassified (xid = fff) packets, which don't have map entries. */
408     if (leaf == NULL) {
409         ident = limiter.stable_instance.last_machine;
410     } else {
411         ident = leaf->parent;
412     }
413
414     while (ident) {
415         pthread_mutex_lock(&ident->table_mutex);
416
417         /* Update the identity's table. */
418         ident->table_sample_function(ident->table, &key);
419
420         pthread_mutex_unlock(&ident->table_mutex);
421
422         ident = ident->parent;
423     }
424
425     pthread_rwlock_unlock(&limiter.limiter_lock); /* CLINK! */
426
427     return 0;
428 }
429
430 /* get all key id's for the keys we are intrested in */
431 static int get_ids(void)
432 {
433     int i;
434     struct intr_id *cur_id;
435
436     for (i = 0; i < INTR_IDS; i++) {
437         cur_id = &intr_ids[i];
438         cur_id->res = keyh_getres(keyh_getid(cur_id->name));
439         if (!cur_id->res) {
440             ulogd_log(ULOGD_ERROR, 
441                     "Cannot resolve keyhash id for %s\n", 
442                     cur_id->name);
443             return 1;
444         }
445     }
446     return 0;
447 }
448
449 static void free_identity(identity_t *ident) {
450     if (ident) {
451         free_comm(&ident->comm);
452
453         if (ident->table) {
454             ident->table_destroy_function(ident->table);
455         }
456
457         pthread_mutex_destroy(&ident->table_mutex);
458
459         free(ident);
460     }
461 }
462
463 static void free_identity_map(map_handle map) {
464     identity_t *tofree = NULL;
465
466     map_reset_iterate(map);
467     while ((tofree = (identity_t *) map_next(map))) {
468         free_identity(tofree);
469     }
470
471     free_map(map, 0);
472 }
473
474 static void free_instance(drl_instance_t *instance) {
475     if (instance->leaves)
476         free(instance->leaves);
477     if (instance->leaf_map)
478         free_map(instance->leaf_map, 0);
479     if (instance->ident_map)
480         free_identity_map(instance->ident_map);
481     if (instance->machines)
482         free(instance->machines);
483     if (instance->sets)
484         free(instance->sets);
485     if (instance->cal) {
486         free(instance->cal);
487     }
488
489     memset(instance, 0, sizeof(drl_instance_t));
490 }
491
492 static void free_failed_config(parsed_configs configs, drl_instance_t *instance) {
493     /* Free configs. */
494     if (configs.machines)
495         free_ident_list(configs.machines);
496     if (configs.sets)
497         free_ident_list(configs.sets);
498
499     /* Free instance. */
500     if (instance)
501         free_instance(instance);
502 }
503
504 static identity_t *new_identity(ident_config *config) {
505     identity_t *ident = malloc(sizeof(identity_t));
506     remote_node_t *comm_nodes = malloc(sizeof(remote_node_t)*config->peer_count);
507     ident_peer *peer = config->peers;
508     int peer_slot = 0;
509
510     if (ident == NULL) {
511         return NULL;
512     }
513
514     if (comm_nodes == NULL) {
515         free(ident);
516         return NULL;
517     }
518
519     memset(ident, 0, sizeof(identity_t));
520     memset(comm_nodes, 0, config->peer_count * sizeof(remote_node_t));
521
522     ident->id = config->id;
523     ident->limit = (uint32_t) (((double) config->limit * 1000000.0) / 8.0);
524     ident->fixed_ewma_weight = config->fixed_ewma_weight;
525     ident->intervals = config->intervals;
526     ident->ewma_weight = pow(ident->fixed_ewma_weight, 
527                              (limiter.estintms/1000.0) * config->intervals);
528     ident->parent = NULL;
529
530     pthread_mutex_init(&ident->table_mutex, NULL);
531     switch (config->accounting) {
532         case ACT_STANDARD:
533             ident->table =
534                 standard_table_create(standard_hasher, &ident->common);
535
536             /* Ugly function pointer casting.  Makes things sufficiently
537              * generic, though. */
538             ident->table_sample_function =
539                 (int (*)(void *, const key_flow *)) standard_table_sample;
540             ident->table_cleanup_function =
541                 (int (*)(void *)) standard_table_cleanup;
542             ident->table_update_function =
543                 (void (*)(void *, struct timeval, double)) standard_table_update_flows;
544             ident->table_destroy_function =
545                 (void (*)(void *)) standard_table_destroy;
546             break;
547
548         case ACT_SAMPLEHOLD:
549             ident->table = sampled_table_create(sampled_hasher,
550                     ident->limit * IDENT_CLEAN_INTERVAL,
551                     1, 20, &ident->common);
552
553             ident->table_sample_function =
554                 (int (*)(void *, const key_flow *)) sampled_table_sample;
555             ident->table_cleanup_function =
556                 (int (*)(void *)) sampled_table_cleanup;
557             ident->table_update_function =
558                 (void (*)(void *, struct timeval, double)) sampled_table_update_flows;
559             ident->table_destroy_function =
560                 (void (*)(void *)) sampled_table_destroy;
561             break;
562
563         case ACT_SIMPLE:
564             ident->table = simple_table_create(&ident->common);
565
566             ident->table_sample_function =
567                 (int (*)(void *, const key_flow *)) simple_table_sample;
568             ident->table_cleanup_function =
569                 (int (*)(void *)) simple_table_cleanup;
570             ident->table_update_function =
571                 (void (*)(void *, struct timeval, double)) simple_table_update_flows;
572             ident->table_destroy_function =
573                 (void (*)(void *)) simple_table_destroy;
574             break;
575     }
576
577     /* Make sure the table was allocated. */
578     if (ident->table == NULL) {
579         free(ident);
580         return NULL;
581     }
582
583     while (peer) {
584         comm_nodes[peer_slot].addr = peer->ip;
585         comm_nodes[peer_slot].port = htons(LIMITER_LISTEN_PORT);
586         peer = peer->next;
587         peer_slot += 1;
588     }
589
590     if (new_comm(&ident->comm, config, comm_nodes)) {
591         printlog(LOG_CRITICAL, "Failed to create communication structure.\n");
592         return NULL;
593     }
594
595     ident->comm.remote_nodes = comm_nodes;
596
597     return ident;
598 }
599
600 /* Determines the validity of the parameters of one ident_config.
601  *
602  * 0 valid
603  * 1 invalid
604  */
605 static int validate_config(ident_config *config) {
606     /* Limit must be a positive integer. */
607     if (config->limit < 1) {
608         return 1;
609     }
610
611     /* Commfabric must be a valid choice (COMM_MESH or COMM_GOSSIP). */
612     if (config->commfabric != COMM_MESH &&
613             config->commfabric != COMM_GOSSIP) {
614         return 1;
615     }
616
617     /* If commfabric is COMM_GOSSIP, this must be a positive integer. */
618     if (config->commfabric == COMM_GOSSIP && config->branch < 1) {
619         return 1;
620     }
621
622     /* Accounting must be a valid choice (ACT_STANDARD, ACT_SAMPLEHOLD,
623      * ACT_SIMPLE). */
624     if (config->accounting != ACT_STANDARD &&
625             config->accounting != ACT_SAMPLEHOLD &&
626             config->accounting != ACT_SIMPLE) {
627         return 1;
628     }
629
630     /* Ewma weight must be greater than or equal to zero. */
631     if (config->fixed_ewma_weight < 0) {
632         return 1;
633     }
634
635     /* Note: Parsing stage requires that each ident has at least one peer. */
636     return 0;
637 }
638
639 /* 0 success
640  * non-zero failure
641  */
642 static int validate_configs(parsed_configs configs, drl_instance_t *instance) {
643
644     ident_config *mlist = configs.machines;
645     ident_config *slist = configs.sets;
646     ident_config *tmp = NULL;
647     int i = 0;
648
649     while (mlist) {
650         /* ID must be non-zero and unique. */
651         /* This is ugly and hackish, but this function will be called rarely.
652          * I'm tired of trying to be clever. */
653         if (mlist->id < 0) {
654             printlog(LOG_CRITICAL, "Negative ident id: %d (%x) ?\n", mlist->id, mlist->id);
655             return EINVAL;
656         }
657         tmp = mlist->next;
658         while (tmp) {
659             if (mlist->id == tmp->id) {
660                 printlog(LOG_CRITICAL, "Duplicate ident id: %d (%x)\n", mlist->id, mlist->id);
661                 return EINVAL;
662             }
663             tmp = tmp->next;
664         }
665         tmp = configs.sets;
666         while (tmp) {
667             if (mlist->id == tmp->id) {
668                 printlog(LOG_CRITICAL, "Duplicate ident id: %d (%x)\n", mlist->id, mlist->id);
669                 return EINVAL;
670             }
671             tmp = tmp->next;
672         }
673
674         if (validate_config(mlist)) {
675             printlog(LOG_CRITICAL, "Invalid ident parameters for id: %d (%x)\n", mlist->id, mlist->id);
676             return EINVAL;
677         }
678
679         mlist = mlist->next;
680     }
681
682     instance->sets = malloc(configs.set_count * sizeof(identity_t *));
683     if (instance->sets == NULL) {
684         return ENOMEM;
685     }
686
687     memset(instance->sets, 0, configs.set_count * sizeof(identity_t *));
688     instance->set_count = configs.set_count;
689
690     /* For sets, make sure that the hierarchy is valid. */
691     while (slist) {
692         ident_member *members = slist->members;
693
694         /* ID must be non-zero and unique. */
695         if (slist->id < 0) {
696             printlog(LOG_CRITICAL, "Negative ident id: %d (%x) ?\n", slist->id, slist->id);
697             return EINVAL;
698         }
699         tmp = slist->next;
700         while (tmp) {
701             if (slist->id == tmp->id) {
702                 printlog(LOG_CRITICAL, "Duplicate ident id: %d (%x)\n", slist->id, slist->id);
703                 return EINVAL;
704             }
705             tmp = tmp->next;
706         }
707
708         if (validate_config(slist)) {
709             printlog(LOG_CRITICAL, "Invalid ident parameters for id: %d (%x)\n", slist->id, slist->id);
710             return EINVAL;
711         }
712
713         /* Allocate an identity_t for this set-type identity. */
714         instance->sets[i] = new_identity(slist);
715
716         if (instance->sets[i] == NULL) {
717             return ENOMEM;
718         }
719
720         /* Loop through children and look up each in leaf or ident map
721          * depending on the type of child.  Set the child's parent pointer
722          * to the identity we just created above, unless it is already set,
723          * in which case we have an error. */
724         while (members) {
725             identity_t *child_ident = NULL;
726             leaf_t *child_leaf = NULL;
727
728             switch (members->type) {
729                 case MEMBER_XID:
730                     child_leaf = map_search(instance->leaf_map, &members->value,
731                                             sizeof(members->value));
732                     if (child_leaf == NULL) {
733                         return EINVAL;
734                     }
735                     if (child_leaf->parent != NULL) {
736                         /* Error - This leaf already has a parent. */
737                         return EINVAL;
738                     }
739                     child_leaf->parent = instance->sets[i];
740                     break;
741                 case MEMBER_GUID:
742                     child_ident = map_search(instance->ident_map, &members->value,
743                                              sizeof(members->value));
744                     if (child_ident == NULL) {
745                         return EINVAL;
746                     }
747                     if (child_ident->parent != NULL) {
748                         /* Error - This identity already has a parent. */
749                         return EINVAL;
750                     }
751                     child_ident->parent = instance->sets[i];
752                     break;
753                 default:
754                     /* Error - shouldn't be possible. */
755                     break;
756             }
757             members = members->next;
758         }
759
760         map_insert(instance->ident_map, &instance->sets[i]->id,
761                    sizeof(instance->sets[i]->id), instance->sets[i]);
762
763         slist = slist->next;
764         i++;
765     }
766     return 0;
767 }
768
769 static int init_identities(parsed_configs configs, drl_instance_t *instance) {
770     int i;
771     ident_config *config = configs.machines;
772     leaf_t *leaf = NULL;
773
774     instance->cal = malloc(sizeof(struct ident_calendar) * SCHEDLEN);
775
776     if (instance->cal == NULL) {
777         return ENOMEM;
778     }
779
780     for (i = 0; i < SCHEDLEN; ++i) {
781         TAILQ_INIT(instance->cal + i);
782     }
783     instance->cal_slot = 0;
784
785     instance->machines = malloc(configs.machine_count * sizeof(drl_instance_t *));
786
787     if (instance->machines == NULL) {
788         return ENOMEM;
789     }
790
791     memset(instance->machines, 0, configs.machine_count * sizeof(drl_instance_t *));
792     instance->machine_count = configs.machine_count;
793
794     /* Allocate and add the machine identities. */
795     for (i = 0; i < configs.machine_count; ++i) {
796         instance->machines[i] = new_identity(config);
797
798         if (instance->machines[i] == NULL) {
799             return ENOMEM;
800         }
801
802         /* The first has no parent - it is the root.  All others have the
803          * previous ident as their parent. */
804         if (i == 0) {
805             instance->machines[i]->parent = NULL;
806         } else {
807             instance->machines[i]->parent = instance->machines[i - 1];
808         }
809
810         instance->last_machine = instance->machines[i];
811
812         /* Add the ident to the guid->ident map. */
813         map_insert(instance->ident_map, &instance->machines[i]->id,
814                    sizeof(instance->machines[i]->id), instance->machines[i]);
815
816         config = config->next;
817
818         TAILQ_INSERT_TAIL(instance->cal + (instance->cal_slot & SCHEDMASK),
819                           instance->machines[i], calendar);
820     }
821
822     /* Connect the set subtree to the machines. Any set or leaf without a
823      * parent will take the last machine as its parent. */
824
825     /* Leaves... */
826     map_reset_iterate(instance->leaf_map);
827     while ((leaf = (leaf_t *) map_next(instance->leaf_map))) {
828         if (leaf->parent == NULL) {
829             leaf->parent = instance->last_machine;
830         }
831     }
832
833     /* Sets... */
834     for (i = 0; i < instance->set_count; ++i) {
835         if (instance->sets[i]->parent == NULL) {
836             instance->sets[i]->parent = instance->last_machine;
837         }
838
839         TAILQ_INSERT_TAIL(instance->cal + (instance->cal_slot & SCHEDMASK),
840                           instance->sets[i], calendar);
841     }
842
843     /* Success. */
844     return 0;
845 }
846
847 static void print_instance(drl_instance_t *instance) {
848     leaf_t *leaf = NULL;
849     identity_t *ident = NULL;
850
851     map_reset_iterate(instance->leaf_map);
852     while ((leaf = (leaf_t *) map_next(instance->leaf_map))) {
853         printf("%x:", leaf->xid);
854         ident = leaf->parent;
855         while (ident) {
856             printf("%d:",ident->id);
857             ident = ident->parent;
858         }
859         printf("Leaf's parent pointer is %p\n", leaf->parent);
860     }
861
862     printf("instance->last_machine is %p\n", instance->last_machine);
863 }
864
865 static int assign_htb_hierarchy(drl_instance_t *instance) {
866     int i, j;
867     int next_node = 0x11;
868
869     /* Chain machine nodes under 1:10. */
870     for (i = 0; i < instance->machine_count; ++i) {
871         if (instance->machines[i]->parent == NULL) {
872             /* Top node. */
873             instance->machines[i]->htb_parent = 0x10;
874         } else {
875             /* Pointerific! */
876             instance->machines[i]->htb_parent =
877                         instance->machines[i]->parent->htb_node;
878         }
879
880         instance->machines[i]->htb_node = next_node;
881         next_node += 1;
882     }
883
884     next_node += 0x10;
885
886     /* Add set nodes under machine nodes. Iterate backwards to ensure parent is
887      * already there. */
888     for (j = (instance->set_count - 1); j >= 0; --j) {
889         if (instance->sets[j]->parent == NULL) {
890             instance->sets[j]->htb_parent = 0x10;
891         } else {
892             instance->sets[j]->htb_parent = instance->sets[j]->parent->htb_node;
893         }
894         instance->sets[j]->htb_node = next_node;
895
896         next_node += 1;
897     }
898
899     return 0;
900 }
901
902 /* Added this so that I could comment one line and kill off all of the
903  * command execution. */
904 static int execute_cmd(const char *cmd) {
905     return system(cmd);
906 }
907
908 static int create_htb_hierarchy(drl_instance_t *instance) {
909     char cmd[300];
910     int i, j, k;
911
912     /* Nuke the hierarchy. */
913     sprintf(cmd, "tc qdisc del dev eth0 root handle 1: htb");
914     execute_cmd(cmd);
915     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
916
917     /* Re-initialize the basics. */
918     sprintf(cmd, "tc qdisc add dev eth0 root handle 1: htb default 1fff");
919     if (execute_cmd(cmd)) {
920         return 1;
921     }
922     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
923     sprintf(cmd, "tc class add dev eth0 parent 1: classid 1:1 htb rate 1000mbit ceil 1000mbit");
924     if (execute_cmd(cmd)) {
925         return 1;
926     }
927     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
928
929     /* Add back 1:10. (Nodelimit : Megabits/sec -> bits/second)*/
930     if (limiter.nodelimit) {
931         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:1 classid 1:10 htb rate 8bit ceil %lubit",
932                 (unsigned long) limiter.nodelimit * 1024 * 1024);
933     } else {
934         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:1 classid 1:10 htb rate 8bit ceil 1000mbit");
935     }
936
937     if (execute_cmd(cmd)) {
938         return 1;
939     }
940     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
941
942     /* Add back 1:20. */
943     sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:1 classid 1:20 htb rate 8bit ceil 1000mbit");
944
945     if (execute_cmd(cmd)) {
946         return 1;
947     }
948     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
949
950
951     /* Add machines. */
952     for (i = 0; i < instance->machine_count; ++i) {
953         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:%x classid 1:%x htb rate 8bit ceil %lubit",
954                 instance->machines[i]->htb_parent,
955                 instance->machines[i]->htb_node,
956                 (unsigned long) instance->machines[i]->limit * 1024 * 1024);
957
958         if (execute_cmd(cmd)) {
959             return 1;
960         }
961         printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
962     }
963
964     /* Add sets. */
965     for (j = (instance->set_count - 1); j >= 0; --j) {
966         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:%x classid 1:%x htb rate 8bit ceil %lubit",
967                 instance->sets[j]->htb_parent,
968                 instance->sets[j]->htb_node,
969                 (unsigned long) instance->sets[j]->limit * 1024 * 1024);
970
971         if (execute_cmd(cmd)) {
972             return 1;
973         }
974         printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
975     }
976
977     /* Add leaves. FIXME: Set static sliver limit as ceil here! */
978     for (k = 0; k < instance->leaf_count; ++k) {
979         if (instance->leaves[k].parent == NULL) {
980             sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:10 classid 1:1%x htb rate 8bit ceil %lubit",
981                 instance->leaves[k].xid,
982                 (unsigned long) 100 * 1024 * 1024);
983         } else {
984             sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:%x classid 1:1%x htb rate 8bit ceil %lubit",
985                     instance->leaves[k].parent->htb_node,
986                     instance->leaves[k].xid,
987                     (unsigned long) 100 * 1024 * 1024);
988         }
989
990         if (execute_cmd(cmd)) {
991             return 1;
992         }
993         printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
994         
995         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:20 classid 1:2%x htb rate 8bit ceil 1000mbit",
996                 instance->leaves[k].xid);
997
998         if (execute_cmd(cmd)) {
999             return 1;
1000         }
1001         printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1002     }
1003
1004     /* Add 1:1000 and 1:2000 */
1005     if (instance->last_machine == NULL) {
1006         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:10 classid 1:1000 htb rate 8bit ceil 1000mbit");
1007     } else {
1008         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:%x classid 1:1000 htb rate 8bit ceil 1000mbit",
1009                 instance->last_machine->htb_node);
1010     }
1011
1012     if (execute_cmd(cmd)) {
1013         return 1;
1014     }
1015     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1016     sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:20 classid 1:2000 htb rate 8bit ceil 1000mbit");
1017
1018     if (execute_cmd(cmd)) {
1019         return 1;
1020     }
1021     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1022
1023     /* Add 1:1fff and 1:2fff */
1024     if (instance->last_machine == NULL) {
1025         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:10 classid 1:1fff htb rate 8bit ceil 1000mbit");
1026     } else {
1027         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:%x classid 1:1fff htb rate 8bit ceil 1000mbit",
1028                 instance->last_machine->htb_node);
1029     }
1030
1031     if (execute_cmd(cmd)) {
1032         return 1;
1033     }
1034     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1035     sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:20 classid 1:2fff htb rate 8bit ceil 1000mbit");
1036
1037     if (execute_cmd(cmd)) {
1038         return 1;
1039     }
1040     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1041
1042     return 0;
1043 }
1044
1045 /* init_drl 
1046  * 
1047  * Initialize this limiter with options 
1048  * Open UDP socket for peer communication
1049  */
1050 static int init_drl(void) {
1051     parsed_configs configs;
1052     struct sockaddr_in server_address;
1053     
1054     memset(&limiter, 0, sizeof(limiter_t));
1055
1056     /* Setup logging. */
1057     system_loglevel = (uint8_t) drl_loglevel.u.value;
1058     logfile = fopen(drl_logfile.u.string, "w");
1059
1060     if (logfile == NULL) {
1061         printf("Couldn't open logfile - ");
1062         perror("fopen()");
1063         exit(EXIT_FAILURE);
1064     }
1065
1066     printlog(LOG_CRITICAL, "ulogd_DRL initializing . . .\n");
1067
1068     limiter.nodelimit = (uint32_t) (((double) nodelimit.u.value * 1000000.0) / 8.0);
1069
1070     init_hashing();  /* for all hash maps */
1071
1072     pthread_rwlock_init(&limiter.limiter_lock,NULL);
1073
1074     /* determine our local IP by iterating through interfaces */
1075     if ((limiter.ip = get_local_ip())==0) {
1076         printlog(LOG_CRITICAL,
1077                  "ulogd_DRL unable to aquire local IP address, not registering.\n");
1078         return (false);
1079     }
1080     limiter.localaddr = inet_addr(limiter.ip);
1081     limiter.port = htons(LIMITER_LISTEN_PORT);
1082     limiter.udp_socket = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
1083     if (limiter.udp_socket < 0) {
1084         printlog(LOG_CRITICAL, "Failed to create UDP socket().\n");
1085         return false;
1086     }
1087
1088     memset(&server_address, 0, sizeof(server_address));
1089     server_address.sin_family = AF_INET;
1090     server_address.sin_addr.s_addr = limiter.localaddr;
1091     server_address.sin_port = limiter.port;
1092
1093     if (bind(limiter.udp_socket, (struct sockaddr *) &server_address, sizeof(server_address)) < 0) {
1094         printlog(LOG_CRITICAL, "Failed to bind UDP socket.\n");
1095         return false;
1096     }
1097
1098     printlog(LOG_WARN, "     POLICY: %s\n",policy.u.string);
1099     if (strcasecmp(policy.u.string,"GRD") == 0) {
1100         limiter.policynum = POLICY_GRD;
1101     } else if (strcasecmp(policy.u.string,"FPS") == 0) {
1102         limiter.policynum = POLICY_FPS;
1103     } else {
1104         printlog(LOG_CRITICAL,
1105                  "Unknown DRL policy %s, aborting.\n",policy.u.string);
1106         return (false);
1107     }
1108
1109     limiter.estintms = estintms.u.value;
1110     if (limiter.estintms > 1000) {
1111         printlog(LOG_CRITICAL,
1112                  "DRL: sorry estimate intervals must be less than 1 second.");
1113         printlog(LOG_CRITICAL,
1114                  "  Simple source mods will allow larger intervals.  Using 1 second.\n");
1115         limiter.estintms = 1000;
1116     }
1117     printlog(LOG_WARN, "     Est interval: %dms\n",limiter.estintms);
1118     
1119     /* Acquire the big limiter lock for writing.  Prevents pretty much
1120      * anything else from happening while the hierarchy is being changed. */
1121     pthread_rwlock_wrlock(&limiter.limiter_lock);
1122
1123     limiter.stable_instance.ident_map = allocate_map();
1124     if (limiter.stable_instance.ident_map == NULL) {
1125         printlog(LOG_CRITICAL, "Failed to allocate memory for identity map.\n");
1126         return false;
1127     }
1128
1129     if (get_eligible_leaves(&limiter.stable_instance)) {
1130         printlog(LOG_CRITICAL, "Failed to read eligigle leaves.\n");
1131         return false;
1132     }
1133
1134     if (parse_drl_config(drl_configfile.u.string, &configs)) {
1135         /* Parse error occured. Return non-zero to notify init_drl(). */
1136         return false;
1137     }
1138
1139     /* Validate identity hierarchy! */
1140     if (validate_configs(configs, &limiter.stable_instance)) {
1141         /* Clean up everything. */
1142         free_failed_config(configs, &limiter.stable_instance);
1143         return false;
1144     }
1145
1146     if (init_identities(configs, &limiter.stable_instance)) {
1147         free_failed_config(configs, &limiter.stable_instance);
1148         return false;
1149     }
1150
1151     /* At this point, we should be done with configs. */
1152     free_ident_list(configs.machines);
1153     free_ident_list(configs.sets);
1154
1155     /* Debugging - FIXME: remove this? */
1156     print_instance(&limiter.stable_instance);
1157
1158     if (assign_htb_hierarchy(&limiter.stable_instance)) {
1159         free_instance(&limiter.stable_instance);
1160         return false;
1161     }
1162
1163     if (create_htb_hierarchy(&limiter.stable_instance)) {
1164         free_instance(&limiter.stable_instance);
1165         return false;
1166     }
1167
1168     pthread_rwlock_unlock(&limiter.limiter_lock);
1169
1170     if (pthread_create(&limiter.udp_recv_thread, NULL, limiter_receive_thread, NULL)) {
1171         printlog(LOG_CRITICAL, "Unable to start UDP receive thread.\n");
1172         return false;
1173     }
1174
1175     printlog(LOG_WARN, "ulogd_DRL init finished.\n");
1176
1177     return true;
1178 }
1179
1180 static void reconfig() {
1181     parsed_configs configs;
1182
1183     printlog(LOG_DEBUG, "--Starting reconfig()--\n");
1184     flushlog();
1185
1186     memset(&configs, 0, sizeof(parsed_configs));
1187     memset(&limiter.new_instance, 0, sizeof(drl_instance_t));
1188
1189     limiter.new_instance.ident_map = allocate_map();
1190     if (limiter.new_instance.ident_map == NULL) {
1191         printlog(LOG_CRITICAL, "Failed to allocate ident_map during reconfig().\n");
1192         return;
1193     }
1194
1195     if (get_eligible_leaves(&limiter.new_instance)) {
1196         free_failed_config(configs, &limiter.new_instance);
1197         printlog(LOG_CRITICAL, "Failed to read leaves during reconfig().\n");
1198         return;
1199     }
1200
1201     if (parse_drl_config(drl_configfile.u.string, &configs)) {
1202         free_failed_config(configs, &limiter.new_instance);
1203         printlog(LOG_CRITICAL, "Failed to parse config during reconfig().\n");
1204         return;
1205     }
1206
1207     /* Lock */
1208     pthread_rwlock_wrlock(&limiter.limiter_lock);
1209
1210     if (validate_configs(configs, &limiter.new_instance)) {
1211         free_failed_config(configs, &limiter.new_instance);
1212         printlog(LOG_CRITICAL, "Validation failed during reconfig().\n");
1213         pthread_rwlock_unlock(&limiter.limiter_lock);
1214         return;
1215     }
1216
1217     if (init_identities(configs, &limiter.new_instance)) {
1218         free_failed_config(configs, &limiter.new_instance);
1219         printlog(LOG_CRITICAL, "Initialization failed during reconfig().\n");
1220         pthread_rwlock_unlock(&limiter.limiter_lock);
1221         return;
1222     }
1223
1224     free_ident_list(configs.machines);
1225     free_ident_list(configs.sets);
1226
1227     /* Debugging - FIXME: remove this? */
1228     print_instance(&limiter.new_instance);
1229
1230     if (assign_htb_hierarchy(&limiter.new_instance)) {
1231         free_instance(&limiter.new_instance);
1232         printlog(LOG_CRITICAL, "Failed to assign HTB hierarchy during reconfig().\n");
1233         pthread_rwlock_unlock(&limiter.limiter_lock);
1234         return;
1235     }
1236
1237     if (create_htb_hierarchy(&limiter.new_instance)) {
1238         free_instance(&limiter.new_instance);
1239         printlog(LOG_CRITICAL, "Failed to create HTB hierarchy during reconfig().\n");
1240
1241         /* Re-create old instance. */
1242         if (create_htb_hierarchy(&limiter.stable_instance)) {
1243             /* Error reinstating the old one - big problem. */
1244             printlog(LOG_CRITICAL, "Failed to reinstate HTB hierarchy during reconfig().\n");
1245             flushlog();
1246             exit(EXIT_FAILURE);
1247         }
1248
1249         pthread_rwlock_unlock(&limiter.limiter_lock);
1250         return;
1251     }
1252
1253     /* Switch over new to stable instance. */
1254     free_instance(&limiter.stable_instance);
1255     memcpy(&limiter.stable_instance, &limiter.new_instance, sizeof(drl_instance_t));
1256
1257     /* Success! - Unlock */
1258     pthread_rwlock_unlock(&limiter.limiter_lock);
1259 }
1260
1261 static ulog_output_t drl_op = {
1262     .name = "drl",
1263     .output = &_output_drl,
1264     .signal = NULL, /* This appears to be broken. Using my own handler. */
1265     .init = NULL,
1266     .fini = NULL,
1267 };
1268
1269 /* Tests the amount of time it takes to call reconfig(). */
1270 static void time_reconfig(int iterations) {
1271     struct timeval start, end;
1272     int i;
1273
1274     gettimeofday(&start, NULL);
1275     for (i = 0; i < iterations; ++i) {
1276         reconfig();
1277     }
1278     gettimeofday(&end, NULL);
1279
1280     printf("%d reconfigs() took %d seconds and %d microseconds.\n",
1281            iterations, end.tv_sec - start.tv_sec, end.tv_usec - start.tv_usec);
1282     exit(0);
1283
1284     // Seems to take about 85ms / iteration
1285 }
1286
1287 static void *signal_thread_func(void *args) {
1288     int sig;
1289     int err;
1290     sigset_t sigs;
1291
1292     sigemptyset(&sigs);
1293     sigaddset(&sigs, SIGHUP);
1294     pthread_sigmask(SIG_BLOCK, &sigs, NULL);
1295
1296     while (1) {
1297         sigemptyset(&sigs);
1298         sigaddset(&sigs, SIGHUP);
1299
1300         err = sigwait(&sigs, &sig);
1301
1302         if (err) {
1303             printlog(LOG_CRITICAL, "sigwait() returned an error.\n");
1304             flushlog();
1305         }
1306
1307         switch (sig) {
1308             case SIGHUP:
1309                 printlog(LOG_WARN, "Caught SIGHUP - re-reading XML file.\n");
1310                 reconfig();
1311                 //time_reconfig(1000); //instrumentation
1312                 flushlog();
1313                 break;
1314             default:
1315                 /* Should be impossible... */
1316                 break;
1317         }
1318     }
1319
1320 }
1321
1322 /* register output plugin with ulogd */
1323 static void _drl_reg_op(void)
1324 {
1325     ulog_output_t *op = &drl_op;
1326     sigset_t signal_mask;
1327
1328     sigemptyset(&signal_mask);
1329     sigaddset(&signal_mask, SIGHUP);
1330     pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
1331
1332     if (pthread_create(&signal_thread, NULL, &signal_thread_func, NULL) != 0) {
1333         printlog(LOG_CRITICAL, "Failed to create signal handling thread.\n");
1334         fprintf(stderr, "An error has occured starting ulogd_DRL.  Refer to your logfile (%s) for additional information.\n", drl_logfile.u.string);
1335         flushlog();
1336         exit(EXIT_FAILURE);
1337     }
1338
1339     if (!init_drl()) {
1340         printlog(LOG_CRITICAL, "Init failed. :(\n");
1341         fprintf(stderr, "An error has occured starting ulogd_DRL.  Refer to your logfile (%s) for additional information.\n", drl_logfile.u.string);
1342         flushlog();
1343         exit(EXIT_FAILURE);
1344     }
1345
1346     register_output(op);
1347
1348     /* start up the thread that will periodically estimate the
1349      * local rate and set the local limits
1350      * see estimate.c
1351      */
1352     if (pthread_create(&estimate_thread, NULL, (void*(*)(void*)) &handle_estimation, &limiter)!=0) {
1353         ulogd_log(ULOGD_ERROR, "couldn't start estimate thread for 0x%x %s\n",limiter.localaddr,
1354                 limiter.ip);
1355         fprintf(stderr, "An error has occured starting ulogd_DRL.  Refer to your logfile (%s) for additional information.\n", drl_logfile.u.string);
1356         exit(EXIT_FAILURE);
1357     }
1358 }
1359
1360 void _init(void)
1361 {
1362     /* have the opts parsed */
1363     config_parse_file("DRL", config_entries);
1364
1365     if (get_ids()) {
1366         ulogd_log(ULOGD_ERROR, "can't resolve all keyhash id's\n");
1367         exit(2);
1368     }
1369
1370     /* Seed the hash function */
1371     salt = getpid() ^ time(NULL);
1372
1373     _drl_reg_op();
1374 }