heartbeat

commit 4a8eea99501a747ffbde7a3f0f428bd09b02727b
parent 4b1ecb1d6273946d4db6e8b428be24e90fabf484
Author: Remy Noulin <loader2x@gmail.com>
Date:   Tue, 11 Jul 2023 09:02:55 +0200

Limit number of packets per period, print packet content when agent id is wrong

heartbeat.c | 31 ++++++++++++++++++++++---------
1 file changed, 22 insertions(+), 9 deletions(-)

Diffstat:
M
heartbeat.c
 | 
31
++++++++++++++++++++++---------

1 file changed, 22 insertions(+), 9 deletions(-)
diff --git a/heartbeat.c b/heartbeat.c
@@ -869,6 +869,7 @@ void probe(char *cfgfile) {
     setTimeout(bridgesock, period, 0);
     listenToOthers:;
     socklen_t addr_size = sizeof(client);
+    memset(buf, 0, sizeof(buf));
     ssize_t r = recvfrom(bridgesock, buf, sizeof(buf), 0, (struct sockaddr *) &client, &addr_size);
     if (r == -1) {
       // timeout
@@ -887,7 +888,7 @@ void probe(char *cfgfile) {
         //logD("store message");
         if (m->id >= ARRAY_SIZE(agentf)) {
           char *ip = inet_ntoa((client).sin_addr);
-          logE("Invalid id: %d, ip %s", m->id, ip);
+          logE("Invalid id: %d, ip %s, packet size %d, packet content:\n%s", m->id, ip, buf);
         }
         elif (m->id == id) {
           char *ip = inet_ntoa((client).sin_addr);
@@ -902,17 +903,29 @@ void probe(char *cfgfile) {
 
           // update state for this agent
           cleanFinishSmallDictP(agent) = getG(completeCfg, rtSmallDictt, agents[m->id]);
+          // count packets in each period, if more than 2 packets arrive during a period,
+          // something is wrong.
           // Don't measure time between packet because
           // they don't arrive regurlarly, some packets are delayed
           /* TODO use specific period for agent */
-          /* u64 expectedNextTime = getG(agent, rtU64, "mono") + (u64)period * 1000000000UL; */
-          /* time                 = getMonotonicTime(); */
-          /* u64 elapsed          = time - expectedNextTime; */
-          /* if (elapsed < (u64)period * 800000000UL) { */
-          /*   char *ip = inet_ntoa((client).sin_addr); */
-          /*   logE("Packet from %s with ip %s came too early.", agents[m->id], ip); */
-          /* } */
-          /* setG(agent, "mono", time); */
+          u64 expectedNextTime = getG(agent, rtU64, "mono") + (u64)period * 1000000000UL;
+          time                 = getMonotonicTime();
+          if (time > expectedNextTime) {
+            // reset packet counter
+            setG(agent, "c", 0);
+            setG(agent, "mono", time);
+          }
+          else {
+            var c = getG(agent, rtU64P, "c");
+            if (c) {
+              inc *c;
+              if (*c > 2) {
+                char *ip = inet_ntoa((client).sin_addr);
+                logW("Too many packets from agent %s and ip %s, drop packet.", agents[m->id], ip);
+                goto cont;
+              }
+            }
+          }
           bool loggerRunning = hasG(agent, "time");
           setG(agent, "time", getCurrentUnixTime());
           char *newstate = m->messageId ? "alive" : "init";