<commit_before>/*
 * Copyright 2013-2017 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <ctype.h>
#include <glob.h>
#include <regex.h>
#include <sys/stat.h>
#include <netdb.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <assert.h>

#include "fnv1a.h"
#include "consistent-hash.h"
#include "server.h"
#include "queue.h"
#include "aggregator.h"
#include "allocator.h"
#include "relay.h"
#include "router.h"
#include "conffile.h"
#include "conffile.tab.h"

struct _router {
	cluster *clusters;
	route *routes;
	aggregator *aggregators;
	servers *srvrs;
	struct _router_collector {
		char *stub;
		int interval;
		char *prefix;
		col_mode mode;
	} collector;
	listener *listeners;
	struct _router_parser_err {
		const char *msg;
		size_t line;
		size_t start;
		size_t stop;
	} parser_err;
	struct _router_conf {
		char workercnt;
		size_t queuesize;
		size_t batchsize;
		int maxstalls;
		unsigned short iotimeout;
		unsigned int sockbufsize;
	} conf;
	allocator *a;
};

/* declarations to avoid symbol/include hell when doing config.yy.h */
int router_yylex_init(void **);
void router_yy_scan_string(char *, void *);
int router_yylex_destroy(void *);

/* catch string used for aggrgator destinations */
#define STUB_AGGR   "_aggregator_stub_"
/* catch string used for collector output */
#define STUB_STATS  "_collector_stub_"

/**
 * Frees regexes from routes.
 */
static void
router_free_intern(route *routes, char workercnt)
{
	/* the only thing that isn't allocated using the allocators are the
	 * regexes */
	while (routes != NULL) {
		if (routes->matchtype == REGEX) {
			int i;
			for(i = 0; i < workercnt; i++)
				regfree(&routes->rule[i]);
		}

		if (routes->next == NULL || routes->next->dests != routes->dests) {
			while (routes->dests != NULL) {
				if (routes->dests->cl->type == GROUP ||
						routes->dests->cl->type == AGGRSTUB ||
						routes->dests->cl->type == STATSTUB)
					router_free_intern(routes->dests->cl->members.routes,
							workercnt);
				if (routes->dests->cl->type == VALIDATION)
					router_free_intern(
							routes->dests->cl->members.validation->rule,
							workercnt);

				routes->dests = routes->dests->next;
			}
		}

		routes = routes->next;
	}
}

/**
 * Examines pattern and sets matchtype and rule or strmatch in route.
 */
static inline int
determine_if_regex(allocator *a, char workercnt, route *r, char *pat)
{
	/* try and see if we can avoid using a regex match, for
	 * it is simply very slow/expensive to do so: most of
	 * the time, people don't need fancy matching rules */
	char patbuf[8192];
	char *e = pat;
	char *pb = patbuf;
	char escape = 0;
	r->matchtype = CONTAINS;
	r->nmatch = 0;

	if (*e == '^' && r->matchtype == CONTAINS) {
		e++;
		r->matchtype = STARTS_WITH;
	}
	for (; *e != '\0'; e++) {
		switch (*e) {
			case '\\':
				if (escape)
					*pb++ = *e;
				escape = !escape;
				break;
			case '.':
			case '^':
			case '*':
			case '+':
				if (!escape)
					r->matchtype = REGEX;
				*pb++ = *e;
				escape = 0;
				break;
			case '$':
				if (!escape && e[1] == '\0') {
					if (r->matchtype == STARTS_WITH) {
						r->matchtype = MATCHES;
					} else {
						r->matchtype = ENDS_WITH;
					}
				} else {
					r->matchtype = REGEX;
				}
				escape = 0;
				break;
			default:
				if (
						!escape && (
							(*e == '_') || (*e == '-') ||
							(*e >= '0' && *e <= '9') ||
							(*e >= 'a' && *e <= 'z') ||
							(*e >= 'A' && *e <= 'Z')
						)
				   )
				{
					*pb++ = *e;
				} else {
					r->matchtype = REGEX;
				}
				escape = 0;
				break;
		}
		if (pb - patbuf == sizeof(patbuf))
			r->matchtype = REGEX;
		if (r->matchtype == REGEX)
			break;
	}
	if (r->matchtype != REGEX) {
		*pb = '\0';
		r->strmatch = ra_strdup(a, patbuf);
		r->pattern = ra_strdup(a, pat);
	} else {
		int i;
		int ret;

		r->rule = ra_malloc(a, sizeof(*r->rule) * workercnt);
		if (r->rule == NULL) {
			logerr("determine_if_regex: malloc failed for "
					"regular expressions\n");
			return REG_ESPACE;  /* lie closest to the truth */
		}
		ret = regcomp(&r->rule[0], pat, REG_EXTENDED);
		if (ret != 0)
			return ret;  /* allow use of regerror */
		if (r->rule[0].re_nsub > 0) {
			/* we need +1 because position 0 contains the entire
			 * expression */
			r->nmatch = r->rule[0].re_nsub + 1;
			if (r->nmatch > RE_MAX_MATCHES) {
				logerr("determine_if_regex: too many match groups, "
						"please increase RE_MAX_MATCHES in router.h\n");
				regfree(&r->rule[0]);
				return REG_ESPACE;  /* lie closest to the truth */
			}
		}
		for (i = 1; i < workercnt; i++) {
			if (regcomp(&r->rule[i], pat, REG_EXTENDED) != 0) {
				while (--i >= 0)
					regfree(&r->rule[i]);
				logerr("determine_if_regex: out of memory allocating "
						"regexes\n");
				return REG_ESPACE;  /* lie closest to the truth */
			}
		}
		r->strmatch = NULL;
		r->pattern = ra_strdup(a, pat);
	}

	return 0;
}

static char serverip_server[256];  /* RFC1035 = 250 */
static const char *
serverip(server *s)
{
	const char *srvr = server_ip(s);

	if (strchr(srvr, ':') != NULL) {
		snprintf(serverip_server, sizeof(serverip_server), "[%s]", srvr);
		return serverip_server;
	}

	return srvr;
}


/**
 * Callback for the bison parser for yyerror
 */
void
router_yyerror(
		void *locptr,
		void *s,
		router *r,
		allocator *ra,
		allocator *a,
		const char *msg)
{
	(void)s;
	ROUTER_YYLTYPE *locp = (ROUTER_YYLTYPE *)locptr;
	regex_t re;
	size_t nmatch = 3;
	regmatch_t pmatch[3];
	char buf1[METRIC_BUFSIZ];
	char buf2[METRIC_BUFSIZ];
	char *dummy;
	char (*sa)[METRIC_BUFSIZ];
	char (*sb)[METRIC_BUFSIZ];
	char (*st)[METRIC_BUFSIZ];

	if (r->parser_err.msg != NULL)
		return;

	/* clean up the "value" types */
	if (regcomp(&re, "cr(STRING|COMMENT|INTVAL)", REG_EXTENDED) != 0) {
		r->parser_err.msg = ra_strdup(a, msg);
		return;
	}
	snprintf(buf1, sizeof(buf1), "%s", msg);
	sa = &buf1;
	sb = &buf2;
	while (regexec(&re, *sa, nmatch, pmatch, 0) == 0) {
		dummy = *sa + strlen(*sa);
		if (router_rewrite_metric(sb, &dummy,
					*sa, dummy, "\\_1", nmatch, pmatch) == 0)
		{
			r->parser_err.msg = ra_strdup(a, msg);
			return;
		}
		st = sa;
		sa = sb;
		sb = st;
	}
	regfree(&re);

	/* clean up the keywords */
	if (regcomp(&re, "cr([A-Z]+)", REG_EXTENDED) != 0) {
		r->parser_err.msg = ra_strdup(a, msg);
		return;
	}
	while (regexec(&re, *sa, nmatch, pmatch, 0) == 0) {
		dummy = *sa + strlen(*sa);
		if (router_rewrite_metric(sb, &dummy,
					*sa, dummy, "'\\_1'", nmatch, pmatch) == 0)
		{
			r->parser_err.msg = ra_strdup(a, msg);
			return;
		}
		st = sa;
		sa = sb;
		sb = st;
	}
	regfree(&re);

	/* clean up bison speak */
	if (regcomp(&re, "\\$end", REG_EXTENDED) != 0) {
		r->parser_err.msg = ra_strdup(a, msg);
		return;
	}
	while (regexec(&re, *sa, nmatch, pmatch, 0) == 0) {
		dummy = *sa + strlen(*sa);
		if (router_rewrite_metric(sb, &dummy,
					*sa, dummy, "end of file", nmatch, pmatch) == 0)
		{
			r->parser_err.msg = ra_strdup(a, msg);
			return;
		}
		st = sa;
		sa = sb;
		sb = st;
	}
	regfree(&re);

	r->parser_err.msg = ra_strdup(a, *sa);
	if (locp != NULL) {
		r->parser_err.line = 1 + locp->first_line;
		r->parser_err.start = locp->first_column;
		if (locp->first_line == locp->last_line) {
			r->parser_err.stop = locp->last_column;
		} else {
			r->parser_err.stop = locp->first_column;
		}
	} else {
		r->parser_err.line = 0;
	}
}

/**
 * Parse ip string and validate it.  When it is a numeric address, write
 * the cannonical version in retip.
 */
char *
router_validate_address(
		router *rtr,
		char **retip, int *retport, void **retsaddr, void **rethint,
		char *ip, serv_ctype proto)
{
	int port = GRAPHITE_PORT;
	struct addrinfo *saddr = NULL;
	struct addrinfo hint;
	char sport[8];
	char *p;
	char *lastcolon = NULL;

	for (p = ip; *p != '\0'; p++)
		if (*p == ':')
			lastcolon = p;

	if (*(p - 1) == ']')
		lastcolon = NULL;
	if (lastcolon != NULL) {
		char *endp = NULL;
		*lastcolon = '\0';
		port = (int)strtol(lastcolon + 1, &endp, 10);
		if (port < 1 || endp != p)
			return(ra_strdup(rtr->a, "invalid port number"));
	}
	if (*ip == '[') {
		ip++;
		if (lastcolon != NULL && *(lastcolon - 1) == ']') {
			*(lastcolon - 1) = '\0';
		} else if (lastcolon == NULL && *(p - 1) == ']') {
			*(p - 1) = '\0';
		} else {
			return(ra_strdup(rtr->a, "expected ']'"));
		}
	}
	if (lastcolon == ip)
		ip = NULL;  /* only resolve port, e.g. any interface */

	memset(&hint, 0, sizeof(hint));
	saddr = NULL;

	/* try to see if this is a "numeric" IP address, which means
	 * re-resolving lateron will make no difference */
	hint.ai_family = PF_UNSPEC;
	hint.ai_socktype = proto == CON_UDP ? SOCK_DGRAM : SOCK_STREAM;
	hint.ai_protocol = proto == CON_UDP ? IPPROTO_UDP : IPPROTO_TCP;
	hint.ai_flags = AI_NUMERICHOST | AI_NUMERICSERV | AI_PASSIVE;
	snprintf(sport, sizeof(sport), "%u", port);

	*rethint = NULL;
	if (getaddrinfo(ip, sport, &hint, &saddr) != 0) {
		int err;
		/* now resolve this the normal way to check validity */
		hint.ai_flags = AI_NUMERICSERV;
		if ((err = getaddrinfo(ip, sport, &hint, &saddr)) != 0) {
			char errmsg[512];
			snprintf(errmsg, sizeof(errmsg), "warning: failed to resolve "
					"%s, port %s, proto %s: %s",
					ip == NULL ? "<any>" : ip, sport,
					proto == CON_UDP ? "udp" : "tcp",
					gai_strerror(err));
			saddr = NULL;
		}

		/* no ra_malloc, it gets owned by server */
		*rethint = malloc(sizeof(hint));
		if (*rethint == NULL) {
			freeaddrinfo(saddr);
			return(ra_strdup(rtr->a, "out of memory copying hint structure"));
		}
		memcpy(*rethint, &hint, sizeof(hint));
	}

	*retip = ip;
	*retport = port;
	*retsaddr = saddr;
	return NULL;
}

/**
 * Parse path string and validate it can be written to.
 */
char *
router_validate_path(router *rtr, char *path)
{
	struct stat st;
	char fileexists = 1;
	int probefd;

	/* if the file doesn't exist, remove it after trying to create it */
	if (stat(path, &st) == -1)
		fileexists = 0;

	if ((probefd = open(path,
					O_WRONLY | O_APPEND | O_CREAT,
					S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)) < 0)
	{
		char errbuf[512];
		if (errno == EOPNOTSUPP && st.st_mode & S_IFSOCK) {
			/* This is pretty lame, but I don't know a better way: we
			 * can't open sockets in use, and as it stands we can't
			 * check it is *us* who have the socket in use.  So, we
			 * assume we do, and ignore this silently.  In the worst
			 * case this yields an error when it opens up for real. */
			return NULL;
		}
		snprintf(errbuf, sizeof(errbuf),
				"failed to open file '%s' for writing: %s",
				path, strerror(errno));
		return ra_strdup(rtr->a, errbuf);
	}
	close(probefd);

	if (!fileexists)
		unlink(path);

	return NULL;
}

/**
 * Checks pat against regular expression syntax (or being '*') and
 * returns a route struct with the parsed info.  If retr is NULL a route
 * struct is allocated.
 */
char *
router_validate_expression(router *rtr, route **retr, char *pat)
{
	route *r = *retr;
	if (r == NULL) {
		r = *retr = ra_malloc(rtr->a, sizeof(route));
		if (r == NULL)
			return ra_strdup(rtr->a, "out of memory allocating route");
		r->next = NULL;
		r->dests = NULL;
	}
	if (strcmp(pat, "*") == 0) {
		r->pattern = NULL;
		r->strmatch = NULL;
		r->matchtype = MATCHALL;
	} else {
		int err = determine_if_regex(rtr->a, rtr->conf.workercnt, r, pat);
		if (err != 0) {
			char ebuf[512];
			size_t s = snprintf(ebuf, sizeof(ebuf),
					"invalid expression '%s': ", pat);
			regerror(err, &r->rule[0], ebuf + s, sizeof(ebuf) - s);
			return ra_strdup(rtr->a, ebuf);
		}
	}

	return NULL;
}

/**
 * Check if cluster is defined, and set the pointer to it.
 */
char *
router_validate_cluster(router *rtr, cluster **clptr, char *cl)
{
	cluster *w;

	/* lookup cluster */
	for (w = rtr->clusters; w != NULL; w = w->next) {
		if (w->type != GROUP &&
				w->type != AGGRSTUB &&
				w->type != STATSTUB &&
				w->type != AGGREGATION &&
				w->type != REWRITE &&
				w->type != VALIDATION &&
				strcmp(w->name, cl) == 0)
			break;
	}
	if (w == NULL)
		return ra_strdup(rtr->a, "unknown cluster");

	*clptr = w;
	return NULL;
}

/**
 * Adds a server to the chain in router, expands if necessary.
 */
char *
router_add_server(
		router *ret,
		char *ip,
		int port,
		char *inst,
		serv_ctype proto,
		struct addrinfo *saddrs,
		struct addrinfo *hint,
		char useall,
		cluster *cl)
{
	struct addrinfo *walk;
	struct addrinfo *lwalk;
	char hnbuf[INET6_ADDRSTRLEN];
	char errbuf[512];
	server *newserver;
	servers *w = NULL;

	walk = saddrs;  /* NULL if file */
	do {
		servers *s;
		lwalk = NULL;

		if (useall) {
			/* serialise the IP address, to make the targets explicit
			 * (since we're expanding all A/AAAA records) */
			saddr_ntop(walk, hnbuf);
			if (hnbuf[0] != '\0')
				ip = hnbuf;
			/* now we've expanded, don't trigger re-resolving */
			if (hint)
				free(hint);
			hint = NULL;
			/* and turn this into a separate entry */
			lwalk = walk->ai_next;
			walk->ai_next = NULL;
		}

		newserver = NULL;
		for (s = ret->srvrs; s != NULL; s = s->next) {
			if (server_cmp(s->server, walk, ip) == 0) {
				newserver = s->server;
				s->refcnt++;
				break;
			}
		}
		if (newserver == NULL) {
			newserver = server_new(ip, (unsigned short)port,
					proto, walk, hint,
					ret->conf.queuesize, ret->conf.batchsize,
					ret->conf.maxstalls, ret->conf.iotimeout,
					ret->conf.sockbufsize);
			if (newserver == NULL) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"failed to add server %s:%d (%s) "
						"to cluster %s", ip, port,
						proto == CON_UDP ? "udp" :
						proto == CON_TCP ? "tcp" :
						proto == CON_FILE ? "file" : "???", cl->name);
				return ra_strdup(ret->a, errbuf);
			}
			if (ret->srvrs == NULL) {
				s = ret->srvrs = ra_malloc(ret->a, sizeof(servers));
			} else {
				for (s = ret->srvrs; s->next != NULL; s = s->next)
					;
				s = s->next = ra_malloc(ret->a, sizeof(servers));
			}
			s->next = NULL;
			s->refcnt = 1;
			s->server = newserver;
		}
		
		/* check instance matches the existing server */
		if (s->refcnt > 1) {
			char *sinst = server_instance(newserver);
			if (sinst == NULL && inst != NULL) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"cannot set instance '%s' for "
						"server %s:%d: server was previously "
						"defined without instance",
						inst, serverip(newserver),
						server_port(newserver));
				return ra_strdup(ret->a, errbuf);
			} else if (sinst != NULL && inst == NULL) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"cannot define server %s:%d without "
						"instance: server was previously "
						"defined with instance '%s'",
						serverip(newserver),
						server_port(newserver), sinst);
				return ra_strdup(ret->a, errbuf);
			} else if (sinst != NULL && inst != NULL &&
					strcmp(sinst, inst) != 0)
			{
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"cannot set instance '%s' for "
						"server %s:%d: server was previously "
						"defined with instance '%s'",
						inst, serverip(newserver),
						server_port(newserver), sinst);
				return ra_strdup(ret->a, errbuf);
			} /* else: sinst == inst == NULL */
		}
		if (inst != NULL)
			server_set_instance(newserver, inst);

		if (cl->type == CARBON_CH ||
				cl->type == FNV1A_CH ||
				cl->type == JUMP_CH)
		{
			if (cl->members.ch->servers == NULL) {
				cl->members.ch->servers = w =
					ra_malloc(ret->a, sizeof(servers));
			} else {
				for (w = cl->members.ch->servers; w->next != NULL; w = w->next)
					;
				w = w->next = ra_malloc(ret->a, sizeof(servers));
			}
			if (w == NULL) {
				snprintf(errbuf, sizeof(errbuf), "malloc failed for %s_ch %s",
						cl->type == CARBON_CH ? "carbon" :
						cl->type == FNV1A_CH ? "fnv1a" :
						"jump_fnv1a", ip);
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				return ra_strdup(ret->a, errbuf);
			}
			w->next = NULL;
			w->server = newserver;
			cl->members.ch->ring = ch_addnode(
					cl->members.ch->ring,
					w->server);
			if (cl->members.ch->ring == NULL) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"failed to add server %s:%d "
						"to ring for cluster %s: out of memory",
						ip, port, cl->name);
				return ra_strdup(ret->a, errbuf);
			}
		} else if (cl->type == FORWARD ||
				cl->type == FILELOG ||
				cl->type == FILELOGIP)
		{
			if (cl->members.forward == NULL) {
				cl->members.forward = w =
					ra_malloc(ret->a, sizeof(servers));
			} else {
				for (w = cl->members.forward; w->next != NULL; w = w->next)
					;
				w = w->next = ra_malloc(ret->a, sizeof(servers));
			}
			if (w == NULL) {
				snprintf(errbuf, sizeof(errbuf), "malloc failed for %s %s",
						cl->type == FORWARD ? "forward" :
						cl->type == FILELOG ? "file" :
						"file ip", ip);
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				return ra_strdup(ret->a, errbuf);
			}
			w->next = NULL;
			w->server = newserver;
		} else if (cl->type == ANYOF ||
				cl->type == FAILOVER)
		{
			if (cl->members.anyof == NULL) {
				cl->members.anyof = ra_malloc(ret->a, sizeof(serverlist));
				if (cl->members.anyof != NULL) {
					cl->members.anyof->list = w =
						ra_malloc(ret->a, sizeof(servers));
					cl->members.anyof->count = 1;
					cl->members.anyof->servers = NULL;
				}
			} else {
				for (w = cl->members.anyof->list; w->next != NULL; w = w->next)
					;
				w = w->next = ra_malloc(ret->a, sizeof(servers));
				cl->members.anyof->count++;
			}
			if (w == NULL) {
				snprintf(errbuf, sizeof(errbuf), "malloc failed for %s %s",
						cl->type == ANYOF ? "any_of" :
						"failover", ip);
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				return ra_strdup(ret->a, errbuf);
			}
			w->next = NULL;
			w->server = newserver;

			if (s->refcnt > 1) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"cannot share server %s:%d with "
						"any_of/failover cluster '%s'",
						serverip(newserver),
						server_port(newserver),
						cl->name);
				return ra_strdup(ret->a, errbuf);
			}
		}

		walk = lwalk;
	} while (walk != NULL);

	return NULL;
}

char *
router_add_cluster(router *r, cluster *cl)
{
	cluster *cw;
	cluster *last = NULL;
	servers *w;

	for (cw = r->clusters; cw != NULL; last = cw, cw = cw->next)
		if (cw->name != NULL && cl->name != NULL &&
				strcmp(cw->name, cl->name) == 0)
			return ra_strdup(r->a, "cluster with the same name already defined");
	if (last == NULL)
		last = r->clusters;
	last->next = cl;

	/* post checks/fixups */
	if (cl->type == ANYOF || cl->type == FAILOVER) {
		size_t i = 0;
		cl->members.anyof->servers =
			ra_malloc(r->a, sizeof(server *) * cl->members.anyof->count);
		if (cl->members.anyof->servers == NULL)
			return ra_strdup(r->a, "malloc failed for anyof servers");
		for (w = cl->members.anyof->list; w != NULL; w = w->next)
			cl->members.anyof->servers[i++] = w->server;
		for (w = cl->members.anyof->list; w != NULL; w = w->next) {
			server_add_secondaries(w->server,
					cl->members.anyof->servers,
					cl->members.anyof->count);
			if (cl->type == FAILOVER)
				server_set_failover(w->server);
		}
	} else if (cl->type == CARBON_CH ||
			cl->type == FNV1A_CH ||
			cl->type == JUMP_CH)
	{
		/* check that replication count is actually <= the
		 * number of servers */
		size_t i = 0;
		char errbuf[512];
		for (w = cl->members.ch->servers; w != NULL; w = w->next)
			i++;
		if (i < cl->members.ch->repl_factor) {
			snprintf(errbuf, sizeof(errbuf),
					"invalid cluster '%s': replication count (%u) is "
					"larger than the number of servers (%zu)\n",
					cl->name, cl->members.ch->repl_factor, i);
			return ra_strdup(r->a, errbuf);
		}
	}
	return NULL;
}

char *
router_add_route(router *rtr, route *rte)
{
	route *rw;
	route *last = NULL;
	route *matchallstop = NULL;

	for (rw = rtr->routes; rw != NULL; last = rw, rw = rw->next)
		if (rw->matchtype == MATCHALL && rw->stop)
			matchallstop = rw;
	if (last == NULL) {
		rtr->routes = rte;
		return NULL;
	}

#define matchtype(r) \
	r->dests->cl->type == AGGREGATION ? "aggregate" : \
	r->dests->cl->type == REWRITE ? "rewrite" : \
	"match"
	if (matchallstop != NULL) {
		logerr("warning: %s %s will never match "
				"due to preceding %s * ... stop\n",
				matchtype(rte), rte->pattern == NULL ? "*" : rte->pattern,
				matchtype(matchallstop));
	}
	last->next = rte;
	return NULL;
}

char *
router_add_aggregator(router *rtr, aggregator *a)
{
	aggregator *aw;
	aggregator *last = NULL;

	for (aw = rtr->aggregators; aw != NULL; last = aw, aw = aw->next)
		;
	if (last == NULL) {
		rtr->aggregators = a;
		return NULL;
	}
	last->next = a;
	return NULL;
}

char *
router_add_stubroute(
		router *rtr,
		enum clusttype type,
		cluster *w,
		destinations *dw)
{
	char stubname[48];
	route *m;
	destinations *d;
	cluster *cl;
	char *err;

	m = ra_malloc(rtr->a, sizeof(route));
	if (m == NULL)
		return ra_strdup(rtr->a, "malloc failed for stub route");
	m->pattern = NULL;
	m->strmatch = NULL;
	m->dests = dw;
	m->stop = 1;
	m->matchtype = MATCHALL;
	m->next = NULL;

	/* inject stub route for dests */
	d = ra_malloc(rtr->a, sizeof(destinations));
	if (d == NULL)
		return ra_strdup(rtr->a, "malloc failed for stub destinations");
	d->next = NULL;
	cl = d->cl = ra_malloc(rtr->a, sizeof(cluster));
	if (cl == NULL)
		return ra_strdup(rtr->a, "malloc failed for stub cluster");
	cl->name = NULL;
	cl->type = type;
	cl->members.routes = m;
	cl->next = NULL;
	err = router_add_cluster(rtr, cl);
	if (err != NULL)
		return err;

	if (type == AGGRSTUB) {
		snprintf(stubname, sizeof(stubname),
				STUB_AGGR "%p__", w->members.aggregation);
	} else if (type == STATSTUB) {
		snprintf(stubname, sizeof(stubname),
				STUB_STATS "%p__", rtr);
	} else {
		return ra_strdup(rtr->a, "unknown stub type");
	}
	m = ra_malloc(rtr->a, sizeof(route));
	if (m == NULL)
		return ra_strdup(rtr->a, "malloc failed for catch stub route");
	m->pattern = ra_strdup(rtr->a, stubname);
	m->strmatch = ra_strdup(rtr->a, stubname);
	if (m->pattern == NULL || m->strmatch == NULL)
		ra_strdup(rtr->a, "malloc failed for catch stub pattern");
	m->dests = d;
	m->stop = 1;
	m->matchtype = STARTS_WITH;
	/* enforce first match to avoid interference */
	m->next = rtr->routes;
	rtr->routes = m;

	if (type == AGGRSTUB) {
		aggregator_set_stub(w->members.aggregation, stubname);
	} else if (type == STATSTUB) {
		rtr->collector.stub = m->pattern;
	}

	return NULL;
}

char *
router_add_listener(
		router *rtr,
		rcptr_lsnrtype ltype,
		rcptr_transport trnsp,
		serv_ctype ctype,
		char *ip,
		int port,
		struct addrinfo *saddrs)
{
	listener *lwalk;
	listener *prevl;
	struct addrinfo *swalk;
	struct addrinfo *lswalk;
	int addrcnt;

	if (saddrs == NULL) {
		char msg[256];
		snprintf(msg, sizeof(msg), "invalid listener %s for port %d", ip, port);
		return ra_strdup(rtr->a, msg);
	}

	if (rtr->listeners == NULL) {
		lwalk = rtr->listeners =
			ra_malloc(rtr->a, sizeof(listener));
	} else {
		char hnbufl[INET6_ADDRSTRLEN];
		char hnbufr[INET6_ADDRSTRLEN];

		prevl = NULL;
		for (lwalk = rtr->listeners; lwalk != NULL; lwalk = lwalk->next) {
			for (swalk = saddrs; swalk != NULL; swalk = swalk->ai_next) {
				saddr_ntop(swalk, hnbufl);
				if (lwalk->ctype == ctype && lwalk->port == port) {
					for (lswalk = lwalk->saddrs; lswalk != NULL;
							lswalk = lswalk->ai_next)
					{
						saddr_ntop(lswalk, hnbufr);
						/* ip is equal if:
						 * - the address family is the same (ipv4/ipv6)
						 * - at least one of the sides is 0.0.0.0 (ipv4 any)
						 * - at least one of the sides is :: (ipv6 any)
						 * - both sides are equal */<commit_msg>router_add_listener: fix address check to be only for IP sockets<commit_after>/*
 * Copyright 2013-2017 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <ctype.h>
#include <glob.h>
#include <regex.h>
#include <sys/stat.h>
#include <netdb.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <assert.h>

#include "fnv1a.h"
#include "consistent-hash.h"
#include "server.h"
#include "queue.h"
#include "aggregator.h"
#include "allocator.h"
#include "relay.h"
#include "router.h"
#include "conffile.h"
#include "conffile.tab.h"

struct _router {
	cluster *clusters;
	route *routes;
	aggregator *aggregators;
	servers *srvrs;
	struct _router_collector {
		char *stub;
		int interval;
		char *prefix;
		col_mode mode;
	} collector;
	listener *listeners;
	struct _router_parser_err {
		const char *msg;
		size_t line;
		size_t start;
		size_t stop;
	} parser_err;
	struct _router_conf {
		char workercnt;
		size_t queuesize;
		size_t batchsize;
		int maxstalls;
		unsigned short iotimeout;
		unsigned int sockbufsize;
	} conf;
	allocator *a;
};

/* declarations to avoid symbol/include hell when doing config.yy.h */
int router_yylex_init(void **);
void router_yy_scan_string(char *, void *);
int router_yylex_destroy(void *);

/* catch string used for aggrgator destinations */
#define STUB_AGGR   "_aggregator_stub_"
/* catch string used for collector output */
#define STUB_STATS  "_collector_stub_"

/**
 * Frees regexes from routes.
 */
static void
router_free_intern(route *routes, char workercnt)
{
	/* the only thing that isn't allocated using the allocators are the
	 * regexes */
	while (routes != NULL) {
		if (routes->matchtype == REGEX) {
			int i;
			for(i = 0; i < workercnt; i++)
				regfree(&routes->rule[i]);
		}

		if (routes->next == NULL || routes->next->dests != routes->dests) {
			while (routes->dests != NULL) {
				if (routes->dests->cl->type == GROUP ||
						routes->dests->cl->type == AGGRSTUB ||
						routes->dests->cl->type == STATSTUB)
					router_free_intern(routes->dests->cl->members.routes,
							workercnt);
				if (routes->dests->cl->type == VALIDATION)
					router_free_intern(
							routes->dests->cl->members.validation->rule,
							workercnt);

				routes->dests = routes->dests->next;
			}
		}

		routes = routes->next;
	}
}

/**
 * Examines pattern and sets matchtype and rule or strmatch in route.
 */
static inline int
determine_if_regex(allocator *a, char workercnt, route *r, char *pat)
{
	/* try and see if we can avoid using a regex match, for
	 * it is simply very slow/expensive to do so: most of
	 * the time, people don't need fancy matching rules */
	char patbuf[8192];
	char *e = pat;
	char *pb = patbuf;
	char escape = 0;
	r->matchtype = CONTAINS;
	r->nmatch = 0;

	if (*e == '^' && r->matchtype == CONTAINS) {
		e++;
		r->matchtype = STARTS_WITH;
	}
	for (; *e != '\0'; e++) {
		switch (*e) {
			case '\\':
				if (escape)
					*pb++ = *e;
				escape = !escape;
				break;
			case '.':
			case '^':
			case '*':
			case '+':
				if (!escape)
					r->matchtype = REGEX;
				*pb++ = *e;
				escape = 0;
				break;
			case '$':
				if (!escape && e[1] == '\0') {
					if (r->matchtype == STARTS_WITH) {
						r->matchtype = MATCHES;
					} else {
						r->matchtype = ENDS_WITH;
					}
				} else {
					r->matchtype = REGEX;
				}
				escape = 0;
				break;
			default:
				if (
						!escape && (
							(*e == '_') || (*e == '-') ||
							(*e >= '0' && *e <= '9') ||
							(*e >= 'a' && *e <= 'z') ||
							(*e >= 'A' && *e <= 'Z')
						)
				   )
				{
					*pb++ = *e;
				} else {
					r->matchtype = REGEX;
				}
				escape = 0;
				break;
		}
		if (pb - patbuf == sizeof(patbuf))
			r->matchtype = REGEX;
		if (r->matchtype == REGEX)
			break;
	}
	if (r->matchtype != REGEX) {
		*pb = '\0';
		r->strmatch = ra_strdup(a, patbuf);
		r->pattern = ra_strdup(a, pat);
	} else {
		int i;
		int ret;

		r->rule = ra_malloc(a, sizeof(*r->rule) * workercnt);
		if (r->rule == NULL) {
			logerr("determine_if_regex: malloc failed for "
					"regular expressions\n");
			return REG_ESPACE;  /* lie closest to the truth */
		}
		ret = regcomp(&r->rule[0], pat, REG_EXTENDED);
		if (ret != 0)
			return ret;  /* allow use of regerror */
		if (r->rule[0].re_nsub > 0) {
			/* we need +1 because position 0 contains the entire
			 * expression */
			r->nmatch = r->rule[0].re_nsub + 1;
			if (r->nmatch > RE_MAX_MATCHES) {
				logerr("determine_if_regex: too many match groups, "
						"please increase RE_MAX_MATCHES in router.h\n");
				regfree(&r->rule[0]);
				return REG_ESPACE;  /* lie closest to the truth */
			}
		}
		for (i = 1; i < workercnt; i++) {
			if (regcomp(&r->rule[i], pat, REG_EXTENDED) != 0) {
				while (--i >= 0)
					regfree(&r->rule[i]);
				logerr("determine_if_regex: out of memory allocating "
						"regexes\n");
				return REG_ESPACE;  /* lie closest to the truth */
			}
		}
		r->strmatch = NULL;
		r->pattern = ra_strdup(a, pat);
	}

	return 0;
}

static char serverip_server[256];  /* RFC1035 = 250 */
static const char *
serverip(server *s)
{
	const char *srvr = server_ip(s);

	if (strchr(srvr, ':') != NULL) {
		snprintf(serverip_server, sizeof(serverip_server), "[%s]", srvr);
		return serverip_server;
	}

	return srvr;
}


/**
 * Callback for the bison parser for yyerror
 */
void
router_yyerror(
		void *locptr,
		void *s,
		router *r,
		allocator *ra,
		allocator *a,
		const char *msg)
{
	(void)s;
	ROUTER_YYLTYPE *locp = (ROUTER_YYLTYPE *)locptr;
	regex_t re;
	size_t nmatch = 3;
	regmatch_t pmatch[3];
	char buf1[METRIC_BUFSIZ];
	char buf2[METRIC_BUFSIZ];
	char *dummy;
	char (*sa)[METRIC_BUFSIZ];
	char (*sb)[METRIC_BUFSIZ];
	char (*st)[METRIC_BUFSIZ];

	if (r->parser_err.msg != NULL)
		return;

	/* clean up the "value" types */
	if (regcomp(&re, "cr(STRING|COMMENT|INTVAL)", REG_EXTENDED) != 0) {
		r->parser_err.msg = ra_strdup(a, msg);
		return;
	}
	snprintf(buf1, sizeof(buf1), "%s", msg);
	sa = &buf1;
	sb = &buf2;
	while (regexec(&re, *sa, nmatch, pmatch, 0) == 0) {
		dummy = *sa + strlen(*sa);
		if (router_rewrite_metric(sb, &dummy,
					*sa, dummy, "\\_1", nmatch, pmatch) == 0)
		{
			r->parser_err.msg = ra_strdup(a, msg);
			return;
		}
		st = sa;
		sa = sb;
		sb = st;
	}
	regfree(&re);

	/* clean up the keywords */
	if (regcomp(&re, "cr([A-Z]+)", REG_EXTENDED) != 0) {
		r->parser_err.msg = ra_strdup(a, msg);
		return;
	}
	while (regexec(&re, *sa, nmatch, pmatch, 0) == 0) {
		dummy = *sa + strlen(*sa);
		if (router_rewrite_metric(sb, &dummy,
					*sa, dummy, "'\\_1'", nmatch, pmatch) == 0)
		{
			r->parser_err.msg = ra_strdup(a, msg);
			return;
		}
		st = sa;
		sa = sb;
		sb = st;
	}
	regfree(&re);

	/* clean up bison speak */
	if (regcomp(&re, "\\$end", REG_EXTENDED) != 0) {
		r->parser_err.msg = ra_strdup(a, msg);
		return;
	}
	while (regexec(&re, *sa, nmatch, pmatch, 0) == 0) {
		dummy = *sa + strlen(*sa);
		if (router_rewrite_metric(sb, &dummy,
					*sa, dummy, "end of file", nmatch, pmatch) == 0)
		{
			r->parser_err.msg = ra_strdup(a, msg);
			return;
		}
		st = sa;
		sa = sb;
		sb = st;
	}
	regfree(&re);

	r->parser_err.msg = ra_strdup(a, *sa);
	if (locp != NULL) {
		r->parser_err.line = 1 + locp->first_line;
		r->parser_err.start = locp->first_column;
		if (locp->first_line == locp->last_line) {
			r->parser_err.stop = locp->last_column;
		} else {
			r->parser_err.stop = locp->first_column;
		}
	} else {
		r->parser_err.line = 0;
	}
}

/**
 * Parse ip string and validate it.  When it is a numeric address, write
 * the cannonical version in retip.
 */
char *
router_validate_address(
		router *rtr,
		char **retip, int *retport, void **retsaddr, void **rethint,
		char *ip, serv_ctype proto)
{
	int port = GRAPHITE_PORT;
	struct addrinfo *saddr = NULL;
	struct addrinfo hint;
	char sport[8];
	char *p;
	char *lastcolon = NULL;

	for (p = ip; *p != '\0'; p++)
		if (*p == ':')
			lastcolon = p;

	if (*(p - 1) == ']')
		lastcolon = NULL;
	if (lastcolon != NULL) {
		char *endp = NULL;
		*lastcolon = '\0';
		port = (int)strtol(lastcolon + 1, &endp, 10);
		if (port < 1 || endp != p)
			return(ra_strdup(rtr->a, "invalid port number"));
	}
	if (*ip == '[') {
		ip++;
		if (lastcolon != NULL && *(lastcolon - 1) == ']') {
			*(lastcolon - 1) = '\0';
		} else if (lastcolon == NULL && *(p - 1) == ']') {
			*(p - 1) = '\0';
		} else {
			return(ra_strdup(rtr->a, "expected ']'"));
		}
	}
	if (lastcolon == ip)
		ip = NULL;  /* only resolve port, e.g. any interface */

	memset(&hint, 0, sizeof(hint));
	saddr = NULL;

	/* try to see if this is a "numeric" IP address, which means
	 * re-resolving lateron will make no difference */
	hint.ai_family = PF_UNSPEC;
	hint.ai_socktype = proto == CON_UDP ? SOCK_DGRAM : SOCK_STREAM;
	hint.ai_protocol = proto == CON_UDP ? IPPROTO_UDP : IPPROTO_TCP;
	hint.ai_flags = AI_NUMERICHOST | AI_NUMERICSERV | AI_PASSIVE;
	snprintf(sport, sizeof(sport), "%u", port);

	*rethint = NULL;
	if (getaddrinfo(ip, sport, &hint, &saddr) != 0) {
		int err;
		/* now resolve this the normal way to check validity */
		hint.ai_flags = AI_NUMERICSERV;
		if ((err = getaddrinfo(ip, sport, &hint, &saddr)) != 0) {
			char errmsg[512];
			snprintf(errmsg, sizeof(errmsg), "warning: failed to resolve "
					"%s, port %s, proto %s: %s",
					ip == NULL ? "<any>" : ip, sport,
					proto == CON_UDP ? "udp" : "tcp",
					gai_strerror(err));
			saddr = NULL;
		}

		/* no ra_malloc, it gets owned by server */
		*rethint = malloc(sizeof(hint));
		if (*rethint == NULL) {
			freeaddrinfo(saddr);
			return(ra_strdup(rtr->a, "out of memory copying hint structure"));
		}
		memcpy(*rethint, &hint, sizeof(hint));
	}

	*retip = ip;
	*retport = port;
	*retsaddr = saddr;
	return NULL;
}

/**
 * Parse path string and validate it can be written to.
 */
char *
router_validate_path(router *rtr, char *path)
{
	struct stat st;
	char fileexists = 1;
	int probefd;

	/* if the file doesn't exist, remove it after trying to create it */
	if (stat(path, &st) == -1)
		fileexists = 0;

	if ((probefd = open(path,
					O_WRONLY | O_APPEND | O_CREAT,
					S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)) < 0)
	{
		char errbuf[512];
		if (errno == EOPNOTSUPP && st.st_mode & S_IFSOCK) {
			/* This is pretty lame, but I don't know a better way: we
			 * can't open sockets in use, and as it stands we can't
			 * check it is *us* who have the socket in use.  So, we
			 * assume we do, and ignore this silently.  In the worst
			 * case this yields an error when it opens up for real. */
			return NULL;
		}
		snprintf(errbuf, sizeof(errbuf),
				"failed to open file '%s' for writing: %s",
				path, strerror(errno));
		return ra_strdup(rtr->a, errbuf);
	}
	close(probefd);

	if (!fileexists)
		unlink(path);

	return NULL;
}

/**
 * Checks pat against regular expression syntax (or being '*') and
 * returns a route struct with the parsed info.  If retr is NULL a route
 * struct is allocated.
 */
char *
router_validate_expression(router *rtr, route **retr, char *pat)
{
	route *r = *retr;
	if (r == NULL) {
		r = *retr = ra_malloc(rtr->a, sizeof(route));
		if (r == NULL)
			return ra_strdup(rtr->a, "out of memory allocating route");
		r->next = NULL;
		r->dests = NULL;
	}
	if (strcmp(pat, "*") == 0) {
		r->pattern = NULL;
		r->strmatch = NULL;
		r->matchtype = MATCHALL;
	} else {
		int err = determine_if_regex(rtr->a, rtr->conf.workercnt, r, pat);
		if (err != 0) {
			char ebuf[512];
			size_t s = snprintf(ebuf, sizeof(ebuf),
					"invalid expression '%s': ", pat);
			regerror(err, &r->rule[0], ebuf + s, sizeof(ebuf) - s);
			return ra_strdup(rtr->a, ebuf);
		}
	}

	return NULL;
}

/**
 * Check if cluster is defined, and set the pointer to it.
 */
char *
router_validate_cluster(router *rtr, cluster **clptr, char *cl)
{
	cluster *w;

	/* lookup cluster */
	for (w = rtr->clusters; w != NULL; w = w->next) {
		if (w->type != GROUP &&
				w->type != AGGRSTUB &&
				w->type != STATSTUB &&
				w->type != AGGREGATION &&
				w->type != REWRITE &&
				w->type != VALIDATION &&
				strcmp(w->name, cl) == 0)
			break;
	}
	if (w == NULL)
		return ra_strdup(rtr->a, "unknown cluster");

	*clptr = w;
	return NULL;
}

/**
 * Adds a server to the chain in router, expands if necessary.
 */
char *
router_add_server(
		router *ret,
		char *ip,
		int port,
		char *inst,
		serv_ctype proto,
		struct addrinfo *saddrs,
		struct addrinfo *hint,
		char useall,
		cluster *cl)
{
	struct addrinfo *walk;
	struct addrinfo *lwalk;
	char hnbuf[INET6_ADDRSTRLEN];
	char errbuf[512];
	server *newserver;
	servers *w = NULL;

	walk = saddrs;  /* NULL if file */
	do {
		servers *s;
		lwalk = NULL;

		if (useall) {
			/* serialise the IP address, to make the targets explicit
			 * (since we're expanding all A/AAAA records) */
			saddr_ntop(walk, hnbuf);
			if (hnbuf[0] != '\0')
				ip = hnbuf;
			/* now we've expanded, don't trigger re-resolving */
			if (hint)
				free(hint);
			hint = NULL;
			/* and turn this into a separate entry */
			lwalk = walk->ai_next;
			walk->ai_next = NULL;
		}

		newserver = NULL;
		for (s = ret->srvrs; s != NULL; s = s->next) {
			if (server_cmp(s->server, walk, ip) == 0) {
				newserver = s->server;
				s->refcnt++;
				break;
			}
		}
		if (newserver == NULL) {
			newserver = server_new(ip, (unsigned short)port,
					proto, walk, hint,
					ret->conf.queuesize, ret->conf.batchsize,
					ret->conf.maxstalls, ret->conf.iotimeout,
					ret->conf.sockbufsize);
			if (newserver == NULL) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"failed to add server %s:%d (%s) "
						"to cluster %s", ip, port,
						proto == CON_UDP ? "udp" :
						proto == CON_TCP ? "tcp" :
						proto == CON_FILE ? "file" : "???", cl->name);
				return ra_strdup(ret->a, errbuf);
			}
			if (ret->srvrs == NULL) {
				s = ret->srvrs = ra_malloc(ret->a, sizeof(servers));
			} else {
				for (s = ret->srvrs; s->next != NULL; s = s->next)
					;
				s = s->next = ra_malloc(ret->a, sizeof(servers));
			}
			s->next = NULL;
			s->refcnt = 1;
			s->server = newserver;
		}
		
		/* check instance matches the existing server */
		if (s->refcnt > 1) {
			char *sinst = server_instance(newserver);
			if (sinst == NULL && inst != NULL) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"cannot set instance '%s' for "
						"server %s:%d: server was previously "
						"defined without instance",
						inst, serverip(newserver),
						server_port(newserver));
				return ra_strdup(ret->a, errbuf);
			} else if (sinst != NULL && inst == NULL) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"cannot define server %s:%d without "
						"instance: server was previously "
						"defined with instance '%s'",
						serverip(newserver),
						server_port(newserver), sinst);
				return ra_strdup(ret->a, errbuf);
			} else if (sinst != NULL && inst != NULL &&
					strcmp(sinst, inst) != 0)
			{
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"cannot set instance '%s' for "
						"server %s:%d: server was previously "
						"defined with instance '%s'",
						inst, serverip(newserver),
						server_port(newserver), sinst);
				return ra_strdup(ret->a, errbuf);
			} /* else: sinst == inst == NULL */
		}
		if (inst != NULL)
			server_set_instance(newserver, inst);

		if (cl->type == CARBON_CH ||
				cl->type == FNV1A_CH ||
				cl->type == JUMP_CH)
		{
			if (cl->members.ch->servers == NULL) {
				cl->members.ch->servers = w =
					ra_malloc(ret->a, sizeof(servers));
			} else {
				for (w = cl->members.ch->servers; w->next != NULL; w = w->next)
					;
				w = w->next = ra_malloc(ret->a, sizeof(servers));
			}
			if (w == NULL) {
				snprintf(errbuf, sizeof(errbuf), "malloc failed for %s_ch %s",
						cl->type == CARBON_CH ? "carbon" :
						cl->type == FNV1A_CH ? "fnv1a" :
						"jump_fnv1a", ip);
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				return ra_strdup(ret->a, errbuf);
			}
			w->next = NULL;
			w->server = newserver;
			cl->members.ch->ring = ch_addnode(
					cl->members.ch->ring,
					w->server);
			if (cl->members.ch->ring == NULL) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"failed to add server %s:%d "
						"to ring for cluster %s: out of memory",
						ip, port, cl->name);
				return ra_strdup(ret->a, errbuf);
			}
		} else if (cl->type == FORWARD ||
				cl->type == FILELOG ||
				cl->type == FILELOGIP)
		{
			if (cl->members.forward == NULL) {
				cl->members.forward = w =
					ra_malloc(ret->a, sizeof(servers));
			} else {
				for (w = cl->members.forward; w->next != NULL; w = w->next)
					;
				w = w->next = ra_malloc(ret->a, sizeof(servers));
			}
			if (w == NULL) {
				snprintf(errbuf, sizeof(errbuf), "malloc failed for %s %s",
						cl->type == FORWARD ? "forward" :
						cl->type == FILELOG ? "file" :
						"file ip", ip);
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				return ra_strdup(ret->a, errbuf);
			}
			w->next = NULL;
			w->server = newserver;
		} else if (cl->type == ANYOF ||
				cl->type == FAILOVER)
		{
			if (cl->members.anyof == NULL) {
				cl->members.anyof = ra_malloc(ret->a, sizeof(serverlist));
				if (cl->members.anyof != NULL) {
					cl->members.anyof->list = w =
						ra_malloc(ret->a, sizeof(servers));
					cl->members.anyof->count = 1;
					cl->members.anyof->servers = NULL;
				}
			} else {
				for (w = cl->members.anyof->list; w->next != NULL; w = w->next)
					;
				w = w->next = ra_malloc(ret->a, sizeof(servers));
				cl->members.anyof->count++;
			}
			if (w == NULL) {
				snprintf(errbuf, sizeof(errbuf), "malloc failed for %s %s",
						cl->type == ANYOF ? "any_of" :
						"failover", ip);
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				return ra_strdup(ret->a, errbuf);
			}
			w->next = NULL;
			w->server = newserver;

			if (s->refcnt > 1) {
				freeaddrinfo(saddrs);
				if (hint)
					free(hint);
				snprintf(errbuf, sizeof(errbuf),
						"cannot share server %s:%d with "
						"any_of/failover cluster '%s'",
						serverip(newserver),
						server_port(newserver),
						cl->name);
				return ra_strdup(ret->a, errbuf);
			}
		}

		walk = lwalk;
	} while (walk != NULL);

	return NULL;
}

char *
router_add_cluster(router *r, cluster *cl)
{
	cluster *cw;
	cluster *last = NULL;
	servers *w;

	for (cw = r->clusters; cw != NULL; last = cw, cw = cw->next)
		if (cw->name != NULL && cl->name != NULL &&
				strcmp(cw->name, cl->name) == 0)
			return ra_strdup(r->a, "cluster with the same name already defined");
	if (last == NULL)
		last = r->clusters;
	last->next = cl;

	/* post checks/fixups */
	if (cl->type == ANYOF || cl->type == FAILOVER) {
		size_t i = 0;
		cl->members.anyof->servers =
			ra_malloc(r->a, sizeof(server *) * cl->members.anyof->count);
		if (cl->members.anyof->servers == NULL)
			return ra_strdup(r->a, "malloc failed for anyof servers");
		for (w = cl->members.anyof->list; w != NULL; w = w->next)
			cl->members.anyof->servers[i++] = w->server;
		for (w = cl->members.anyof->list; w != NULL; w = w->next) {
			server_add_secondaries(w->server,
					cl->members.anyof->servers,
					cl->members.anyof->count);
			if (cl->type == FAILOVER)
				server_set_failover(w->server);
		}
	} else if (cl->type == CARBON_CH ||
			cl->type == FNV1A_CH ||
			cl->type == JUMP_CH)
	{
		/* check that replication count is actually <= the
		 * number of servers */
		size_t i = 0;
		char errbuf[512];
		for (w = cl->members.ch->servers; w != NULL; w = w->next)
			i++;
		if (i < cl->members.ch->repl_factor) {
			snprintf(errbuf, sizeof(errbuf),
					"invalid cluster '%s': replication count (%u) is "
					"larger than the number of servers (%zu)\n",
					cl->name, cl->members.ch->repl_factor, i);
			return ra_strdup(r->a, errbuf);
		}
	}
	return NULL;
}

char *
router_add_route(router *rtr, route *rte)
{
	route *rw;
	route *last = NULL;
	route *matchallstop = NULL;

	for (rw = rtr->routes; rw != NULL; last = rw, rw = rw->next)
		if (rw->matchtype == MATCHALL && rw->stop)
			matchallstop = rw;
	if (last == NULL) {
		rtr->routes = rte;
		return NULL;
	}

#define matchtype(r) \
	r->dests->cl->type == AGGREGATION ? "aggregate" : \
	r->dests->cl->type == REWRITE ? "rewrite" : \
	"match"
	if (matchallstop != NULL) {
		logerr("warning: %s %s will never match "
				"due to preceding %s * ... stop\n",
				matchtype(rte), rte->pattern == NULL ? "*" : rte->pattern,
				matchtype(matchallstop));
	}
	last->next = rte;
	return NULL;
}

char *
router_add_aggregator(router *rtr, aggregator *a)
{
	aggregator *aw;
	aggregator *last = NULL;

	for (aw = rtr->aggregators; aw != NULL; last = aw, aw = aw->next)
		;
	if (last == NULL) {
		rtr->aggregators = a;
		return NULL;
	}
	last->next = a;
	return NULL;
}

char *
router_add_stubroute(
		router *rtr,
		enum clusttype type,
		cluster *w,
		destinations *dw)
{
	char stubname[48];
	route *m;
	destinations *d;
	cluster *cl;
	char *err;

	m = ra_malloc(rtr->a, sizeof(route));
	if (m == NULL)
		return ra_strdup(rtr->a, "malloc failed for stub route");
	m->pattern = NULL;
	m->strmatch = NULL;
	m->dests = dw;
	m->stop = 1;
	m->matchtype = MATCHALL;
	m->next = NULL;

	/* inject stub route for dests */
	d = ra_malloc(rtr->a, sizeof(destinations));
	if (d == NULL)
		return ra_strdup(rtr->a, "malloc failed for stub destinations");
	d->next = NULL;
	cl = d->cl = ra_malloc(rtr->a, sizeof(cluster));
	if (cl == NULL)
		return ra_strdup(rtr->a, "malloc failed for stub cluster");
	cl->name = NULL;
	cl->type = type;
	cl->members.routes = m;
	cl->next = NULL;
	err = router_add_cluster(rtr, cl);
	if (err != NULL)
		return err;

	if (type == AGGRSTUB) {
		snprintf(stubname, sizeof(stubname),
				STUB_AGGR "%p__", w->members.aggregation);
	} else if (type == STATSTUB) {
		snprintf(stubname, sizeof(stubname),
				STUB_STATS "%p__", rtr);
	} else {
		return ra_strdup(rtr->a, "unknown stub type");
	}
	m = ra_malloc(rtr->a, sizeof(route));
	if (m == NULL)
		return ra_strdup(rtr->a, "malloc failed for catch stub route");
	m->pattern = ra_strdup(rtr->a, stubname);
	m->strmatch = ra_strdup(rtr->a, stubname);
	if (m->pattern == NULL || m->strmatch == NULL)
		ra_strdup(rtr->a, "malloc failed for catch stub pattern");
	m->dests = d;
	m->stop = 1;
	m->matchtype = STARTS_WITH;
	/* enforce first match to avoid interference */
	m->next = rtr->routes;
	rtr->routes = m;

	if (type == AGGRSTUB) {
		aggregator_set_stub(w->members.aggregation, stubname);
	} else if (type == STATSTUB) {
		rtr->collector.stub = m->pattern;
	}

	return NULL;
}

char *
router_add_listener(
		router *rtr,
		rcptr_lsnrtype ltype,
		rcptr_transport trnsp,
		serv_ctype ctype,
		char *ip,
		int port,
		struct addrinfo *saddrs)
{
	listener *lwalk;
	listener *prevl;
	struct addrinfo *swalk;
	struct addrinfo *lswalk;
	int addrcnt;

	if (saddrs == NULL && (ctype == CON_UDP || ctype == CON_TCP)) {
		char msg[256];
		snprintf(msg, sizeof(msg), "invalid listener %s for port %d", ip, port);
		return ra_strdup(rtr->a, msg);
	}

	if (rtr->listeners == NULL) {
		lwalk = rtr->listeners =
			ra_malloc(rtr->a, sizeof(listener));
	} else {
		char hnbufl[INET6_ADDRSTRLEN];
		char hnbufr[INET6_ADDRSTRLEN];

		prevl = NULL;
		for (lwalk = rtr->listeners; lwalk != NULL; lwalk = lwalk->next) {
			for (swalk = saddrs; swalk != NULL; swalk = swalk->ai_next) {
				saddr_ntop(swalk, hnbufl);
				if (lwalk->ctype == ctype && lwalk->port == port) {
					for (lswalk = lwalk->saddrs; lswalk != NULL;
							lswalk = lswalk->ai_next)
					{
						saddr_ntop(lswalk, hnbufr);
						/* ip is equal if:
						 * - the address family is the same (ipv4/ipv6)
						 * - at least one of the sides is 0.0.0.0 (ipv4 any)
						 * - at least one of the sides is :: (ipv6 any)
						 * - both sides are equal */
{'fraction_non_alphanumeric': 0.1695161653596511, 'fraction_numerical': 0.006142631913020332, 'mean_word_length': 2.9895441921254697, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 687, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3149834', 'n_tokens_mistral': 21796, 'n_tokens_neox': 18825, 'n_words': 4897}