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 --- src/main.c | 92 ++++++++++++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 77 insertions(+), 15 deletions(-) (limited to 'src/main.c') 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(); -- cgit v1.2.3