Add prometheus monitoring

11 files changed, 446 insertions, 56 deletions

diff --git a/src/benc.c b/src/benc.c
index 4622698..c2c2782 100644
--- a/src/benc.c
+++ b/src/benc.c
@@ -178,14 +178,14 @@ int64_t benc_decode(const char** cursor, const char* end, int* depth, struct ben
 
 int bcur_fill(struct bcursor* cursor, size_t ignoring) {
 	if(cursor->source == cursor->source_end) {
-		return EOF;
+		return BENC_EBADP;
 	}
 
 	int read;
 	while(true) {
 		read = benc_decode(&cursor->source, cursor->source_end, &cursor->source_depth, cursor->base, cursor->base_len);
 		if(read < 0) {
-			return EINVAL;
+			return BENC_EBADP;
 		}
 		if(read > ignoring) {
 			break;
@@ -229,7 +229,7 @@ int bcur_next_sibling(struct bcursor* cursor) {
 		cursor->readhead++;
 		while(cursor->readhead->type != BNT_END || cursor->readhead->depth != tdepth) {
 			if(bcur_next(cursor, 1) != 0) {
-				fatal("Invalid dict/list");
+				return BENC_EBADP;
 			}
 		}
 	}
@@ -247,7 +247,7 @@ ssize_t bcur_find_key(struct bcursor* cursor, const enum benc_nodetype* keyTypes
 		if(cursor->readhead->type == BNT_LIST || cursor->readhead->type == BNT_DICT) {
 			rc = bcur_next_sibling(cursor);
 			if(rc != 0) {
-				fatal("Invalid dict");
+				return -rc;
 			}
 		} else {
 			// Check if the key one of the ones we are looking for
@@ -261,25 +261,25 @@ ssize_t bcur_find_key(struct bcursor* cursor, const enum benc_nodetype* keyTypes
 		// Skip the key part
 		rc = bcur_next(cursor, 1);
 		if(rc != 0) {
-			fatal("Invalid dict");
+			return -rc;
 		}
 
 		if(cursor->readhead->type == BNT_LIST || cursor->readhead->type == BNT_DICT) {
 			// Skip a multitoken element
 			rc = bcur_next_sibling(cursor);
 			if(rc != 0) {
-				fatal("Invalid dict");
+				return -rc;
 			}
 		} else {
 			// Skip a single token element
 			rc = bcur_next(cursor, 1);
 			if(rc != 0) {
-				fatal("Invalid dict");
+				return -rc;
 			}
 		}
 
 		if(cursor->readhead->type == BNT_END) {
-			return -1;
+			return -BENC_EENDP;
 		}
 	}
 }
diff --git a/src/benc.h b/src/benc.h
index e3ca026..8e2cd3a 100644
--- a/src/benc.h
+++ b/src/benc.h
@@ -6,6 +6,7 @@
 #include <unistd.h>
 
 #define BENC_EBADP 1
+#define BENC_EENDP 2
 
 enum benc_nodetype {
 	BNT_INT,
diff --git a/src/log.h b/src/log.h
index 77eab35..fe1c49a 100644
--- a/src/log.h
+++ b/src/log.h
@@ -7,17 +7,17 @@
 	dbgl(format "\n", ##  __VA_ARGS__)
 
 #define dbgl(format, ...) do{\
-		printf(format, ## __VA_ARGS__); \
+		fprintf(stderr, format, ## __VA_ARGS__); \
 		fflush(stderr); \
 	} while(0)
 
 #define err(format, ...) do{\
-		printf(format "\n", ## __VA_ARGS__); \
+		fprintf(stderr, format "\n", ## __VA_ARGS__); \
 		fflush(stderr); \
 	} while(0)
 
 #define fatal(format, ...) do{\
-		printf(format "\n", ## __VA_ARGS__); \
+		fprintf(stderr, format "\n", ## __VA_ARGS__); \
 		fflush(stderr); \
 		abort(); \
 	} while(0)
diff --git a/src/main.c b/src/main.c
index 8545bf4..a51fcfc 100644
--- a/src/main.c
+++ b/src/main.c
@@ -1,12 +1,57 @@
 #include "proto.h"
 #include "peers.h"
 #include "log.h"
+#include "metrics.h"
 
 #include <time.h>
 #include <assert.h>
 #include <errno.h>
 #include <signal.h>
 
+#include <stdint.h>
+#include <stdlib.h>
+
+static char encoding_table[] = {
+	'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H',
+	'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P',
+	'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X',
+	'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f',
+	'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n',
+	'o', 'p', 'q', 'r', 's', 't', 'u', 'v',
+	'w', 'x', 'y', 'z', '0', '1', '2', '3',
+	'4', '5', '6', '7', '8', '9', '+', '/'
+};
+static int mod_table[] = {0, 2, 1};
+
+
+char *base64_encode(const unsigned char *data, size_t input_length, size_t *output_length) {
+	*output_length = 4 * ((input_length + 2) / 3);
+
+	char *encoded_data = malloc(*output_length + 1);
+	assert(encoded_data != NULL);
+
+	for (int i = 0, j = 0; i < input_length;) {
+		uint32_t octet_a = i < input_length ? (unsigned char)data[i++] : 0;
+		uint32_t octet_b = i < input_length ? (unsigned char)data[i++] : 0;
+		uint32_t octet_c = i < input_length ? (unsigned char)data[i++] : 0;
+
+		uint32_t triple = (octet_a << 0x10) + (octet_b << 0x08) + octet_c;
+
+		encoded_data[j++] = encoding_table[(triple >> 3 * 6) & 0x3F];
+		encoded_data[j++] = encoding_table[(triple >> 2 * 6) & 0x3F];
+		encoded_data[j++] = encoding_table[(triple >> 1 * 6) & 0x3F];
+		encoded_data[j++] = encoding_table[(triple >> 0 * 6) & 0x3F];
+	}
+
+	for (int i = 0; i < mod_table[input_length % 3]; i++) {
+		encoded_data[*output_length - 1 - i] = '=';
+	}
+
+	encoded_data[*output_length] = 0;
+
+	return encoded_data;
+}
+
 static volatile bool killed = false;
 void sigint_handler(int sig) {
 	killed = true;
@@ -72,12 +117,14 @@ int read_config() {
 
 void flush_messages(int sfd, struct message* cursor, const struct message* const end) {
 	dbg("Flushing %ld pending messages", end - cursor);
+	prom_counter_add(requests, end - cursor, NULL);
 	for(; cursor < end; cursor++) {
 		//now reply the client with the same data
 		int rc = sendto(sfd, cursor->payload, cursor->payload_len, 0, (const struct sockaddr*)&cursor->dest, cursor->dest_len);
 		if (rc < 0) {
 			fatal("Failed to send message %m");
 		}
+		prom_counter_add(bytesSent, cursor->payload_len, NULL);
 	}
 }
 
@@ -95,6 +142,7 @@ struct lookup {
 
 #define OUTBOX_SIZE 32
 int main(int argc, char** argv) {
+	srand(time(NULL));
 	struct message outbuff[OUTBOX_SIZE] = {0};
 
 	struct sigaction sa;
@@ -105,11 +153,17 @@ int main(int argc, char** argv) {
 	if(sigaction(SIGINT, &sa, NULL) == -1)
 		fatal("Couldn't set signal handler");
 
+	if(sigaction(SIGTERM, &sa, NULL) == -1)
+		fatal("Couldn't set signal handler");
+
 	struct dht dht = {0};
 	{
 		int rc = read_config();
 		if(rc == CONF_ENO) {
-			myID = (struct nodeid){.inner={0xebe9bbf1, 0x3cdba6b3, 0x993e0c87, 0x900d5e25, 0x00000000}};
+			for(uint16_t i = 0; i < sizeof(myID.inner_b); i++) {
+				myID.inner_b[i] = rand();
+			}
+			/* myID = (struct nodeid){.inner={0xebe9bbf1, 0x3cdba6b3, 0x993e0c87, 0x900d5e25, 0x00000000}}; */
 			routing_flush();
 			allocate_hashtable();
 		}
@@ -117,24 +171,30 @@ int main(int argc, char** argv) {
 		dht.self = myID;
 	}
 
+	metric_init();
+
+	size_t outLen;
+	const char *id = base64_encode((unsigned char*)myID.inner_b, 20, &outLen);
+	prom_counter_inc(meta, (const char *[]){id});
+
 	struct message* message_cursor = outbuff;
 	proto_begin(&dht, time(NULL), &message_cursor, outbuff+32);
 	flush_messages(dht.sfd, outbuff, message_cursor);
 
-	struct lookup lookup;
-	// Init the lookup
-	{
-		lookup.wake = 0;
-		lookup.target = (struct nodeid){.inner={0x19b8a941, 0x38fa0191, 0x1403fac2, 0x581000ab, 0x19583cda}};
-
-		struct entry* entry[8];
-		int found = routing_closest(&lookup.target, 8, entry);
-		for(size_t i = 0; i < found; i++) {
-			lookup.closest[i] = entry[i]->id;
-			lookup.closest_addr[i] = entry[i]->addr;
-			lookup.closest_valid[i] = true;
-		}
-	}
+	/* struct lookup lookup; */
+	/* // Init the lookup */
+	/* { */
+	/* 	lookup.wake = 0; */
+	/* 	lookup.target = (struct nodeid){.inner={0x19b8a941, 0x38fa0191, 0x1403fac2, 0x581000ab, 0x19583cda}}; */
+
+	/* 	struct entry* entry[8]; */
+	/* 	int found = routing_closest(&lookup.target, 8, entry); */
+	/* 	for(size_t i = 0; i < found; i++) { */
+	/* 		lookup.closest[i] = entry[i]->id; */
+	/* 		lookup.closest_addr[i] = entry[i]->addr; */
+	/* 		lookup.closest_valid[i] = true; */
+	/* 	} */
+	/* } */
 
 #define RECV_BUFF_SIZE 4096
 	char buff_storage[RECV_BUFF_SIZE+1];
@@ -180,6 +240,7 @@ int main(int argc, char** argv) {
 				dbg("Receive buffer too small");
 				continue;
 			}
+			prom_counter_add(bytesRecv, recv_len, NULL);
 			// Null terminate the packet
 			if(buff != NULL) {
 				buff[recv_len] = '\0';
@@ -196,6 +257,7 @@ int main(int argc, char** argv) {
 	}
 
 	proto_end(&dht);
+	metric_end();
 	dbg("Writing out config");
 	save_config();
 
diff --git a/src/metrics.c b/src/metrics.c
new file mode 100644
index 0000000..58cc0cf
--- /dev/null
+++ b/src/metrics.c
@@ -0,0 +1,193 @@
+#include "metrics.h"
+#include "log.h"
+
+#include "microhttpd.h"
+
+prom_counter_t *bytesRecv = NULL;
+prom_counter_t *bytesSent = NULL;
+
+prom_counter_t *meta = NULL;
+prom_counter_t *peers = NULL;
+prom_counter_t *hashes = NULL;
+prom_counter_t *hash_expired = NULL;
+prom_counter_t *queries = NULL;
+prom_counter_t *requests = NULL;
+prom_histogram_t *offered = NULL;
+prom_gauge_t *activeNodes = NULL;
+prom_gauge_t *requestsInFlight = NULL;
+prom_counter_t *retries = NULL;
+
+enum MHD_Result promhttp_handler(
+	void *cls,
+	struct MHD_Connection *connection,
+	const char *url,
+	const char *method,
+	const char *version,
+	const char *upload_data,
+	size_t *upload_data_size,
+	void **con_cls
+) {
+	if (strcmp(method, "GET") != 0) {
+		char *buf = "Invalid HTTP Method\n";
+		struct MHD_Response *response = MHD_create_response_from_buffer(strlen(buf), (void *)buf, MHD_RESPMEM_PERSISTENT);
+		int ret = MHD_queue_response(connection, MHD_HTTP_BAD_REQUEST, response);
+		MHD_destroy_response(response);
+		return ret;
+	}
+
+	if (strcmp(url, "/") == 0) {
+		char *buf = "OK\n";
+		struct MHD_Response *response = MHD_create_response_from_buffer(strlen(buf), (void *)buf, MHD_RESPMEM_PERSISTENT);
+		int ret = MHD_queue_response(connection, MHD_HTTP_OK, response);
+		MHD_destroy_response(response);
+		return ret;
+	}
+
+	if (strcmp(url, "/metrics") == 0) {
+		const char *buf = prom_collector_registry_bridge(PROM_COLLECTOR_REGISTRY_DEFAULT);
+		struct MHD_Response *response = MHD_create_response_from_buffer(strlen(buf), (void *)buf, MHD_RESPMEM_MUST_FREE);
+		int ret = MHD_queue_response(connection, MHD_HTTP_OK, response);
+		MHD_destroy_response(response);
+		return ret;
+	}
+
+	char *buf = "Bad Request\n";
+	struct MHD_Response *response = MHD_create_response_from_buffer(strlen(buf), (void *)buf, MHD_RESPMEM_PERSISTENT);
+	int ret = MHD_queue_response(connection, MHD_HTTP_BAD_REQUEST, response);
+	MHD_destroy_response(response);
+	return ret;
+}
+
+#define PORT 6981
+
+struct MHD_Daemon *mDaemon;
+
+// @CLEAN: We should call this begin
+void metric_init() {
+	if(prom_collector_registry_default_init() != 0) {
+		fatal("Failed to initialize prometheus registry");
+	}
+
+	mDaemon = MHD_start_daemon(MHD_USE_SELECT_INTERNALLY, PORT, NULL, NULL, &promhttp_handler, NULL, MHD_OPTION_END);
+	if(mDaemon == NULL) {
+		fatal("Failed to start http server");
+	}
+	dbg("Metrics server started on port %d", PORT);
+
+	bytesRecv = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"bytes_received",
+			"Number of bytes received",
+			0,
+			NULL
+		)
+	);
+
+	bytesSent = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"bytes_sent",
+			"Number of bytes sent",
+			0,
+			NULL
+		)
+	);
+
+	meta = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"meta",
+			"Meta information value is always 1",
+			1,
+			(const char *[]){ "nodeid" }
+		)
+	);
+
+	peers = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"peers",
+			"Number of total peers",
+			0,
+			NULL
+		)
+	);
+
+	hashes = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"hashes",
+			"Number of hashes currently stored",
+			0,
+			NULL
+		)
+	);
+
+	hash_expired = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"hashes_expired",
+			"Number of hashes expired due to inactivity",
+			0,
+			NULL
+		)
+	);
+
+	queries = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"queries",
+			"Amount of queries handled",
+			1,
+			(const char *[]){"outcome"}
+		)
+	);
+
+	requests = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"requests",
+			"Number of requests sent by us",
+			0,
+			NULL
+		)
+	);
+
+	requestsInFlight = prom_collector_registry_must_register_metric(
+		prom_gauge_new(
+			"requests_in_flight",
+			"Requests currently in flight",
+			0,
+			NULL
+		)
+	);
+
+	retries = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"retries",
+			"Retries sent for unanswered requests",
+			0,
+			NULL
+		)
+	);
+
+	offered = prom_collector_registry_must_register_metric(
+		prom_histogram_new(
+			"offered",
+			"Nodes seen and considered for the routing table",
+			prom_histogram_buckets_linear(0, 1, 157),
+			0,
+			NULL
+		)
+	);
+
+	activeNodes = prom_collector_registry_must_register_metric(
+		prom_gauge_new(
+			"active_nodes",
+			"Nodes active in the routing table",
+			0,
+			NULL
+		)
+	);
+}
+
+void metric_end() {
+	MHD_stop_daemon(mDaemon);
+	if(prom_collector_registry_destroy(PROM_COLLECTOR_REGISTRY_DEFAULT) != 0) {
+		fatal("Failed to destroy the registry");
+	}
+	dbg("Metric server stopped");
+}
diff --git a/src/metrics.h b/src/metrics.h
new file mode 100644
index 0000000..25afcce
--- /dev/null
+++ b/src/metrics.h
@@ -0,0 +1,25 @@
+#pragma once
+
+#include <stdint.h>
+
+#include "prom.h"
+
+extern prom_counter_t *bytesRecv;
+extern prom_counter_t *bytesSent;
+
+extern prom_counter_t *meta;
+
+extern prom_counter_t *peers;
+extern prom_counter_t *hashes;
+extern prom_counter_t *hash_expired;
+extern prom_counter_t *queries;
+extern prom_counter_t *requests;
+extern prom_histogram_t *offered;
+extern prom_gauge_t *activeNodes;
+extern prom_gauge_t *requestsInFlight;
+extern prom_counter_t *retries;
+
+extern prom_counter_t *requestsProcessed;
+
+void metric_init();
+void metric_end();
diff --git a/src/peers.c b/src/peers.c
index 55f1f79..781bc7e 100644
--- a/src/peers.c
+++ b/src/peers.c
@@ -1,6 +1,7 @@
 #include "peers.h"
 
 #include "log.h"
+#include "metrics.h"
 
 #include <stdlib.h>
 #include <string.h>
@@ -24,8 +25,11 @@ int allocate_hashtable() {
 }
 
 static uint64_t hash(struct infohash* key, size_t size) {
-	uint64_t hash = (key->inner[4] << 4) | key->inner[3];
-	return hash % size;
+	uint64_t x = 5381;
+	for(uint8_t i = 0; i < sizeof(key->inner_b); i++) {
+		x = ((x << 5) + x) + key->inner_b[i];
+	}
+	return x % size;
 }
 
 static void find(struct peer_entry* table, size_t size, struct infohash* infohash, struct peer_entry** entry) {
@@ -71,7 +75,7 @@ uint64_t next_pow2(uint64_t x) {
 	return 1 << (64 - leading);
 }
 
-int add_peer(struct infohash* infohash, struct addr* peer) {
+int add_peer(struct infohash* infohash, struct addr* peer, time_t now) {
 	assert(peer_table_load < peer_table_size);
 
 	if(load_factor(peer_table_size, peer_table_load + 1) > 0.75) {
@@ -85,10 +89,13 @@ int add_peer(struct infohash* infohash, struct addr* peer) {
 	if(!entry->set) {
 		entry->set = true;
 		entry->key = *infohash;
+		entry->last_seen = now;
 		peer_table_load++;
 		peer_status.hashes++;
+		prom_counter_inc(hashes, NULL);
 	}
 
+	entry->last_seen = now;
 	size_t peern = entry->value_len;
 	if(peern == PEERS_PER_HASH)
 		return PEER_EFULL;
@@ -96,11 +103,12 @@ int add_peer(struct infohash* infohash, struct addr* peer) {
 	entry->value[peern] = *peer;
 	entry->value_len++;
 	peer_status.peers++;
+	prom_counter_inc(peers, NULL);
 
 	return 0;
 }
 
-void get_peers(struct infohash* infohash, struct addr *peers[PEERS_PER_HASH], size_t *peers_len) {
+void get_peers(struct infohash* infohash, struct addr **peers, size_t *peers_len) {
 	struct peer_entry* entry;
 	find(peer_table, peer_table_size, infohash, &entry);
 
@@ -113,3 +121,40 @@ void get_peers(struct infohash* infohash, struct addr *peers[PEERS_PER_HASH], si
 	*peers = entry->value;
 	*peers_len = entry->value_len;
 }
+
+void expire_hashes(time_t now) {
+	assert(peer_table_load < peer_table_size);
+	for(size_t i = 0; i < peer_table_size; i++) {
+		struct peer_entry *entry = &peer_table[i];
+		if(!entry->set) continue;
+
+		if(difftime(now, entry->last_seen + HASH_TIMEOUT) < 0.0) continue;
+
+		// Find the last hash that collides with us
+		uint64_t entry_hash = hash(&entry->key, peer_table_size);
+		size_t last_in_slot = i;
+		while(true) {
+			size_t next = (last_in_slot + 1) % peer_table_size;
+			assert(next != i);
+
+			// There can't be holes in the chain
+			if(!peer_table[next].set) break;
+
+			// If the two hashes are different we've reached the end of the probe chain
+			uint64_t next_hash = hash(&peer_table[next].key, peer_table_size);
+			if(next_hash != entry_hash) break;
+
+			last_in_slot = next;
+		}
+
+		// If some hashes were chained on us, we copy the last one into our slot
+		if(last_in_slot != i) {
+			*entry = peer_table[last_in_slot];
+			entry = &peer_table[last_in_slot];
+		}
+
+		// Remove the slot
+		entry->set = false;
+		prom_counter_inc(hash_expired, NULL);
+	}
+}
diff --git a/src/peers.h b/src/peers.h
index 106df7b..8bd6b44 100644
--- a/src/peers.h
+++ b/src/peers.h
@@ -5,6 +5,8 @@
 #define PEER_ENOMEM 1
 #define PEER_EFULL 1
 
+#define HASH_TIMEOUT (12 * 60 * 60)
+
 struct infohash {
 	union {
 		uint32_t inner[5];
@@ -17,6 +19,7 @@ struct peer_entry {
 	struct infohash key;
 	struct addr value[PEERS_PER_HASH];
 	size_t value_len;
+	time_t last_seen;
 	bool set;
 };
 
@@ -30,5 +33,7 @@ extern size_t peer_table_size;
 extern size_t peer_table_load;
 
 int allocate_hashtable();
-int add_peer(struct infohash* key, struct addr* peer);
-void get_peers(struct infohash* infohash, struct addr *peers[PEERS_PER_HASH], size_t *peers_len);
+int add_peer(struct infohash* infohash, struct addr* peer, time_t now);
+void get_peers(struct infohash* infohash, struct addr **peers, size_t *peers_len);
+
+void expire_hashes(time_t now);
diff --git a/src/proto.c b/src/proto.c
index 9e1e7aa..b956356 100644
--- a/src/proto.c
+++ b/src/proto.c
@@ -3,6 +3,7 @@
 #include "benc.h"
 #include "query.h"
 #include "log.h"
+#include "metrics.h"
 #include "peers.h"
 
 #include <errno.h>
@@ -244,6 +245,46 @@ int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is
 	return 0;
 }
 
+int send_announce(struct dht* dht, struct nodeid* expected, time_t now, bool node_is_new, const struct sockaddr* dest_addr, socklen_t dest_len, struct msgbuff* msgbuff) {
+	if(*msgbuff->messages >= msgbuff->messages_end)
+		return PROTO_ENOREQ;
+	struct message* message = *msgbuff->messages;
+
+	uint16_t reqId;
+	if(!alloc_req(dht, &reqId)) {
+		return PROTO_ENOREQ;
+	}
+
+	memcpy(&message->dest, dest_addr, dest_len);
+	message->dest_len = dest_len;
+
+	struct ping* data = &dht->requestdata[reqId].cont.ping;
+	if(!node_is_new) {
+		data->remote_id = *expected;
+	} else {
+		expected = &dht->self;
+	}
+	data->is_new = node_is_new;
+	data->attempt = 0;
+
+	dht->requestdata[reqId].fun = &getclient_response;
+	dht->requestdata[reqId].timeout = now + PROTO_TMOUT;
+	dht->requestdata[reqId].timeout_fun = &getclient_timeout;
+	memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len);
+	dht->requestdata[reqId].addr_len = dest_len;
+
+	struct nodeid target = rand_nodeid_in_bucket(&dht->self, expected);
+
+	message->payload_len = sizeof(message->payload);
+	int rc = write_find_node(message->payload, &message->payload_len, &dht->self, &target, reqId);
+	if(rc != 0) {
+		return rc;
+	}
+	(*msgbuff->messages)++;
+
+	return 0;
+}
+
 PROCESS_TIMEOUT(getclient_timeout) {
 	// @HACK: This really sucks. maybe we should just pass in the request id
 	size_t reqId = (typeof(dht->requestdata[0])*)((void*)cont - offsetof(typeof(dht->requestdata[0]), cont)) - dht->requestdata;
@@ -268,9 +309,9 @@ PROCESS_TIMEOUT(getclient_timeout) {
 
 	struct nodeid *remote;
 	if(!cont->ping.is_new) {
-		 remote = &dht->requestdata[reqId].cont.ping.remote_id;
+		remote = &dht->requestdata[reqId].cont.ping.remote_id;
 	} else {
-		 remote = &dht->self;
+		remote = &dht->self;
 	}
 	struct nodeid target = rand_nodeid_in_bucket(&dht->self, remote);
 
@@ -366,6 +407,8 @@ PROCESS_REPONSE(getclient_response) {
 					// Skip the value
 					bcur_next(&bcursor, 1);
 					break;
+				case -BENC_EBADP:
+					fatal("Bad dict");
 			}
 		}
 
@@ -386,6 +429,7 @@ PROCESS_REPONSE(getclient_response) {
 		} else {
 			dbg("We are no longer interested");
 		}
+
 	} else {
 		struct entry* entry = routing_get(&id);
 		// @CLEANUP: Figure out why this can be null. Is the node getting
@@ -424,6 +468,7 @@ PROCESS_REPONSE(getclient_response) {
 }
 
 enum commandType {
+	CT_UNK,
 	CT_QUERY,
 	CT_RESPONSE,
 	CT_ERROR,
@@ -523,19 +568,14 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 		dht->requestdata[reqId].timeout = 0;
 		dht->reqalloc[reqId] = false;
 	} else if(type == CT_QUERY) { // Must be a query
-		if(query == NULL) {
-			err("No query method in request body");
-			return QUERY_EBADQ;
-		}
 		if(transaction == NULL)
 			fatal("No transaction in request");
 		if(transaction_len > 16) {
+			prom_counter_inc(queries, (const char *[]){"discard"});
 			err("DISCARD: Transaction ID is too long");
 			return 0;
 		}
 
-		assert(strlen(query) == query_len);
-
 		assert(*msgbuff->messages < msgbuff->messages_end);
 		struct message* message = *msgbuff->messages;
 
@@ -552,7 +592,7 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 		char respType = 'r';
 		rc = handle_request(&dht->self, &dht->tokens, now, query, (const struct sockaddr*)remote, remote_len, packet, packet_len, &cursor, end-cursor-1);
 		if(rc == QUERY_EUNK) {
-			dbg("Unknown method");
+			prom_counter_inc(queries, (const char *[]){"unknown"});
 			// @FRAGILE: @HACK: Static offsets to fiddle with already written
 			// out packet data. Acceptable because this is the uncommon error
 			// case.
@@ -569,7 +609,7 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 
 			// Use the normal finalize flow
 		} else if(rc == QUERY_EBADQ) {
-			dbg("Invalid query");
+			prom_counter_inc(queries, (const char *[]){"badquery"});
 			// @FRAGILE: @HACK: Static offsets to fiddle with already written
 			// out packet data. Acceptable because this is the uncommon error
 			// case.
@@ -585,7 +625,8 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 			assert(cursor < end);
 
 			// Use the normal finalize flow
-		} else if(rc != 0) fatal("Error handling request");
+		} else if(rc == 0) prom_counter_inc(queries, (const char *[]){"ok"});
+		else fatal("Error handling request");
 
 		rc = snprintf(cursor, end-cursor , "1:t%ld:", transaction_len);
 		if(rc < 0)
@@ -607,7 +648,7 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 	} else if(type == CT_ERROR) {
 		dbg("Unhandled error");
 	} else {
-		fatal("HOW");
+		dbg("Unknown request type");
 	}
 
 	return 0;
@@ -648,6 +689,7 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 			if(difftime(now, dht->requestdata[i].timeout) < 0)
 				continue;
 
+			prom_counter_inc(retries, NULL);
 			int rc = dht->requestdata[i].timeout_fun(dht, &dht->self, now, &dht->requestdata[i].cont, &msgbuff);
 
 			if(rc == PROTO_ENOREQ) {
@@ -691,6 +733,11 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 			routing_oldest(&oldest);
 		}
 
+		// @HACK 0 means unitialized, only happens in tests
+		if(peer_table_size != 0) {
+			expire_hashes(now);
+		}
+
 		recalulate_waketime(dht);
 		return 0;
 	}
@@ -704,8 +751,7 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 	}
 	bcur_next(&bcursor, 1);
 
-	bool discard = false;
-	enum commandType type;
+	enum commandType type = CT_UNK;
 	bool transaction_set = false;
 	char transaction[64];
 	size_t transaction_len;
@@ -751,16 +797,15 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 				memcpy(query, bcursor.readhead->loc, query_len);
 				query[query_len] = '\0';
 				bcur_next(&bcursor, 1);
+				break;
 			}
+			case -BENC_EBADP:
+				fatal("Bad dict");
 		}
 	}
 
 	recalulate_waketime(dht);
 
-	if(discard){
-		return 0;
-	}
-
 	int rc = handle_packet(dht, now, type, transaction_set ? transaction : NULL, transaction_len, query_set ? query : NULL, query_len, buff, recv_len, remote, remote_len, &msgbuff);
 	assert(rc == 0);
 
@@ -768,14 +813,17 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 
 	{
 		printf("In flight |");
+		uint16_t inflight = 0;
 		for(int i = 0; i < MAX_INFLIGHT; i++) {
 			if(dht->reqalloc[i]) {
 				printf("#");
+				inflight++;
 			} else {
 				printf(" ");
 			}
 		}
 		printf("|\n");
+		prom_gauge_set(requestsInFlight, inflight, NULL);
 	}
 	{
 		int filled;
@@ -783,6 +831,7 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 #define LFACLEN 64
 		double load_factor[LFACLEN] = {0};
 		routing_status(&filled, &total, load_factor, LFACLEN);
+		prom_gauge_set(activeNodes, filled, NULL);
 		dbg("%d/%d nodes in routing table", filled, total);
 
 #define GRAPHY 5
@@ -806,9 +855,5 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 #undef LFACLEN
 	}
 
-	{
-		printf("%ld peers in %ld hashes\n", peer_status.peers, peer_status.hashes);
-	}
-
 	return 0;
 }
diff --git a/src/query.c b/src/query.c
index 6cc256a..f0b9014 100644
--- a/src/query.c
+++ b/src/query.c
@@ -10,6 +10,10 @@
 #include <arpa/inet.h>
 
 int handle_request(struct nodeid* self, struct tokens *tokens, time_t now, const char* method, const struct sockaddr* src, socklen_t src_len, const char* packet, size_t packet_len, char** response, size_t response_len) {
+	if(method == NULL) {
+		return QUERY_EBADQ;
+	}
+
 	if(strcmp(method, "ping") == 0) {
 		struct bcursor bcursor;
 		struct benc_node stream[256];
@@ -59,6 +63,8 @@ int handle_request(struct nodeid* self, struct tokens *tokens, time_t now, const
 					// Skip the value
 					bcur_next(&bcursor, 1);
 					break;
+				case -BENC_EBADP:
+					fatal("Bad dict");
 			}
 		}
 
@@ -128,6 +134,8 @@ int handle_request(struct nodeid* self, struct tokens *tokens, time_t now, const
 					// Skip the value
 					bcur_next(&bcursor, 1);
 					break;
+				case -BENC_EBADP:
+					fatal("Bad dict");
 			}
 		}
 
@@ -224,6 +232,8 @@ int handle_request(struct nodeid* self, struct tokens *tokens, time_t now, const
 					// Skip the value
 					bcur_next(&bcursor, 1);
 					break;
+				case -BENC_EBADP:
+					fatal("Bad dict");
 			}
 		}
 
@@ -432,6 +442,8 @@ int handle_request(struct nodeid* self, struct tokens *tokens, time_t now, const
 
 					bcur_next(&bcursor, 1);
 					break;
+				case -BENC_EBADP:
+					fatal("Bad dict");
 			}
 		}
 
@@ -455,7 +467,7 @@ int handle_request(struct nodeid* self, struct tokens *tokens, time_t now, const
 			if(!implied_port) {
 				src_addr.port = htons(port);
 			}
-			int rc = add_peer(&infohash, &src_addr);
+			int rc = add_peer(&infohash, &src_addr, now);
 			if(rc == PEER_EFULL) {
 			} else if(rc != 0) {
 				fatal("Could not add peer (%d)", rc);
diff --git a/src/routing.c b/src/routing.c
index e2f436e..47dee12 100644
--- a/src/routing.c
+++ b/src/routing.c
@@ -1,6 +1,7 @@
 #include "routing.h"
 
 #include "log.h"
+#include "metrics.h"
 
 #include <assert.h>
 #include <limits.h>
@@ -148,6 +149,7 @@ void routing_remove(struct nodeid* id) {
 	struct entry* entry = routing_get(id);
 
 	entry->set = false;
+	entry->expire = 0;
 }
 
 bool routing_interested(struct nodeid* id) {
@@ -156,6 +158,7 @@ bool routing_interested(struct nodeid* id) {
 	if(bucketIndex == RT_IDBITS) {
 		return false;
 	}
+	prom_histogram_observe(offered, bucketIndex, NULL);
 
 	uint16_t baseIndex = base_bucket(id);
 	int8_t inBucketIndex = scan(baseIndex, id);
@@ -246,7 +249,6 @@ void routing_oldest(struct entry** dest) {
 
 	for(struct entry* entry = pTable->table; entry < pTable->table+RT_SIZE; entry++){
 		if(!entry->set)
-			continue;
 
 		if(entry->expire == 0)
 			continue;