Reduce routine event traffic

CHANGELOG.md

@@ -13,6 +13,7 @@ All Nullnet releases with the relative changes are documented in this file.
 - Per-service egress/ingress traffic filters: arbitrary AND/OR/group combinations of Country, Organization, Src IP (ingress), and Dst IP (egress) conditions, evaluated via `rpn-predicate-interpreter` — replaces the country-only egress/ingress policy ([#171](https://github.com/NullNet-ai/nullnet/pull/171) — fixes [#143](https://github.com/NullNet-ai/nullnet/issues/143))
 - Persist ingress and egress sessions to SQLite and show the full history on the Sessions page, filterable by status, service, direction, and policy verdict, with its own retention window ([#170](https://github.com/NullNet-ai/nullnet/pull/170) — fixes [#156](https://github.com/NullNet-ai/nullnet/issues/156))
 ### Changed
+- Keep routine successful proxy requests out of Events, isolate diagnostic event RPC admission from routing, and group event filters by scope ([#204](https://github.com/NullNet-ai/nullnet/pull/204)).
 - Reduce fresh-edge cleanup work, batch conntrack scans, and share TC classifier infrastructure while preserving dedicated encrypted endpoints ([#201](https://github.com/NullNet-ai/nullnet/pull/201)).
 - Keep active backend chains visible to conntrack so liveness tracking does not release endpoints that still carry traffic ([#201](https://github.com/NullNet-ai/nullnet/pull/201)).
 - Prevent Docker Swarm VNI collisions and keep source-scoped egress routing valid for the full network-ID range ([#201](https://github.com/NullNet-ai/nullnet/pull/201)).

members/nullnet-grpc-lib/src/lib.rs

@@ -23,6 +23,8 @@ use tonic::transport::{Channel, ClientTlsConfig};
 const MAX_IN_FLIGHT_UNARY: usize = 32;
 const MAX_IN_FLIGHT_LIFECYCLE: usize = 8;
 const MAX_IN_FLIGHT_POLICY: usize = 32;
+// Diagnostic bursts must not consume routing admission.
+const MAX_IN_FLIGHT_EVENTS: usize = 8;
 
 /// Why a `proxy` lookup failed.
 #[derive(Debug)]
@@ -40,6 +42,7 @@ pub struct NullnetGrpcInterface {
     unary_slots: Arc<Semaphore>,
     lifecycle_slots: Arc<Semaphore>,
     policy_slots: Arc<Semaphore>,
+    event_slots: Arc<Semaphore>,
 }
 
 impl NullnetGrpcInterface {
@@ -72,6 +75,7 @@ impl NullnetGrpcInterface {
                     unary_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_UNARY)),
                     lifecycle_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_LIFECYCLE)),
                     policy_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_POLICY)),
+                    event_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_EVENTS)),
                 });
             }
 
@@ -328,10 +332,13 @@ impl NullnetGrpcInterface {
 
     #[allow(clippy::missing_errors_doc)]
     pub async fn report_event(&self, event: AgentEvent) -> Result<(), String> {
-        self.unary(self.client.clone().report_event(Request::new(event)))
-            .await
-            .map(|_| ())
-            .map_err(|e| e.to_string())
+        Self::admitted(
+            &self.event_slots,
+            self.client.clone().report_event(Request::new(event)),
+        )
+        .await
+        .map(|_| ())
+        .map_err(|e| e.to_string())
     }
 
     /// Subscribe to certificate changes: the returned stream yields the full
@@ -389,6 +396,7 @@ mod admission_tests {
             unary_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_UNARY)),
             lifecycle_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_LIFECYCLE)),
             policy_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_POLICY)),
+            event_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_EVENTS)),
         };
         let _busy = interface
             .unary_slots
@@ -410,6 +418,37 @@ mod admission_tests {
         );
     }
 
+    #[tokio::test]
+    async fn event_reporting_progresses_when_setup_admission_is_full() {
+        let interface = NullnetGrpcInterface {
+            client: NullnetGrpcClient::new(
+                Channel::from_static("http://127.0.0.1:1").connect_lazy(),
+            ),
+            unary_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_UNARY)),
+            lifecycle_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_LIFECYCLE)),
+            policy_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_POLICY)),
+            event_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_EVENTS)),
+        };
+        let _busy = interface
+            .unary_slots
+            .acquire_many(MAX_IN_FLIGHT_UNARY as u32)
+            .await
+            .unwrap();
+        let result = tokio::time::timeout(
+            std::time::Duration::from_millis(200),
+            interface.report_event(AgentEvent { event: None }),
+        )
+        .await;
+        assert!(
+            result.is_ok(),
+            "event reporting was queued behind setup admission"
+        );
+        assert!(
+            result.unwrap().is_err(),
+            "the intentionally absent server should reject the call"
+        );
+    }
+
     #[tokio::test]
     async fn clones_share_admission_and_cancellation_releases_capacity() {
         let interface = NullnetGrpcInterface {
@@ -419,6 +458,7 @@ mod admission_tests {
             unary_slots: Arc::new(Semaphore::new(2)),
             lifecycle_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_LIFECYCLE)),
             policy_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_POLICY)),
+            event_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_EVENTS)),
         };
         let (entered, mut received) = mpsc::unbounded_channel();
         let mut calls = tokio::task::JoinSet::new();
@@ -456,6 +496,7 @@ mod admission_tests {
             unary_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_UNARY)),
             lifecycle_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_LIFECYCLE)),
             policy_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_POLICY)),
+            event_slots: Arc::new(Semaphore::new(MAX_IN_FLIGHT_EVENTS)),
         };
         let _busy = interface
             .unary_slots

members/nullnet-proxy/src/main.rs

@@ -14,8 +14,8 @@ use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use nullnet_grpc_lib::NullnetGrpcInterface;
 use nullnet_grpc_lib::nullnet_grpc::{
-    AgentEvent, AgentProxyClientNotInet, AgentProxyRequestRouted, AgentTlsCertificateInvalid,
-    AgentUpstreamLookupFailed, ProxyRequest, agent_event::Event as AgentEventKind,
+    AgentEvent, AgentProxyClientNotInet, AgentTlsCertificateInvalid, AgentUpstreamLookupFailed,
+    ProxyRequest, agent_event::Event as AgentEventKind,
 };
 use nullnet_liberror::{ErrorHandler, Location, location};
 use pingora_core::listeners::tls::TlsSettings;
@@ -379,25 +379,6 @@ impl ProxyHttp for NullnetProxy {
         println!("upstream: {upstream}\n");
 
         let latency_ms = init_t.elapsed().as_millis() as u64;
-        let server = self.server.clone();
-        let svc = service_name.clone();
-        let cip = client_ip.clone();
-        let uip = upstream.ip().to_string();
-        tokio::spawn(async move {
-            let _ = server
-                .report_event(AgentEvent {
-                    event: Some(AgentEventKind::ProxyRequestRouted(
-                        AgentProxyRequestRouted {
-                            service_name: svc,
-                            client_ip: cip,
-                            upstream_ip: uip,
-                            latency_ms,
-                        },
-                    )),
-                })
-                .await;
-        });
-
         println!("TOTAL VLANS SETUP TIME: {} ms\n", latency_ms);
 
         Ok(Box::new(HttpPeer::new(upstream, false, String::new())))

members/nullnet-server/src/events.rs

@@ -220,12 +220,6 @@ pub(crate) enum Event {
         client_ip: String,
         timestamp: u64,
     },
-    StickySessionReused {
-        service: String,
-        client_ip: String,
-        proxy_ip: String,
-        timestamp: u64,
-    },
     StaleSessionEvicted {
         service: String,
         client_ip: String,
@@ -495,15 +489,6 @@ pub(crate) enum Event {
         timestamp: u64,
     },
 
-    // --- Proxy info events ---
-    ProxyRequestRouted {
-        service_name: String,
-        client_ip: String,
-        upstream_ip: String,
-        latency_ms: u64,
-        timestamp: u64,
-    },
-
     // --- Certificate events ---
     UiTlsCertificateActive {
         domain: String,
@@ -576,7 +561,6 @@ impl Event {
             Self::AllReplicasRemoved { .. } => "all_replicas_removed",
             Self::ServiceReachabilityToggled { .. } => "service_reachability_toggled",
             Self::ProxyClientTimedOut { .. } => "proxy_client_timed_out",
-            Self::StickySessionReused { .. } => "sticky_session_reused",
             Self::StaleSessionEvicted { .. } => "stale_session_evicted",
             Self::MaxNetworksLimitEnforced { .. } => "max_networks_limit_enforced",
             Self::NetIdPoolExhausted { .. } => "net_id_pool_exhausted",
@@ -627,7 +611,6 @@ impl Event {
             Self::UdpListenerBindFailed { .. } => "udp_listener_bind_failed",
             Self::TcpUpstreamConnectFailed { .. } => "tcp_upstream_connect_failed",
             Self::UdpUpstreamConnectFailed { .. } => "udp_upstream_connect_failed",
-            Self::ProxyRequestRouted { .. } => "proxy_request_routed",
             Self::UiTlsCertificateActive { .. } => "ui_tls_certificate_active",
             Self::UiTlsCertificateUnavailable { .. } => "ui_tls_certificate_unavailable",
             Self::CertificateInstalled { .. } => "certificate_installed",
@@ -669,12 +652,10 @@ impl Event {
             | Self::ConfigStackRemoved { .. }
             | Self::ServiceUnregistered { .. }
             | Self::ServiceReachabilityToggled { .. }
-            | Self::StickySessionReused { .. }
             | Self::VxlanSetupCompleted { .. }
             | Self::VlanSetupCompleted { .. }
             | Self::ControlChannelEstablished { .. }
             | Self::ServicesListUpdated { .. }
-            | Self::ProxyRequestRouted { .. }
             | Self::ProxyConnected { .. }
             | Self::UiTlsCertificateActive { .. }
             | Self::CertificateInstalled { .. }
@@ -948,19 +929,6 @@ impl Event {
         }
     }
 
-    pub(crate) fn sticky_session_reused(
-        service: String,
-        client_ip: String,
-        proxy_ip: String,
-    ) -> Self {
-        Self::StickySessionReused {
-            service,
-            client_ip,
-            proxy_ip,
-            timestamp: now_secs(),
-        }
-    }
-
     pub(crate) fn stale_session_evicted(
         service: String,
         client_ip: String,
@@ -1297,21 +1265,6 @@ impl Event {
         }
     }
 
-    pub(crate) fn proxy_request_routed(
-        service_name: String,
-        client_ip: String,
-        upstream_ip: String,
-        latency_ms: u64,
-    ) -> Self {
-        Self::ProxyRequestRouted {
-            service_name,
-            client_ip,
-            upstream_ip,
-            latency_ms,
-            timestamp: now_secs(),
-        }
-    }
-
     pub(crate) fn ui_tls_certificate_active(domain: String) -> Self {
         Self::UiTlsCertificateActive {
             domain,

members/nullnet-server/src/nullnet_grpc_impl.rs

@@ -578,15 +578,6 @@ impl NullnetGrpcImpl {
             if intact {
                 println!("'{client_ip}' ---> '{service_name}' is already set up");
 
-                self.orchestrator
-                    .events
-                    .emit(Event::sticky_session_reused(
-                        service_name.to_string(),
-                        client_ip.to_string(),
-                        proxy_ip.to_string(),
-                    ))
-                    .await;
-
                 // update the latest timestamp for this client since it's being used again
                 let mut services_mut = self.services.write().await;
                 if let Some(stack_map) = services_mut.get_mut(&stack)
@@ -2868,12 +2859,8 @@ impl NullnetGrpc for NullnetGrpcImpl {
             AgentEventKind::UdpUpstreamConnectFailed(e) => {
                 Event::udp_upstream_connect_failed(e.service_name, e.client_ip, e.error_message)
             }
-            AgentEventKind::ProxyRequestRouted(e) => Event::proxy_request_routed(
-                e.service_name,
-                e.client_ip,
-                e.upstream_ip,
-                e.latency_ms,
-            ),
+            // Older proxies may still report this routine success event.
+            AgentEventKind::ProxyRequestRouted(_) => return Ok(Response::new(Empty {})),
         };
         self.orchestrator.events.emit(event).await;
         Ok(Response::new(Empty {}))
@@ -2887,6 +2874,29 @@ mod proxy_presence_tests {
     use tokio_stream::StreamExt;
     use tonic::transport::server::TcpConnectInfo;
 
+    #[tokio::test]
+    async fn legacy_proxy_request_routed_event_is_ignored() {
+        let server = NullnetGrpcImpl::new_for_test(StackMap::new());
+        let mut events = server.orchestrator.events.subscribe();
+        server
+            .report_event(Request::new(AgentEvent {
+                event: Some(AgentEventKind::ProxyRequestRouted(
+                    nullnet_grpc_lib::nullnet_grpc::AgentProxyRequestRouted {
+                        service_name: "service".into(),
+                        client_ip: "198.18.0.1".into(),
+                        upstream_ip: "10.0.0.1".into(),
+                        latency_ms: 1,
+                    },
+                )),
+            }))
+            .await
+            .unwrap();
+        assert!(matches!(
+            events.try_recv(),
+            Err(tokio::sync::broadcast::error::TryRecvError::Empty)
+        ));
+    }
+
     #[tokio::test]
     async fn idle_proxy_presence_survives_overlapping_streams() {
         let mut server = NullnetGrpcImpl::new_for_test(StackMap::new());