From 04c5b9d5ef4723d469ea3472012787a8e2a5bdbd Mon Sep 17 00:00:00 2001 From: Jesper Jensen Date: Sat, 12 Apr 2025 09:35:32 +0200 Subject: Add prometheus monitoring --- Makefile | 2 +- src/benc.c | 16 ++--- src/benc.h | 1 + src/log.h | 6 +- src/main.c | 92 +++++++++++++++++++++++----- src/metrics.c | 193 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/metrics.h | 25 ++++++++ src/peers.c | 53 ++++++++++++++-- src/peers.h | 9 ++- src/proto.c | 89 ++++++++++++++++++++------- src/query.c | 14 ++++- src/routing.c | 4 +- test/benc.c | 4 +- test/peers.c | 61 +++++++++++++++---- test/proto.c | 143 ++++++++++++++++++++++++++++++++++++++++++- 15 files changed, 639 insertions(+), 73 deletions(-) create mode 100644 src/metrics.c create mode 100644 src/metrics.h diff --git a/Makefile b/Makefile index 9ffc7e0..1c04ac4 100644 --- a/Makefile +++ b/Makefile @@ -5,7 +5,7 @@ TSTDIR ?= test GENDIR ?= gen OBJDIR ?= obj -LIBS = -lm +LIBS = -lm -lmicrohttpd -lprom INCS = -Isrc/ -Igen/ -I. -Ithirdparty/crypto-algorithms/ CFLAGS ?= -O3 -D_FORTIFY_SOURCE=2 -Wall -g 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 #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 #include #include #include +#include +#include + +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 + +#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 #include @@ -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 @@ -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 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 #include @@ -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; diff --git a/test/benc.c b/test/benc.c index a9d6ddb..3916f87 100644 --- a/test/benc.c +++ b/test/benc.c @@ -262,7 +262,7 @@ void test_key_not_found() { ssize_t found = bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING}, (const char*[]){"a"}, (const size_t[]){1}, 1); - TEST_ASSERT_EQUAL(-1, found); + TEST_ASSERT_EQUAL(-BENC_EENDP, found); TEST_ASSERT_EQUAL_PTR(stream+3, bcursor.readhead); } @@ -351,7 +351,7 @@ void test_stop_at_end() { ssize_t found = bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING}, (const char*[]){"x"}, (const size_t[]){1}, 1); - TEST_ASSERT_EQUAL_MESSAGE(-1, found, "Found something"); + TEST_ASSERT_EQUAL_MESSAGE(-BENC_EENDP, found, "Found something"); TEST_ASSERT_EQUAL_PTR_MESSAGE(stream+4, bcursor.readhead, "Didn't stop at dict end"); } diff --git a/test/peers.c b/test/peers.c index 1e9a454..49bf6c1 100644 --- a/test/peers.c +++ b/test/peers.c @@ -13,7 +13,7 @@ void test_add_single_peer() { struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; - int rc = add_peer(&sometorrent, &addr); + int rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); } @@ -23,24 +23,24 @@ void test_add_9_peers_same_torrent() { struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; - int rc = add_peer(&sometorrent, &addr); + int rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); - rc = add_peer(&sometorrent, &addr); + rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); - rc = add_peer(&sometorrent, &addr); + rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); - rc = add_peer(&sometorrent, &addr); + rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); - rc = add_peer(&sometorrent, &addr); + rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); - rc = add_peer(&sometorrent, &addr); + rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); - rc = add_peer(&sometorrent, &addr); + rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); - rc = add_peer(&sometorrent, &addr); + rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); // There's room for 8 peers per infohash - rc = add_peer(&sometorrent, &addr); + rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(PEER_EFULL, rc); } @@ -52,9 +52,9 @@ void test_peers_for_only_torrent() { struct addr someaddr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; struct addr otheraddr = (struct addr){.ip = IP(128,0,0,1), .port = 1}; - int rc = add_peer(&sometorrent, &someaddr); + int rc = add_peer(&sometorrent, &someaddr, 0); TEST_ASSERT_EQUAL(0, rc); - rc = add_peer(&othertorrent, &otheraddr); + rc = add_peer(&othertorrent, &otheraddr, 0); TEST_ASSERT_EQUAL(0, rc); struct addr* found; @@ -88,7 +88,7 @@ void test_grows() { for(int i = 0; i < 17; i++) { // Peer table size + 1 to make sure it HAS to grow struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, i}}; - int rc = add_peer(&sometorrent, &addr); + int rc = add_peer(&sometorrent, &addr, 0); TEST_ASSERT_EQUAL(0, rc); } @@ -102,3 +102,38 @@ void test_grows() { TEST_ASSERT_EQUAL(1, found_len); } } + +void test_expired() { + allocate_hashtable(); + + struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; + struct infohash other = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008470, 0x0ab00521}}; + struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; + struct addr other_addr = (struct addr){.ip = IP(128,0,0,1), .port = 1}; + + int rc = add_peer(&sometorrent, &addr, 0); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&other, &addr, 0); + TEST_ASSERT_EQUAL(0, rc); + + rc = add_peer(&other, &other_addr, 1); + TEST_ASSERT_EQUAL(0, rc); + + expire_hashes(HASH_TIMEOUT + 1); + + { + struct addr* found; + size_t found_len; + get_peers(&sometorrent, &found, &found_len); + TEST_ASSERT_NULL(found); + TEST_ASSERT_EQUAL(0, found_len); + } + + { + struct addr* found; + size_t found_len; + get_peers(&other, &found, &found_len); + TEST_ASSERT_NOT_NULL(found); + TEST_ASSERT_EQUAL(2, found_len); + } +} diff --git a/test/proto.c b/test/proto.c index c69a293..54cf158 100644 --- a/test/proto.c +++ b/test/proto.c @@ -72,6 +72,8 @@ void test_begin_pings_bootstrap_node() { TEST_ASSERT_EQUAL(91, outbuff[0].payload_len); TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:ad2:id20:BBBBBBBBBBBBBBBBBBBB6:target20:", outbuff[0].payload, 43); TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:q9:find_node1:t1:01:y1:qe", outbuff[0].payload+63, 28); + + proto_end(&dht); } void test_response_from_initial_probe() { @@ -112,6 +114,8 @@ void test_response_from_initial_probe() { TEST_ASSERT_EQUAL(((struct sockaddr_in*)&remote)->sin_addr.s_addr, entry->addr.ip); TEST_ASSERT_EQUAL(((struct sockaddr_in*)&remote)->sin_port, entry->addr.port); TEST_ASSERT_EQUAL(PROTO_UNCTM+10, entry->expire); + + proto_end(&dht); } void test_reponse_from_wrong_ip() { @@ -145,6 +149,8 @@ void test_reponse_from_wrong_ip() { struct nodeid other = (struct nodeid){.inner={0x61616161, 0x61616161, 0x61616161, 0x61616161, 0x61616161}}; struct entry* entry = routing_get(&other); TEST_ASSERT_NULL(entry); + + proto_end(&dht); } void test_ping() { @@ -172,6 +178,8 @@ void test_ping() { TEST_ASSERT_EQUAL(47, outbuff[0].payload_len); TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:rd2:id20:BBBBBBBBBBBBBBBBBBBBe1:t2:aa1:y1:re", outbuff[0].payload, 47); + + proto_end(&dht); } void test_unknown_method() { @@ -199,6 +207,8 @@ void test_unknown_method() { TEST_ASSERT_EQUAL(42, outbuff[0].payload_len); TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:eli204e14:Unknown Methode1:t2:aa1:y1:ee", outbuff[0].payload, 42); + + proto_end(&dht); } void test_note_times_out() { @@ -239,6 +249,8 @@ void test_note_times_out() { TEST_ASSERT_GREATER_THAN(19, prefix(&dht.self, &target)); TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:q9:find_node1:t1:01:y1:qe", outbuff[1].payload+63, 28); + + proto_end(&dht); } void test_response_after_retry() { @@ -266,6 +278,8 @@ void test_response_after_retry() { TEST_ASSERT_EQUAL_MEMORY_MESSAGE(&dht.self, &target, sizeof(struct nodeid), "Target should be our own id"); TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:q9:find_node1:t1:01:y1:qe", outbuff[0].payload+63, 28); + + proto_end(&dht); } void test_remove_from_routing_after_3_retries() { @@ -330,6 +344,8 @@ void test_remove_from_routing_after_3_retries() { TEST_ASSERT_EQUAL_PTR_MESSAGE(message_cursor, outbuff, "The timeout should send not a message"); entry = routing_get(&other); TEST_ASSERT_NULL(entry); + + proto_end(&dht); } void test_ping_node_when_uncertain() { @@ -387,6 +403,8 @@ void test_ping_node_when_uncertain() { TEST_ASSERT_NOT_NULL(entry); TEST_ASSERT_GREATER_THAN(now, entry->expire); } + + proto_end(&dht); } void test_query_find_node() { @@ -438,6 +456,8 @@ void test_query_find_node() { TEST_ASSERT_EQUAL_MEMORY(&((struct sockaddr_in*)&remote)->sin_port, outbuff[0].payload+66, 2); TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:t2:aa1:y1:re", outbuff[0].payload+68, 15); } + + proto_end(&dht); } void test_query_get_peers_have_one() { @@ -513,7 +533,6 @@ void test_query_get_peers_have_one() { // Node announces that it's a peer for that torrent char buff[] = "d1:ad2:id20:abcdefghij012345678912:implied_porti1e9:info_hash20:aaaaaaaaaaaaaaaaaaaa4:porti1337e5:token32: e1:q13:announce_peer1:t2:aa1:y1:qe"; memcpy(buff+106, token, SHA256_BLOCK_SIZE); - dbg("PKT %s", buff); struct message* message_cursor = outbuff; proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&other, sizeof(other), now, &message_cursor, outbuff+2); TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); @@ -548,4 +567,126 @@ void test_query_get_peers_have_one() { TEST_ASSERT_EQUAL_CHAR_ARRAY("6:valuesl6:\x80\x00\x00\x01\x23\x82""ee1:t2:aa1:y1:re", cursor, 33); cursor+=33; } + + proto_end(&dht); +} + +void test_node_closer_to_infohash_is_discovered() { + struct message outbuff[10] = {0}; + time_t now = 0; + + struct sockaddr_storage remote; + socklen_t remote_len; + + allocate_hashtable(); + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + routing_init(&dht.self); + struct nodeid other = (struct nodeid){.inner={0x30303030, 0x30303030, 0x30303030, 0x30303030, 0x3030303}}; + + { + struct message* message_cursor = outbuff; + proto_begin(&dht, 0, &message_cursor, outbuff+2); + remote_len = outbuff[0].dest_len; + memcpy(&remote, &outbuff[0].dest, remote_len); + } + now += 10; + + // Some other then asks for peers + char token[SHA256_BLOCK_SIZE]; + { + struct sockaddr_in other; + other.sin_family = AF_INET; + inet_pton(AF_INET, "128.0.0.1", &other.sin_addr.s_addr); + other.sin_port = htons(3); + + // Node asks for peers to get token + char buff[] = "d1:ad2:id20:000000000000000000009:info_hash20:00000000000000000002e1:q9:get_peers1:t2:aa1:y1:qe"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), &other, sizeof(other), 0, &message_cursor, outbuff+2); + + // We should have sent a response + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + + TEST_ASSERT_EQUAL(98, outbuff[0].payload_len); + char* cursor = outbuff[0].payload; + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:rd2:id20:BBBBBBBBBBBBBBBBBBBB5:nodes0:", cursor, 41); + cursor+=41; + TEST_ASSERT_EQUAL_CHAR_ARRAY("5:token32:", cursor, 10); + cursor+=10; + memcpy(token, cursor, SHA256_BLOCK_SIZE); + cursor+=SHA256_BLOCK_SIZE; + TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:t2:aa1:y1:re", cursor, 15); + cursor+=15; + } + now += 1; + + // Using the token from before that node then announces that it's a peer + { + struct sockaddr_in other; + other.sin_family = AF_INET; + inet_pton(AF_INET, "128.0.0.1", &other.sin_addr.s_addr); + other.sin_port = htons(3); + + // Node announces that it's a peer for that torrent + char buff[] = "d1:ad2:id20:0000000000000000000012:implied_porti1e9:info_hash20:000000000000000000024:porti1337e5:token32: e1:q13:announce_peer1:t2:aa1:y1:qe"; + memcpy(buff+106, token, SHA256_BLOCK_SIZE); + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&other, sizeof(other), now, &message_cursor, outbuff+2); + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + + TEST_ASSERT_EQUAL(47, outbuff[0].payload_len); + char* cursor = outbuff[0].payload; + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:rd2:id20:BBBBBBBBBBBBBBBBBBBBe1:t2:aa1:y1:re", cursor, 47); + cursor+=47; + } + now += 1; + + // The initial node responds with an id that happens to be closer to the + // announced infohash than we are. We should reannounce that infohash to + // it. + { + char buff[] = "d1:y1:r1:t1:01:rd2:id20:000000000000000000015:nodes0:ee"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+2); + + +// Disable this part since we don't currently implementing this functionality. +// It's technically part of the Kademlia spec, but I don't see how you can +// implement it in DHT +#if 0 + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + TEST_ASSERT_EQUAL(94, outbuff[0].payload_len); + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:ad2:id20:BBBBBBBBBBBBBBBBBBBB9:info_hash20:00000000000000000002e1:q9:get_peers1:t2:aa1:y:qe", outbuff[0].payload, 94); +#else + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff); +#endif + } + +#if 0 + struct entry* entry = routing_get(&other); + TEST_ASSERT_NOT_NULL(entry); + { + struct sockaddr_in other; + other.sin_family = AF_INET; + inet_pton(AF_INET, "255.255.255.255", &other.sin_addr.s_addr); + other.sin_port = htons(6881); + + char buff[] = "d1:ad2:id20:abcdefghij01234567896:target20:aaaaaaaaaaaaaaaaaaaae1:q9:find_node1:t2:aa1:y1:qe"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), &other, sizeof(other), 0, &message_cursor, outbuff+2); + + // We should have sent a response + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + + TEST_ASSERT_EQUAL(83, outbuff[0].payload_len); + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:rd2:id20:BBBBBBBBBBBBBBBBBBBB5:nodes26:aaaaaaaaaaaaaaaaaaaa", outbuff[0].payload, 62); + TEST_ASSERT_EQUAL_MEMORY(&((struct sockaddr_in*)&remote)->sin_addr.s_addr, outbuff[0].payload+62, 4); + TEST_ASSERT_EQUAL_MEMORY(&((struct sockaddr_in*)&remote)->sin_port, outbuff[0].payload+66, 2); + TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:t2:aa1:y1:re", outbuff[0].payload+68, 15); + } +#endif + + proto_end(&dht); } -- cgit v1.2.3