fix: issues with prometheus kubernetes pod discovery (#9605)

plugins/inputs/prometheus/kubernetes.go

@@ -111,35 +111,43 @@ func (p *Prometheus) watchPod(ctx context.Context, client *kubernetes.Clientset)
 		LabelSelector: p.KubernetesLabelSelector,
 		FieldSelector: p.KubernetesFieldSelector,
 	})
+	defer watcher.Stop()
 	if err != nil {
 		return err
 	}
-	pod := &corev1.Pod{}
-	go func() {
-		for event := range watcher.ResultChan() {
-			pod = &corev1.Pod{}
-			// If the pod is not "ready", there will be no ip associated with it.
-			if pod.Annotations["prometheus.io/scrape"] != "true" ||
-				!podReady(pod.Status.ContainerStatuses) {
-				continue
-			}
 
-			switch event.Type {
+	for {
-			case watch.Added:
+		select {
-				registerPod(pod, p)
+		case <-ctx.Done():
-			case watch.Modified:
+			return nil
-				// To avoid multiple actions for each event, unregister on the first event
+		default:
-				// in the delete sequence, when the containers are still "ready".
+			for event := range watcher.ResultChan() {
-				if pod.GetDeletionTimestamp() != nil {
+				pod, ok := event.Object.(*corev1.Pod)
-					unregisterPod(pod, p)
+				if !ok {
-				} else {
+					return fmt.Errorf("Unexpected object when getting pods")
+				}
+
+				// If the pod is not "ready", there will be no ip associated with it.
+				if pod.Annotations["prometheus.io/scrape"] != "true" ||
+					!podReady(pod.Status.ContainerStatuses) {
+					continue
+				}
+
+				switch event.Type {
+				case watch.Added:
 					registerPod(pod, p)
+				case watch.Modified:
+					// To avoid multiple actions for each event, unregister on the first event
+					// in the delete sequence, when the containers are still "ready".
+					if pod.GetDeletionTimestamp() != nil {
+						unregisterPod(pod, p)
+					} else {
+						registerPod(pod, p)
+					}
 				}
 			}
 		}
-	}()
+	}
-
-	return nil
 }
 
 func (p *Prometheus) cAdvisor(ctx context.Context, bearerToken string) error {