new: implement short lived conn filter

Signed-off-by: Andrea Terzolo <andreaterzolo3@gmail.com>

Author: Andreagit97

checks/net_pod_correlation.go

@@ -173,7 +173,7 @@ func (pi *podCorrelationInfo) startKubernetesInformer(cfg *config.PodCorrelation
 		return fmt.Errorf("failed to create kubernetes informer: %w", err)
 	}
 	// if needed we can configure the refresh interval for the deleted pods cache
-	pi.observer, err = kube.NewObserver(prometheus.DefaultRegisterer, kube.WithPodDebugEndpoint())
+	pi.observer, err = kube.NewObserver(prometheus.DefaultRegisterer, kube.WithPodDebugEndpoint(), kube.WithShortLivedConnectionsInterval(cfg.ShortLivedConnectionsInterval))
 	if err != nil {
 		return fmt.Errorf("failed to create kubernetes observer: %w", err)
 	}
@@ -336,6 +336,11 @@ func (pi *podCorrelationInfo) processConnections(conns []network.ConnectionStats
 			continue
 		}
 
+		// skip short lived connections
+		if pi.observer.FilterShortLivedConnection(conn.Duration) {
+			continue
+		}
+
 		// Possible cases
 		// 1. Pod -> Pod
 		// 2. Pod -> Pod HostNetwork == Pod -> Host

config/config.go

@@ -73,6 +73,8 @@ type PodCorrelationConfig struct {
 	Exporter ExporterConfig
 	// List of attribute keys to include in the correlation
 	AttributesKeys []string
+	// Interval to consider a connection short-lived
+	ShortLivedConnectionsInterval time.Duration
 }
 
 // NetworkTracerConfig contains some[1] of the network tracer configuration options
@@ -288,7 +290,8 @@ func NewDefaultAgentConfig() *AgentConfig {
 					Endpoint: "",
 					Interval: 0,
 				},
-				AttributesKeys: []string{},
+				AttributesKeys:                []string{},
+				ShortLivedConnectionsInterval: 0,
 			},
 		},
 
@@ -707,6 +710,7 @@ func mergeEnvironmentVariables(c *AgentConfig) *AgentConfig {
 		c.NetworkTracer.PodCorrelation.Exporter.Type = translateExporterType(os.Getenv("STS_POD_CORRELATION_EXPORTER_TYPE"))
 		c.NetworkTracer.PodCorrelation.Exporter.Endpoint = os.Getenv("STS_POD_CORRELATION_EXPORTER_OTLP_ENDPOINT")
 		c.NetworkTracer.PodCorrelation.Exporter.Interval, _ = time.ParseDuration(os.Getenv("STS_POD_CORRELATION_EXPORTER_INTERVAL"))
+		c.NetworkTracer.PodCorrelation.ShortLivedConnectionsInterval, _ = time.ParseDuration(os.Getenv("STS_POD_CORRELATION_SHORT_LIVED_CONNECTIONS_INTERVAL"))
 	}
 
 	return c

config/yaml_config.go

@@ -168,7 +168,8 @@ type YamlAgentConfig struct {
 				Endpoint string `yaml:"endpoint"`
 				Interval int    `yaml:"interval"`
 			} `yaml:"exporter"`
-			Attributes []string `yaml:"attributes"`
+			Attributes                    []string `yaml:"attributes"`
+			ShortLivedConnectionsInterval int      `yaml:"short_lived_connections_interval"`
 		} `yaml:"pod_correlation"`
 	} `yaml:"network_tracer_config"`
 	TransactionManager struct {
@@ -456,4 +457,5 @@ func validatePodCorrelationConfig(agentConf *AgentConfig, networkConf *YamlAgent
 	agentConf.NetworkTracer.PodCorrelation.Exporter.Type = translateExporterType(networkConf.Network.PodCorrelation.Exporter.Type)
 	agentConf.NetworkTracer.PodCorrelation.Exporter.Interval = time.Duration(networkConf.Network.PodCorrelation.Exporter.Interval) * time.Second
 	agentConf.NetworkTracer.PodCorrelation.Exporter.Endpoint = networkConf.Network.PodCorrelation.Exporter.Endpoint
+	agentConf.NetworkTracer.PodCorrelation.ShortLivedConnectionsInterval = time.Duration(networkConf.Network.PodCorrelation.ShortLivedConnectionsInterval) * time.Second
 }

k8s-deployment/Makefile

@@ -5,7 +5,7 @@ SETUP_AGENT_HELM_CHART_DIR ?= ~/personal/helm-charts/stable/suse-observability-a
 # SETUP_TYPE can be 'local' or 'remote' (default: local)
 SETUP_TYPE ?= local
 SETUP_API_KEY ?= null
-SETUP_CLUSTER_NAME ?= minikube-local-setup
+SETUP_CLUSTER_NAME ?= local-setup
 SETUP_RECEIVER_ENDPOINT ?= http://test-server-service:7077/stsAgent
 # To be set only in remote mode
 SETUP_OTEL_ENDPOINT ?= null

pkg/kube/observer.go

@@ -17,9 +17,10 @@ import (
 )
 
 const (
-	defaultDeletePodsAfter       = 2 * time.Minute
-	defaultCleanCacheInterval    = 10 * time.Minute
-	defaultMaxEstimatedCPLatency = 5 * time.Second
+	defaultDeletePodsAfter        = 2 * time.Minute
+	defaultCleanCacheInterval     = 10 * time.Minute
+	defaultMaxEstimatedCPLatency  = 5 * time.Second
+	defaultShortLivedConnInterval = 0
 
 	prometheusNamespace = "stackstate_process_agent"
 	prometheusSubsystem = "observer"
@@ -54,6 +55,8 @@ type Observer struct {
 	maxControlPlaneLatency int64
 	// just used for testing
 	nowFunc func() time.Time
+	// if the connection is younger than this interval, we consider it a short lived connection and we don't try to correlate it.
+	shortLivedConnInterval time.Duration
 
 	// Metrics
 	controlPlaneLatency prometheus.Histogram
@@ -114,6 +117,12 @@ func WithPodDebugEndpoint() ObserverOption {
 	}
 }
 
+func WithShortLivedConnectionsInterval(interval time.Duration) ObserverOption {
+	return func(o *Observer) {
+		o.shortLivedConnInterval = interval
+	}
+}
+
 // NewObserver creates a new Observer instance.
 func NewObserver(reg prometheus.Registerer, opts ...ObserverOption) (*Observer, error) {
 	// we need the boot time because all what we receive from ebpf is the time in nanoseconds since boot
@@ -131,6 +140,7 @@ func NewObserver(reg prometheus.Registerer, opts ...ObserverOption) (*Observer,
 		cleanCacheInterval:      defaultCleanCacheInterval,
 		deletePodsAfter:         defaultDeletePodsAfter,
 		nowFunc:                 time.Now,
+		shortLivedConnInterval:  defaultShortLivedConnInterval,
 
 		controlPlaneLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
 			Namespace: prometheusNamespace,
@@ -198,7 +208,7 @@ func NewObserver(reg prometheus.Registerer, opts ...ObserverOption) (*Observer,
 	// update it after we apply the options, since nowFunc can be overridden
 	obs.lastCacheClean = obs.nowFunc()
 
-	log.Infof("Observer created with clean cache interval: %s, delete pods after: %s", obs.cleanCacheInterval, obs.deletePodsAfter)
+	log.Infof("Observer created (clean cache interval: %s), (delete pods after: %s), (short lived connections interval: %s)", obs.cleanCacheInterval, obs.deletePodsAfter, obs.shortLivedConnInterval)
 	return obs, nil
 }
 
@@ -352,6 +362,18 @@ func (o *Observer) ConnectionNeedsRetry(nsFromBoot time.Duration) bool {
 	return false
 }
 
+// FilterShortLivedConnection returns true if the connection is short lived and should be ignored.
+func (o *Observer) FilterShortLivedConnection(nsFromBoot time.Duration) bool {
+	// if the interval is not configured we return false
+	if o.shortLivedConnInterval == 0 {
+		return false
+	}
+
+	connCreationTime := o.bootTime + int64(nsFromBoot.Seconds())
+	// if the connection is younger than the configured interval we consider it a short lived connection
+	return connCreationTime > (o.nowFunc().Unix() - int64(o.shortLivedConnInterval.Seconds()))
+}
+
 // ResolvePodsByIPs resolves the pods by their IPs, returning the PodInfo for each IP.
 func (o *Observer) ResolvePodsByIPs(srcIP, dstIP util.Address, nsFromBoot time.Duration) (*PodInfo, *PodInfo) {
 	o.access.RLock()