[POA-2928] Changed the event watcher to watch pod changes not event changes

cmd/internal/kube/daemonset/apidump_process.go

@@ -22,15 +22,21 @@ func (d *Daemonset) StartApiDumpProcess(podUID types.UID) error {
 		return err
 	}
 
-	err = podArgs.changePodTrafficMonitorState(TrafficMonitoringStarted, PodDetected, PodInitialized)
+	err = podArgs.changePodTrafficMonitorState(TrafficMonitoringRunning, PodRunning)
 	if err != nil {
 		return errors.Wrapf(err, "failed to change pod state, pod name: %s, from: %s to: %s",
-			podArgs.PodName, podArgs.PodTrafficMonitorState, TrafficMonitoringStarted)
+			podArgs.PodName, podArgs.PodTrafficMonitorState, TrafficMonitoringRunning)
 	}
 
+	// Increment the wait group counter
+	d.ApidumpProcessesWG.Add(1)
+
 	go func() (funcErr error) {
 		// defer function handle the error (if any) in the apidump process and change the pod state accordingly
 		defer func() {
+			// Decrement the wait group counter
+			d.ApidumpProcessesWG.Done()
+
 			nextState := TrafficMonitoringEnded
 
 			if err := recover(); err != nil {
@@ -41,22 +47,17 @@ func (d *Daemonset) StartApiDumpProcess(podUID types.UID) error {
 				printer.Errorf("Error occurred in apidump process for pod %s, err: %v\n", podArgs.PodName, funcErr)
 				nextState = TrafficMonitoringFailed
 			} else {
-				printer.Infof("Apidump process ended for pod %s", podArgs.PodName)
+				printer.Infof("Apidump process ended for pod %s\n", podArgs.PodName)
 			}
 
-			err = podArgs.changePodTrafficMonitorState(nextState, TrafficMonitoringStarted)
+			// Move monitoring state to final apidump processing state
+			err = podArgs.changePodTrafficMonitorState(nextState,
+				TrafficMonitoringRunning, PodSucceeded, PodFailed, PodTerminated, DaemonSetShutdown)
 			if err != nil {
 				printer.Errorf("Failed to change pod state, pod name: %s, from: %s to: %s, error: %v\n",
 					podArgs.PodName, podArgs.PodTrafficMonitorState, nextState, err)
 				return
 			}
-
-			// It is possible that the apidump process is already stopped and the stopChannel is of no use
-			// This is just a safety check
-			err := d.StopApiDumpProcess(podUID, err)
-			if err != nil {
-				printer.Errorf("Failed to stop api dump process, pod name: %s, error: %v\n", podArgs.PodName, err)
-			}
 		}()
 
 		networkNamespace, err := d.CRIClient.GetNetworkNamespace(podArgs.ContainerUUID)
@@ -90,32 +91,26 @@ func (d *Daemonset) StartApiDumpProcess(podUID types.UID) error {
 		}
 
 		if err := apidump.Run(apidumpArgs); err != nil {
-			funcErr = errors.Wrapf(err, "Failed to run apidump process for pod %s", podArgs.PodName)
+			funcErr = errors.Wrapf(err, "failed to run apidump process for pod %s", podArgs.PodName)
 		}
 		return
 	}()
 
 	return nil
 }
 
-// StopApiDumpProcess stops the API dump process for a given pod identified by its UID.
-// It retrieves the pod arguments from a map and changes the pod's traffic monitor state.
-// If successful, it sends a stop signal through the pod's stop channel.
+// StopApiDumpProcess signals the API dump process to stop for a given pod
+// identified by its UID. It retrieves the process's stop channel object from a map
+// and sends a stop signal to trigger apidump shutdown.
 func (d *Daemonset) StopApiDumpProcess(podUID types.UID, stopErr error) error {
 	podArgs, err := d.getPodArgsFromMap(podUID)
 	if err != nil {
 		return err
 	}
 
-	err = podArgs.changePodTrafficMonitorState(TrafficMonitoringStopped,
-		PodTerminated, DaemonSetShutdown, TrafficMonitoringFailed, TrafficMonitoringEnded)
-	if err != nil {
-		return errors.Wrapf(err, "failed to change pod state, pod name: %s, from: %s to: %s",
-			podArgs.PodName, podArgs.PodTrafficMonitorState, TrafficMonitoringStopped)
-	}
-
 	printer.Infof("Stopping API dump process for pod %s\n", podArgs.PodName)
 	podArgs.StopChan <- stopErr
+	close(podArgs.StopChan)
 
 	return nil
 }

cmd/internal/kube/daemonset/constants.go

@@ -2,13 +2,16 @@ package daemonset
 
 import "time"
 
-type podEnvVars string
-
 const (
 	// Pod environment variables
-	POSTMAN_INSIGHTS_PROJECT_ID podEnvVars = "POSTMAN_INSIGHTS_PROJECT_ID"
-	POSTMAN_INSIGHTS_API_KEY    podEnvVars = "POSTMAN_INSIGHTS_API_KEY"
-	POSTMAN_INSIGHTS_ENV        podEnvVars = "POSTMAN_INSIGHTS_ENV"
+	POSTMAN_INSIGHTS_PROJECT_ID = "POSTMAN_INSIGHTS_PROJECT_ID"
+	POSTMAN_INSIGHTS_API_KEY    = "POSTMAN_INSIGHTS_API_KEY"
+
+	// Daemonset environment variables
+	POSTMAN_INSIGHTS_ENV                = "POSTMAN_ENV" // This is same as root POSTMAN_ENV
+	POSTMAN_INSIGHTS_VERIFICATION_TOKEN = "POSTMAN_INSIGHTS_VERIFICATION_TOKEN"
+	POSTMAN_INSIGHTS_CLUSTER_NAME       = "POSTMAN_INSIGHTS_CLUSTER_NAME"
+	POSTMAN_INSIGHTS_REPRO_MODE_ENABLED = "POSTMAN_INSIGHTS_REPRO_MODE_ENABLED"
 
 	// Workers intervals
 	DefaultTelemetryInterval      = 5 * time.Minute

cmd/internal/kube/daemonset/daemonset.go

@@ -28,6 +28,7 @@ const (
 
 type Daemonset struct {
 	ClusterName              string
+	InsightsEnvironment      string
 	InsightsReproModeEnabled bool
 
 	KubeClient  kube_apis.KubeClient
@@ -38,6 +39,9 @@ type Daemonset struct {
 	// and do not have the agent sidecar container
 	PodArgsByNameMap sync.Map
 
+	// WaitGroup to wait for all apidump processes to stop
+	ApidumpProcessesWG sync.WaitGroup
+
 	PodHealthCheckInterval time.Duration
 	TelemetryInterval      time.Duration
 }
@@ -49,7 +53,7 @@ func StartDaemonset() error {
 	}
 
 	// Initialize the front client
-	postmanInsightsVerificationToken := os.Getenv("POSTMAN_INSIGHTS_VERIFICATION_TOKEN")
+	postmanInsightsVerificationToken := os.Getenv(POSTMAN_INSIGHTS_VERIFICATION_TOKEN)
 	frontClient := rest.NewFrontClient(
 		rest.Domain,
 		telemetry.GetClientID(),
@@ -59,7 +63,7 @@ func StartDaemonset() error {
 	defer cancel()
 
 	// Send initial telemetry
-	clusterName := os.Getenv("POSTMAN_CLUSTER_NAME")
+	clusterName := os.Getenv(POSTMAN_INSIGHTS_CLUSTER_NAME)
 	telemetryInterval := DefaultTelemetryInterval
 	if clusterName == "" {
 		printer.Infof(
@@ -81,8 +85,6 @@ func StartDaemonset() error {
 		}
 	}
 
-	insightsReproModeEnabled := os.Getenv("POSTMAN_INSIGHTS_REPRO_MODE_ENABLED") == "true"
-
 	kubeClient, err := kube_apis.NewKubeClient()
 	if err != nil {
 		return errors.Wrap(err, "failed to create kube client")
@@ -93,8 +95,12 @@ func StartDaemonset() error {
 		return errors.Wrap(err, "failed to create CRI client")
 	}
 
+	insightsReproModeEnabled := os.Getenv(POSTMAN_INSIGHTS_REPRO_MODE_ENABLED) == "true"
+	insightsEnvironment := os.Getenv(POSTMAN_INSIGHTS_ENV)
+
 	daemonsetRun := &Daemonset{
 		ClusterName:              clusterName,
+		InsightsEnvironment:      insightsEnvironment,
 		InsightsReproModeEnabled: insightsReproModeEnabled,
 		KubeClient:               kubeClient,
 		CRIClient:                criClient,
@@ -126,8 +132,8 @@ func (d *Daemonset) Run() error {
 	go d.TelemetryWorker(done)
 
 	// Start the kubernetes events worker
-	printer.Infof("Starting kubernetes events worker...\n")
-	go d.KubernetesEventsWorker(done)
+	printer.Infof("Starting kubernetes pods events worker...\n")
+	go d.KubernetesPodEventsWorker(done)
 
 	// Start the pods health worker
 	printer.Infof("Starting pods health worker...\n")
@@ -241,7 +247,9 @@ func (d *Daemonset) StartProcessInExistingPods() error {
 
 	// Iterate over each pod without the agent sidecar
 	for _, pod := range podsWithoutAgentSidecar {
-		args, err := d.inspectPodForEnvVars(pod)
+		// Empty pod_args object for PodPending state
+		args := NewPodArgs(pod.Name)
+		err := d.inspectPodForEnvVars(pod, args)
 		if err != nil {
 			switch err {
 			case allRequiredEnvVarsAbsentErr:
@@ -254,7 +262,7 @@ func (d *Daemonset) StartProcessInExistingPods() error {
 			continue
 		}
 
-		err = d.addPodArgsToMap(pod.UID, args, PodDetected)
+		err = d.addPodArgsToMap(pod.UID, args, PodRunning)
 		if err != nil {
 			printer.Errorf("Failed to add pod args to map, pod name: %s, error: %v\n", pod.Name, err)
 			continue
@@ -269,26 +277,34 @@ func (d *Daemonset) StartProcessInExistingPods() error {
 	return nil
 }
 
-// KubernetesEventsWorker listens for Kubernetes events and handles them accordingly.
+// KubernetesPodEventsWorker listens for Kubernetes events and handles them accordingly.
 // It runs indefinitely until the provided done channel is closed.
-func (d *Daemonset) KubernetesEventsWorker(done <-chan struct{}) {
+func (d *Daemonset) KubernetesPodEventsWorker(done <-chan struct{}) {
 	for {
 		select {
 		case <-done:
+			printer.Debugf("Kubernetes pod events worker stopped\n")
 			return
-		case event := <-d.KubeClient.EventWatch.ResultChan():
+		case event := <-d.KubeClient.PodEventWatch.ResultChan():
 			switch event.Type {
 			case watch.Added:
-				printer.Debugf("Got k8s added event: %v\n", event.Object)
-				if e, ok := event.Object.(*coreV1.Event); ok {
-					go d.handlePodAddEvent(e.InvolvedObject.UID)
+				printer.Debugf("Got k8s pod added event: %v\n", event.Object)
+				if p, ok := event.Object.(*coreV1.Pod); ok {
+					go d.handlePodAddEvent(*p)
 				}
+			// A pod is deleted
 			case watch.Deleted:
-				printer.Debugf("Got k8s deleted event: %v\n", event.Object)
-				if e, ok := event.Object.(*coreV1.Event); ok {
-					go d.handlePodDeleteEvent(e.InvolvedObject.UID)
+				printer.Debugf("Got k8s pod deleted event: %v\n", event.Object)
+				if p, ok := event.Object.(*coreV1.Pod); ok {
+					go d.handlePodDeleteEvent(*p)
+				}
+			case watch.Modified:
+				printer.Debugf("Got k8s pod modified event: %v\n", event.Object)
+				if p, ok := event.Object.(*coreV1.Pod); ok {
+					go d.handlePodModifyEvent(*p)
 				}
 			}
+
 		}
 	}
 }
@@ -319,11 +335,17 @@ func (d *Daemonset) PodsHealthWorker(done <-chan struct{}) {
 // 2. Stops the API dump process for the pod.
 // 3. Logs any errors encountered during the state change or stopping process.
 // 4. Removes the pod from the PodArgsByNameMap.
+// 5. Wait for all the apidump processes to stop.
 func (d *Daemonset) StopAllApiDumpProcesses() {
 	d.PodArgsByNameMap.Range(func(k, v interface{}) bool {
 		podUID := k.(types.UID)
 		podArgs := v.(*PodArgs)
 
+		if podArgs.isEndState() {
+			printer.Debugf("API dump process for pod %s already stopped, state: %s\n", podArgs.PodName, podArgs.PodTrafficMonitorState)
+			return true
+		}
+
 		// Since this state can happen at any time so no check for allowed current states
 		err := podArgs.changePodTrafficMonitorState(DaemonSetShutdown)
 		if err != nil {
@@ -341,4 +363,8 @@ func (d *Daemonset) StopAllApiDumpProcesses() {
 		d.PodArgsByNameMap.Delete(podUID)
 		return true
 	})
+
+	// Wait for all apidump processes to stop
+	printer.Debugf("Waiting for all apidump processes to stop...\n")
+	d.ApidumpProcessesWG.Wait()
 }