broker: add overlay.trace streaming RPC

Problem: there is no way to observe the message the broker is
sending and receving on the overlay network.

Add the overlay.trace streaming RPC which can send a stream of
message summaries to a tool.

Author: garlick

src/broker/overlay.c

@@ -191,6 +191,7 @@ struct overlay {
     void *recv_arg;
 
     struct flux_msglist *health_requests;
+    struct flux_msglist *trace_requests;
 };
 
 static void overlay_mcast_child (struct overlay *ov, flux_msg_t *msg);
@@ -203,6 +204,10 @@ static int overlay_control_parent (struct overlay *ov,
                                    int status);
 static void overlay_health_respond_all (struct overlay *ov);
 static struct child *child_lookup_byrank (struct overlay *ov, uint32_t rank);
+static void message_trace (struct overlay *ov,
+                           const char *prefix,
+                           int rank,
+                           const flux_msg_t *msg);
 
 /* Convenience iterator for ov->children
  */
@@ -517,8 +522,11 @@ static int overlay_sendmsg_parent (struct overlay *ov, const flux_msg_t *msg)
         goto done;
     }
     rc = zmqutil_msg_send (ov->parent.zsock, msg);
-    if (rc == 0)
+    if (rc == 0) {
         ov->parent.lastsent = flux_reactor_now (ov->reactor);
+        if (flux_msglist_count (ov->trace_requests) > 0)
+            message_trace (ov, "tx", ov->parent.rank, msg);
+    }
 done:
     return rc;
 }
@@ -811,6 +819,22 @@ static int overlay_sendmsg_child (struct overlay *ov, const flux_msg_t *msg)
         }
         errno = saved_errno;
     }
+    if (rc == 0 && flux_msglist_count (ov->trace_requests) > 0) {
+        const char *uuid;
+        struct child *child = NULL;
+        int rank = -1;
+        int type = 0;
+
+        (void)flux_msg_get_type (msg, &type);
+
+        // N.B. events are traced in overlay_mcast_child()
+        if (type != FLUX_MSGTYPE_EVENT) {
+            if ((uuid = flux_msg_route_last (msg))
+                && (child = child_lookup_online (ov, uuid)))
+                rank = child->rank;
+            message_trace (ov, "tx", rank, msg);
+        }
+    }
 done:
     return rc;
 }
@@ -831,6 +855,7 @@ static int overlay_mcast_child_one (struct overlay *ov,
 static void overlay_mcast_child (struct overlay *ov, flux_msg_t *msg)
 {
     struct child *child;
+    int count = 0;
 
     flux_msg_route_enable (msg);
 
@@ -843,8 +868,12 @@ static void overlay_mcast_child (struct overlay *ov, flux_msg_t *msg)
                                     (unsigned long)child->rank);
                 }
             }
+            else
+                count++;
         }
     }
+    if (count > 0 && flux_msglist_count (ov->trace_requests) > 0)
+        message_trace (ov, "tx", -1, msg);
 }
 
 static void logdrop (struct overlay *ov,
@@ -891,6 +920,67 @@ static int clear_msg_role (flux_msg_t *msg, uint32_t role)
     return 0;
 }
 
+static void message_trace (struct overlay *ov,
+                           const char *prefix,
+                           int rank,
+                           const flux_msg_t *msg)
+{
+    const flux_msg_t *req;
+    double now = flux_reactor_now (ov->reactor);
+    int type = 0;
+    char buf[64];
+    const char *topic = NULL;
+    int payload_size = 0;
+
+    (void)flux_msg_get_type (msg, &type);
+    if (type == FLUX_MSGTYPE_CONTROL) {
+        int ctype;
+        int cstatus;
+        if (flux_control_decode (msg, &ctype, &cstatus) == 0) {
+            snprintf (buf,
+                      sizeof (buf),
+                      "%s %d",
+                      ctype == CONTROL_HEARTBEAT ? "heartbeat" :
+                      ctype == CONTROL_STATUS ? "status" :
+                      ctype == CONTROL_DISCONNECT ? "disconnect" : "unknown",
+                      cstatus);
+            topic = buf;
+        }
+    }
+    else {
+        (void)flux_msg_get_topic (msg, &topic);
+        (void)flux_msg_get_payload (msg, NULL, &payload_size);
+    }
+
+    req = flux_msglist_first (ov->trace_requests);
+    while (req) {
+        struct flux_match match = FLUX_MATCH_ANY;
+        int nodeid;
+        if (flux_request_unpack (req,
+                                 NULL,
+                                 "{s:i s:s s:i}",
+                                 "typemask", &match.typemask,
+                                 "topic_glob", &match.topic_glob,
+                                 "nodeid", &nodeid) < 0
+            || (nodeid != FLUX_NODEID_ANY && nodeid != rank)
+            || !flux_msg_cmp (msg, match))
+            goto next;
+
+        if (flux_respond_pack (ov->h,
+                               req,
+                               "{s:f s:s s:i s:i s:s s:i}",
+                               "timestamp", now,
+                               "prefix", prefix,
+                               "rank", rank,
+                               "type", type,
+                               "topic", topic ? topic : "NO-TOPIC",
+                               "payload_size", payload_size) < 0)
+            flux_log_error (ov->h, "error responding to overlay.trace");
+next:
+        req = flux_msglist_next (ov->trace_requests);
+    }
+}
+
 /* Handle a message received from TBON child (downstream).
  */
 static void child_cb (flux_reactor_t *r,
@@ -929,6 +1019,7 @@ static void child_cb (flux_reactor_t *r,
             && flux_msg_get_topic (msg, &topic) == 0
             && streq (topic, "overlay.hello")
             && !ov->shutdown_in_progress) {
+            message_trace (ov, "rx", -1, msg);
             hello_request_handler (ov, msg);
         }
         /* Or one of the following cases occurred that requires (or at least
@@ -956,8 +1047,10 @@ static void child_cb (flux_reactor_t *r,
         case FLUX_MSGTYPE_CONTROL: {
             int type, status;
             if (flux_control_decode (msg, &type, &status) == 0
-                && type == CONTROL_STATUS)
+                && type == CONTROL_STATUS) {
+                message_trace (ov, "rx", child->rank, msg);
                 overlay_child_status_update (ov, child, status, NULL);
+            }
             goto done;
         }
         case FLUX_MSGTYPE_REQUEST:
@@ -975,6 +1068,8 @@ static void child_cb (flux_reactor_t *r,
         case FLUX_MSGTYPE_EVENT:
             break;
     }
+    if (flux_msglist_count (ov->trace_requests) > 0)
+        message_trace (ov, "rx", child->rank, msg);
     if (ov->recv_cb (&msg, OVERLAY_DOWNSTREAM, ov->recv_arg) < 0)
         goto done;
     return;
@@ -1064,6 +1159,9 @@ static void parent_cb (flux_reactor_t *r,
         default:
             break;
     }
+    if (flux_msglist_count (ov->trace_requests) > 0) {
+        message_trace (ov, "rx", ov->parent.rank, msg);
+    }
     if (ov->recv_cb (&msg, OVERLAY_UPSTREAM, ov->recv_arg) < 0)
         goto done;
     return;
@@ -1731,6 +1829,10 @@ static void disconnect_cb (flux_t *h,
         flux_log_error (h, "error handling overlay.disconnect");
     if (count > 0)
         flux_log (h, LOG_DEBUG, "overlay: goodbye to %d health clients", count);
+    if ((count = flux_msglist_disconnect (ov->trace_requests, msg)) < 0)
+        flux_log_error (h, "error handling overlay.disconnect");
+    if (count > 0)
+        flux_log (h, LOG_DEBUG, "overlay: goodbye to %d trace clients", count);
 }
 
 const char *overlay_get_subtree_status (struct overlay *ov, int rank)
@@ -1851,6 +1953,34 @@ static void overlay_disconnect_subtree_cb (flux_t *h,
         flux_log_error (h, "error responding to overlay.disconnect-subtree");
 }
 
+static void overlay_trace_cb (flux_t *h,
+                              flux_msg_handler_t *mh,
+                              const flux_msg_t *msg,
+                              void *arg)
+{
+    struct overlay *ov = arg;
+    struct flux_match match = FLUX_MATCH_ANY;
+    int nodeid;
+
+    if (flux_request_unpack (msg,
+                             NULL,
+                             "{s:i s:s s:i}",
+                             "typemask", &match.typemask,
+                             "topic_glob", &match.topic_glob,
+                             "nodeid", &nodeid) < 0)
+        goto error;
+    if (!flux_msg_is_streaming (msg)) {
+        errno = EPROTO;
+        goto error;
+    }
+    if (flux_msglist_append (ov->trace_requests, msg) < 0)
+        goto error;
+    return;
+error:
+    if (flux_respond_error (h, msg, errno, NULL) < 0)
+        flux_log_error (h, "error responding to overlay.trace");
+}
+
 int overlay_cert_load (struct overlay *ov, const char *path)
 {
     struct stat sb;
@@ -2231,6 +2361,7 @@ void overlay_destroy (struct overlay *ov)
                 free (mon);
             zlist_destroy (&ov->monitor_callbacks);
         }
+        flux_msglist_destroy (ov->trace_requests);
         topology_decref (ov->topo);
         if (!ov->zctx_external)
             zmq_ctx_term (ov->zctx);
@@ -2240,6 +2371,12 @@ void overlay_destroy (struct overlay *ov)
 }
 
 static const struct flux_msg_handler_spec htab[] = {
+    {
+        FLUX_MSGTYPE_REQUEST,
+        "overlay.trace",
+        overlay_trace_cb,
+        0,
+    },
     {
         FLUX_MSGTYPE_REQUEST,
         "overlay.stats-get",
@@ -2358,7 +2495,8 @@ struct overlay *overlay_create (flux_t *h,
         goto error;
     if (!(ov->cert = cert_create ()))
         goto nomem;
-    if (!(ov->health_requests = flux_msglist_create ()))
+    if (!(ov->health_requests = flux_msglist_create ())
+        || !(ov->trace_requests = flux_msglist_create ()))
         goto error;
     if (!(ov->parent.f_goodbye = flux_future_create (NULL, NULL)))
         goto error;