cleanup(userspace/libsinsp): call sinsp_observer methods after an event has been processed by all parsers

userspace/libsinsp/parsers.cpp

@@ -1276,9 +1276,14 @@
 		return;
 	}
 
-	/* If there's a listener, invoke it */
+	//
+	// If there's a listener, add a callback to later invoke it.
+	//
 	if(m_inspector->get_observer()) {
-		m_inspector->get_observer()->on_clone(evt, new_child.get(), tid_collision);
+		m_inspector->m_post_process_cbs.emplace(
+		        [new_child, tid_collision](sinsp_observer *observer, sinsp_evt *evt) {
+			        observer->on_clone(evt, new_child.get(), tid_collision);
+		        });
 	}
 
 	/* If we had to erase a previous entry for this tid and rebalance the table,
@@ -1292,8 +1297,6 @@
 		               new_child->m_comm.c_str());
 	}
 	/*=============================== ADD THREAD TO THE TABLE ===========================*/
-
-	return;
 }
 
 void sinsp_parser::parse_clone_exit_child(sinsp_evt *evt) {
@@ -1761,17 +1764,19 @@
 	evt->set_tinfo(new_child.get());
 
 	//
-	// If there's a listener, invoke it
+	// If there's a listener, add a callback to later invoke it.
 	//
 	if(m_inspector->get_observer()) {
-		m_inspector->get_observer()->on_clone(evt, new_child.get(), tid_collision);
+		m_inspector->m_post_process_cbs.emplace(
+		        [new_child, tid_collision](sinsp_observer *observer, sinsp_evt *evt) {
+			        observer->on_clone(evt, new_child.get(), tid_collision);
+		        });
 	}
 
 	/* If we had to erase a previous entry for this tid and rebalance the table,
 	 * make sure we reinitialize the child_tinfo pointer for this event, as the thread
 	 * generating it might have gone away.
 	 */
-
 	if(tid_collision != -1) {
 		reset(evt);
 		/* Right now we have collisions only on the clone() caller */
@@ -1781,7 +1786,6 @@
 	}
 
 	/*=============================== CREATE NEW THREAD-INFO ===========================*/
-	return;
 }
 
 void sinsp_parser::parse_clone_exit(sinsp_evt *evt) {
@@ -2230,15 +2234,11 @@
 	evt->get_tinfo()->m_flags |= PPM_CL_NAME_CHANGED;
 
 	//
-	// Recompute the program hash
-	//
-	evt->get_tinfo()->compute_program_hash();
-
-	//
-	// If there's a listener, invoke it
+	// If there's a listener, add a callback to later invoke it.
 	//
 	if(m_inspector->get_observer()) {
-		m_inspector->get_observer()->on_execve(evt);
+		m_inspector->m_post_process_cbs.emplace(
+		        [](sinsp_observer *observer, sinsp_evt *evt) { observer->on_execve(evt); });
 	}
 
 	/* If any of the threads in a thread group performs an
@@ -2787,10 +2787,11 @@
 	evt->get_fd_info()->m_name = evt->get_param_as_str(1, &parstr, sinsp_evt::PF_SIMPLE);
 
 	//
-	// If there's a listener callback, invoke it
+	// If there's a listener, add a callback to later invoke it.
 	//
 	if(m_inspector->get_observer()) {
-		m_inspector->get_observer()->on_bind(evt);
+		m_inspector->m_post_process_cbs.emplace(
+		        [](sinsp_observer *observer, sinsp_evt *evt) { observer->on_bind(evt); });
 	}
 }
 
@@ -2982,7 +2983,6 @@
 
 void sinsp_parser::parse_connect_exit(sinsp_evt *evt) {
 	const sinsp_evt_param *parinfo;
-	uint8_t *packed_data;
 	int64_t retval;
 	int64_t fd;
 	bool force_overwrite_stale_data = false;
@@ -3048,15 +3048,18 @@
 		return;
 	}
 
-	packed_data = (uint8_t *)parinfo->m_val;
+	uint8_t *packed_data = (uint8_t *)parinfo->m_val;
 
 	fill_client_socket_info(evt, packed_data, force_overwrite_stale_data);
 
 	//
-	// If there's a listener callback, invoke it
+	// If there's a listener, add a callback to later invoke it.
 	//
 	if(m_inspector->get_observer()) {
-		m_inspector->get_observer()->on_connect(evt, packed_data);
+		m_inspector->m_post_process_cbs.emplace(
+		        [packed_data](sinsp_observer *observer, sinsp_evt *evt) {
+			        observer->on_connect(evt, packed_data);
+		        });
 	}
 }
 
@@ -3144,8 +3147,14 @@
 	fdi->m_name = evt->get_param_as_str(1, &parstr, sinsp_evt::PF_SIMPLE);
 	fdi->m_flags = 0;
 
+	//
+	// If there's a listener, add a callback to later invoke it.
+	//
 	if(m_inspector->get_observer()) {
-		m_inspector->get_observer()->on_accept(evt, fd, packed_data, fdi.get());
+		m_inspector->m_post_process_cbs.emplace(
+		        [fd, packed_data](sinsp_observer *observer, sinsp_evt *evt) {
+			        observer->on_accept(evt, fd, packed_data, evt->get_fd_info());
+		        });
 	}
 
 	//
@@ -3205,8 +3214,21 @@
 		m_inspector->get_fds_to_remove().push_back(params->m_fd);
 	}
 
+	//
+	// If there's a listener, add a callback to later invoke it.
+	//
 	if(m_inspector->get_observer()) {
-		m_inspector->get_observer()->on_erase_fd(params);
+		auto ts = params->m_ts;
+		auto remove_from_table = params->m_remove_from_table;
+		auto fd = params->m_fd;
+		auto tinfo = params->m_tinfo;
+		auto fdinfo = params->m_fdinfo;
+		m_inspector->m_post_process_cbs.emplace(
+		        [ts, remove_from_table, fd, tinfo, fdinfo](sinsp_observer *observer,
+		                                                   sinsp_evt *evt) {
+			        erase_fd_params p = {remove_from_table, fd, tinfo, fdinfo, ts};
+			        observer->on_erase_fd(&p);
+		        });
 	}
 }
 
@@ -3669,8 +3691,6 @@
 		}
 
 		if(eflags & EF_READS_FROM_FD) {
-			const char *data;
-			uint32_t datalen;
 			int32_t tupleparam = -1;
 
 			if(etype == PPME_SOCKET_RECVFROM_X) {
@@ -3729,20 +3749,23 @@
 				parinfo = evt->get_param(1);
 			}
 
-			datalen = parinfo->m_len;
-			data = parinfo->m_val;
+			uint32_t datalen = parinfo->m_len;
+			const char *data = parinfo->m_val;
 
 			//
-			// If there's an fd listener, call it now
+			// If there's a listener, add a callback to later invoke it.
 			//
 			if(m_inspector->get_observer()) {
-				m_inspector->get_observer()->on_read(evt,
-				                                     tid,
-				                                     evt->get_tinfo()->m_lastevent_fd,
-				                                     evt->get_fd_info(),
-				                                     data,
-				                                     (uint32_t)retval,
-				                                     datalen);
+				m_inspector->m_post_process_cbs.emplace(
+				        [tid, data, retval, datalen](sinsp_observer *observer, sinsp_evt *evt) {
+					        observer->on_read(evt,
+					                          tid,
+					                          evt->get_tinfo()->m_lastevent_fd,
+					                          evt->get_fd_info(),
+					                          data,
+					                          (uint32_t)retval,
+					                          datalen);
+				        });
 			}
 
 			//
@@ -3791,7 +3814,6 @@
 #endif
 
 		} else {
-			const char *data;
 			uint32_t datalen;
 			int32_t tupleparam = -1;
 
@@ -3849,19 +3871,22 @@
 			//
 			parinfo = evt->get_param(1);
 			datalen = parinfo->m_len;
-			data = parinfo->m_val;
+			const char *data = parinfo->m_val;
 
 			//
-			// If there's an fd listener, call it now
+			// If there's a listener, add a callback to later invoke it.
 			//
 			if(m_inspector->get_observer()) {
-				m_inspector->get_observer()->on_write(evt,
-				                                      tid,
-				                                      evt->get_tinfo()->m_lastevent_fd,
-				                                      evt->get_fd_info(),
-				                                      data,
-				                                      (uint32_t)retval,
-				                                      datalen);
+				m_inspector->m_post_process_cbs.emplace(
+				        [tid, data, retval, datalen](sinsp_observer *observer, sinsp_evt *evt) {
+					        observer->on_write(evt,
+					                           tid,
+					                           evt->get_tinfo()->m_lastevent_fd,
+					                           evt->get_fd_info(),
+					                           data,
+					                           (uint32_t)retval,
+					                           datalen);
+				        });
 			}
 
 			// perform syslog decoding if applicable
@@ -3873,8 +3898,14 @@
 		if(evt->get_fd_info()->m_type == SCAP_FD_IPV4_SOCK ||
 		   evt->get_fd_info()->m_type == SCAP_FD_IPV6_SOCK) {
 			evt->get_fd_info()->set_socket_failed();
+			//
+			// If there's a listener, add a callback to later invoke it.
+			//
 			if(m_inspector->get_observer()) {
-				m_inspector->get_observer()->on_socket_status_changed(evt);
+				m_inspector->m_post_process_cbs.emplace(
+				        [](sinsp_observer *observer, sinsp_evt *evt) {
+					        observer->on_socket_status_changed(evt);
+				        });
 			}
 		}
 	}
@@ -3897,7 +3928,6 @@
 	//
 	if(retval >= 0) {
 		sinsp_evt *enter_evt = &m_tmp_evt;
-		int64_t fdin;
 
 		if(!retrieve_enter_event(enter_evt, evt)) {
 			return;
@@ -3906,13 +3936,16 @@
 		//
 		// Extract the in FD
 		//
-		fdin = enter_evt->get_param(1)->as<int64_t>();
+		int64_t fdin = enter_evt->get_param(1)->as<int64_t>();
 
 		//
-		// If there's an fd listener, call it now
+		// If there's a listener, add a callback to later invoke it.
 		//
 		if(m_inspector->get_observer()) {
-			m_inspector->get_observer()->on_sendfile(evt, fdin, (uint32_t)retval);
+			m_inspector->m_post_process_cbs.emplace(
+			        [fdin, retval](sinsp_observer *observer, sinsp_evt *evt) {
+				        observer->on_sendfile(evt, fdin, (uint32_t)retval);
+			        });
 		}
 	}
 }
@@ -4068,8 +4101,13 @@
 			return;
 		}
 
+		//
+		// If there's a listener, add a callback to later invoke it.
+		//
 		if(m_inspector->get_observer()) {
-			m_inspector->get_observer()->on_socket_shutdown(evt);
+			m_inspector->m_post_process_cbs.emplace([](sinsp_observer *observer, sinsp_evt *evt) {
+				observer->on_socket_shutdown(evt);
+			});
 		}
 	}
 }
@@ -5048,8 +5086,13 @@
 		} else {
 			evt->get_fd_info()->set_socket_connected();
 		}
+		//
+		// If there's a listener, add a callback to later invoke it.
+		//
 		if(m_inspector->get_observer()) {
-			m_inspector->get_observer()->on_socket_status_changed(evt);
+			m_inspector->m_post_process_cbs.emplace([](sinsp_observer *observer, sinsp_evt *evt) {
+				observer->on_socket_status_changed(evt);
+			});
 		}
 	}
 }

userspace/libsinsp/sinsp.cpp

@@ -1316,6 +1316,20 @@ int32_t sinsp::next(sinsp_evt** puevt) {
 			// todo(jason): should we log parsing errors here?
 			pp.process_event(evt, m_event_sources);
 		}
+
+		// Once we processed all events, make sure to call all
+		// requested post-process callbacks.
+		// At this point, any plugin could have modified state tables,
+		// thus we can guarantee that any post-process callback
+		// will see the full post-event-processed state.
+		// NOTE: we don't use a RAII object because
+		// we cannot guarantee that no exception will be thrown by the callbacks.
+		if(m_observer != nullptr) {
+			for(; !m_post_process_cbs.empty(); m_post_process_cbs.pop()) {
+				auto cb = m_post_process_cbs.front();
+				cb(m_observer, evt);
+			}
+		}
 	}
 
 	// Finally set output evt;

userspace/libsinsp/sinsp.h

@@ -82,6 +82,7 @@ limitations under the License.
 #include <string>
 #include <unordered_set>
 #include <vector>
+#include <queue>
 
 #define ONE_SECOND_IN_NS 1000000000LL
 
@@ -1242,6 +1243,8 @@ class SINSP_PUBLIC sinsp : public capture_stats_source {
 
 	sinsp_observer* m_observer{nullptr};
 
+	std::queue<std::function<void(sinsp_observer* observer, sinsp_evt* evt)>> m_post_process_cbs{};
+
 	bool m_inited;
 	static std::atomic<int> instance_count;
 };