Fixes to heartbeats and NEB socket status logging

Switch to a 32-bit sequence number in heartbeats.

Future versions of ZeroMQ get rid of zmq_event_t, and we're already
doing the right thing, just wrapping it up in zmq_event_t. Get rid
of this weird opaque structure and future proof the socket status
logging code.

Author: jbreams

.gitignore

@@ -3,8 +3,10 @@
 *.a
 *.i
 *.s
+*.tar.gz
 .libs
 .deps
+m4/*
 Makefile
 Makefile.in
 aclocal.m4

dnxmq/heartbeats.c

@@ -8,7 +8,7 @@
 #include <time.h>
 #include "mqexec.h"
 
-int64_t last_sent_seq = -1, last_recv_seq = -1;
+int32_t last_sent_seq = -1, last_recv_seq = -1;
 int32_t heartbeat_interval = -1, heartbeat_curr_interval = -1;
 time_t last_heartbeat = 0;
 char heartbeat_reply_string[255];
@@ -55,7 +55,7 @@ void subscribe_heartbeat(void * sock) {
 
 void process_heartbeat(json_t * inputmsg) {
 	char * replyto;
-	int64_t sequence;
+	int32_t sequence;
 	if(json_unpack(inputmsg, "{ s:s s:i }",
 		"pong_target", &replyto,
 		"sequence", &sequence) != 0) {
@@ -93,14 +93,15 @@ void heartbeat_timeout_cb(struct ev_loop * loop, ev_timer * t, int event) {
 	}
 	heartbeat_curr_interval = heartbeat_interval;
 
-	if(last_recv_seq == last_sent_seq -1) {
+	if(last_recv_seq == last_sent_seq) {
 		logit(DEBUG, "Heartbeat didn't time out! Resetting timer.");
 		send_heartbeat(loop);
 		return;
 	}
 
 	if(last_recv_seq != -1) {
-		logit(DEBUG, "We recieved a pong message, but it wasn't right. Retrying.");
+		logit(DEBUG, "We recieved a pong message, but it wasn't right. Retrying. (%08x != %08x)",
+			last_recv_seq, last_sent_seq, last_recv_seq);
 		send_heartbeat(loop);
 		return;
 	}
@@ -162,7 +163,7 @@ void send_heartbeat(struct ev_loop * loop) {
 
 	output = json_pack("{s:s s:i s:s}",
 		"type", "ping",
-		"sequence", last_sent_seq++,
+		"sequence", ++last_sent_seq,
 		"replyto", heartbeat_reply_string
 	);
 	if(output == NULL) {

dnxmq/mqexec.c

@@ -250,7 +250,8 @@ void recv_job_cb(struct ev_loop * loop, ev_io * i, int event) {
 void sock_monitor_cb(struct ev_loop * loop, ev_io * i, int event) {
 	while(1) {
 		void * sock = i->data;
-		zmq_event_t sockevent;
+		uint16_t event;
+		int32_t value;
 		zmq_msg_t addrmsg, eventmsg;
 		int rc, shouldlog = 1;
 
@@ -274,9 +275,8 @@ void sock_monitor_cb(struct ev_loop * loop, ev_io * i, int event) {
 		}
 
 		const char* eventdata = (char*)zmq_msg_data(&eventmsg);
-		memcpy(&(sockevent.event), eventdata, sizeof(sockevent.event));
-		memcpy(&(sockevent.value), eventdata + sizeof(sockevent.event),
-			sizeof(sockevent.value));
+		memcpy(&event, eventdata, sizeof(event));
+		memcpy(&value, eventdata + sizeof(event), sizeof(value));
 		zmq_msg_close(&eventmsg);
 
 		zmq_msg_init(&addrmsg);
@@ -293,14 +293,14 @@ void sock_monitor_cb(struct ev_loop * loop, ev_io * i, int event) {
 			}
 		}
 
-		if(sockevent.event == 0) {
+		if(event == 0) {
 			zmq_close(sock);
 			ev_io_stop(loop, i);
 			return;
 		}
 
 		// These are super chatting log messages, skip em.
-		switch(sockevent.event) {
+		switch(event) {
 			case ZMQ_EVENT_CLOSED:
 			case ZMQ_EVENT_CONNECT_DELAYED:
 				shouldlog = 0;
@@ -313,7 +313,7 @@ void sock_monitor_cb(struct ev_loop * loop, ev_io * i, int event) {
 		}
 
 		char * event_string;
-		switch(sockevent.event) {
+		switch(event) {
 			case ZMQ_EVENT_CONNECTED:
 				event_string = "Socket event on %.*s: connection established (fd: %d)";
 				if(i == &pullmonio)
@@ -360,7 +360,7 @@ void sock_monitor_cb(struct ev_loop * loop, ev_io * i, int event) {
 		}
 
 		logit(INFO, event_string, zmq_msg_size(&addrmsg),
-			(char*)zmq_msg_data(&addrmsg), sockevent.value);
+			(char*)zmq_msg_data(&addrmsg), value);
 		zmq_msg_close(&addrmsg);
 	}
 }

mods/nagmq_pull.c

@@ -26,7 +26,7 @@ extern int errno;
 
 static void process_ping(json_t * payload) {
 	char * target;
-	int64_t seq;
+	int32_t seq;
 	char * extra = NULL;
 	struct timeval curtime;
 

mods/socketstatus.c

@@ -10,9 +10,10 @@ int pullmonfd = -1, reqmonfd = -1, pubmonfd;
 #if ZMQ_VERSION_MAJOR >= 3 && defined(HAVE_NAGIOS4)
 int sock_monitor_cb(int sd, int events, void * sock) {
 	while(1) {
-		zmq_event_t sockevent;
 		zmq_msg_t addrmsg, eventmsg;
 		int rc, shouldlog = 1;
+		uint16_t event;
+		int32_t value;
 
 		zmq_msg_init(&eventmsg);
 		rc = zmq_msg_recv(&eventmsg, sock, ZMQ_DONTWAIT);
@@ -34,9 +35,8 @@ int sock_monitor_cb(int sd, int events, void * sock) {
 		}
 
 		const char* eventdata = (char*)zmq_msg_data(&eventmsg);
-		memcpy(&(sockevent.event), eventdata, sizeof(sockevent.event));
-		memcpy(&(sockevent.value), eventdata + sizeof(sockevent.event),
-			sizeof(sockevent.value));
+		memcpy(&event, eventdata, sizeof(event));
+		memcpy(&value, eventdata + sizeof(event), sizeof(value));
 		zmq_msg_close(&eventmsg);
 
 		zmq_msg_init(&addrmsg);
@@ -54,7 +54,7 @@ int sock_monitor_cb(int sd, int events, void * sock) {
 		}
 
 		// These are super chatting log messages, skip em.
-		switch(sockevent.event) {
+		switch(event) {
 			case ZMQ_EVENT_CLOSED:
 			case ZMQ_EVENT_CONNECT_DELAYED:
 				shouldlog = 0;
@@ -67,7 +67,7 @@ int sock_monitor_cb(int sd, int events, void * sock) {
 		}
 
 		char * event_string;
-		switch(sockevent.event) {
+		switch(event) {
 			case ZMQ_EVENT_CONNECTED:
 				event_string = "NagMQ socket event on %.*s: connection established (fd: %d)";
 				break;
@@ -106,7 +106,7 @@ int sock_monitor_cb(int sd, int events, void * sock) {
 		}
 
 		logit(NSLOG_INFO_MESSAGE, TRUE, event_string, zmq_msg_size(&addrmsg),
-			(char*)zmq_msg_data(&addrmsg), sockevent.value);
+			(char*)zmq_msg_data(&addrmsg), value);
 		zmq_msg_close(&addrmsg);
 	}
 	return 0;