Repository: open-telemetry/opentelemetry-operator

Issue:
[target-allocator] TargetAllocator assigns Targets to an unhealthy collector?
### Component(s)

target allocator

### What happened?

## Description
I am wondering if it's expected that TargetAllocator still assign Targets to an unhealthy collector?

From the code [[link]](https://github.com/open-telemetry/opentelemetry-operator/blob/main/cmd/otel-allocator/internal/collector/collector.go#L116-L129), it seems there's no pod status check at all.

Can someone help double check if this is true?

## Steps to Reproduce
1. Create regular TargetAllocator and OTEL Collectors.
2. Specify the OTEL Collector StatefulSet to use a bad image or etc, this will cause a OTEL Collector pod to be in a non-Running state.
3. Port forward to the TargetAllocator, curl `http://localhost:8080/jobs/kube-state-metrics/targets`, and you can see the unhealthy OTEL Collector pod is still a candidate to assign targets to.

## Expected Result
We shouldn't assign targets to an unhealthy OTEL Collector? Or we should have a configuration that enables such a feature?

## Actual Result
I was able to make a collector unhealthy but still seeing the unhealthy collector on `http://localhost:8080/jobs/kube-state-metrics/targets`.


### Kubernetes Version

1.31

### Operator version

v0.120.0

### Collector version

v0.120.0

Discussion:
@swiatekm: The problem here is that it's not clear what the behaviour should be. Ideally, we shouldn't assign targets to collectors which aren't healthy, but we also don't want to reallocate all targets just because a single collector was temporarily unhealthy. I had a basic idea on how this could be done in https://github.com/open-telemetry/opentelemetry-operator/issues/2201, but I'm not sure what the possible consequences could be, or how to test it exhaustively.

Maybe the way to go is to add a configurable grace period, and let users turn the behaviour off completely by setting that to 0. If anyone wants to submit a change like that, I'd be willing to review it. The other issue is assigned to me, but I've clearly not made much progress on it since then.@rice-junhaoyu: Hey Mikolaj,

I agree that always trigger a target reallocation for a bad collector is not the ideal solution, that could be too much.

Just one more question, I am still confused about `Maybe the way to go is to add a configurable grace period, and let users turn the behaviour off completely by setting that to 0.`. Can you elaborate a bit more and maybe point me to somewhere? I am kinda new to OTEL and not aware of such a `grace period`.

Appreciate! And please assign me this issue, I will file a PR and send you for a review!@swiatekm: > Just one more question, I am still confused about Maybe the way to go is to add a configurable grace period, and let users turn the behaviour off completely by setting that to 0.. Can you elaborate a bit more and maybe point me to somewhere? I am kinda new to OTEL and not aware of such a grace period.

Basically, the way this [works right now](https://github.com/open-telemetry/opentelemetry-operator/blob/main/cmd/otel-allocator/internal/collector/collector.go#L122) is that we take all the Collector Pods that fulfill our selectors, and which also have a Node assigned.

I was thinking that instead we make this condition something like "the Pod is Ready, or it was Ready less than `grace period` seconds ago". Then if the grace period is, say, 30 seconds, then a Collector Pod could be unhealthy for 15 seconds without triggering a reallocation. 

> Appreciate! And please assign me this issue, I will file a PR and send you for a review!

Sure, if you have any questions, don't hestitate to ask them here!@rice-junhaoyu: Oh, I see, totally agree with your idea. Having such a grace period of `30s` or something prevents too frequent reallocation, epecially in a case of pod restart, which only takes few seconds if there're already available nodes.

In addition, I will make the `grace period` configurable.

Now start working on a PR.@rice-junhaoyu: @swiatekm I've submitted a Pull Request!! https://github.com/open-telemetry/opentelemetry-operator/pull/3807

Basically I wrote an unit test and tested in local to validate my change is working.

In terms of making the `grace period` configurable, I will like to get your opinion first. I was a bit indecisive because I noticed currently `minUpdateInterval` is also un-configurable, so I am not pretty sure if I need to make `grace period` configurable.

---

Actually NOT YET!!!! I believe I have a very basic misunderstanding about `Pod Phase = Running`, **IT DOES NOT NECESSARILY MEAN A OTEL COLLECTOR POD IS HEALTHY**.

Following is an `EC2 Machine Downtime` on our service side, the EC2 Machine started to be unhealthy at `02:45`.
![Image](https://github.com/user-attachments/assets/6aff0757-f44f-4993-bc04-8d97e9802120)

The OTEL Collector pod on that bad EC2 node was already unable to function properly since `02:45`, but the collector was still maintaining `Running` state for quite a while until the EC2 Machine self-recovered.
```
| 2025-03-15 03:01:12.255 | Pending |
| 2025-03-15 03:01:12.252 | Failed |
| 2025-03-15 03:01:12.244 | Failed |
| 2025-03-15 03:01:05.224 | Running |
| 2025-03-15 03:01:01.233 | Running |
| 2025-03-15 03:01:01.228 | Running |
| 2025-03-15 02:50:08.528 | Running |
| 2025-03-15 02:50:08.527 | Running |
| 2025-03-15 02:45:03.419 | Running |
```

I noticed that `PodStatusCondition: Ready` was already `False` from the very beginning of the event, therefore, I believe we should also consider this Pod Condition.
```
[{"type":"PodReadyToStartContainers","status":"True","lastProbeTime":null,"lastTransitionTime":"2025-03-11T15:15:48Z"},
{"type":"Initialized","status":"True","lastProbeTime":null,"lastTransitionTime":"2025-03-11T15:16:05Z"},
{"type":"Ready","status":"False","lastProbeTime":null,"lastTransitionTime":"2025-03-15T02:45:03Z"},
{"type":"ContainersReady","status":"True","lastProbeTime":null,"lastTransitionTime":"2025-03-11T15:16:07Z"},
{"type":"PodScheduled","status":"True","lastProbeTime":null,"lastTransitionTime":"2025-03-11T15:15:40Z"}]
```

---

Did a bit more investigations.

#### `Running` is a Pod Phase that is defined as: "The Pod has been bound to a node, and all of the containers have been created. At least one container is still running, or is in the process of starting or restarting." 
=> It turns out that `Pod Phase = Running` does not necessarily mean a pod is healthy.

#### `Ready` is a Pod Condition that is defined as: "The Pod is able to serve requests and should be added to the load balancing pools of all matching Services."
=> This seems to be something we indeed want.

#### What do you think? I am fine to keep both of them in the checks although I think `Ready` pod condition check is covering the check for `Pod Phase = Running`.@rice-junhaoyu: Pull Request [#3807] updated to consider both `PodPhase` and `PodCondition`!@swiatekm: I'll reopen this, as we still need to switch to a 30s grace period by default.@swiatekm: I created a new issue to track the default change: #3989, so we can close this one.

Solution patch:
diff --git a/.chloggen/ta_not_assign_targets_to_bad_collectors.yaml b/.chloggen/ta_not_assign_targets_to_bad_collectors.yaml
new file mode 100644
index 00000000..5127f5ac
--- /dev/null
+++ b/.chloggen/ta_not_assign_targets_to_bad_collectors.yaml
@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. collector, target allocator, auto-instrumentation, opamp, github action)
+component: target allocator
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Do not assign targets to a collector pod that is not Ready for longer than a non-zero grace period
+
+# One or more tracking issues related to the change
+issues: [3781]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
\ No newline at end of file
diff --git a/cmd/otel-allocator/internal/collector/collector.go b/cmd/otel-allocator/internal/collector/collector.go
index db1f3bfa..2f53a5e3 100644
--- a/cmd/otel-allocator/internal/collector/collector.go
+++ b/cmd/otel-allocator/internal/collector/collector.go
@@ -31,23 +31,25 @@ var (
 )
 
 type Watcher struct {
-	log               logr.Logger
-	k8sClient         kubernetes.Interface
-	close             chan struct{}
-	minUpdateInterval time.Duration
+	log                          logr.Logger
+	k8sClient                    kubernetes.Interface
+	close                        chan struct{}
+	minUpdateInterval            time.Duration
+	collectorNotReadyGracePeriod time.Duration
 }
 
-func NewCollectorWatcher(logger logr.Logger, kubeConfig *rest.Config) (*Watcher, error) {
+func NewCollectorWatcher(logger logr.Logger, kubeConfig *rest.Config, collectorNotReadyGracePeriod time.Duration) (*Watcher, error) {
 	clientset, err := kubernetes.NewForConfig(kubeConfig)
 	if err != nil {
 		return &Watcher{}, err
 	}
 
 	return &Watcher{
-		log:               logger.WithValues("component", "opentelemetry-targetallocator"),
-		k8sClient:         clientset,
-		close:             make(chan struct{}),
-		minUpdateInterval: defaultMinUpdateInterval,
+		log:                          logger.WithValues("component", "opentelemetry-targetallocator"),
+		k8sClient:                    clientset,
+		close:                        make(chan struct{}),
+		minUpdateInterval:            defaultMinUpdateInterval,
+		collectorNotReadyGracePeriod: collectorNotReadyGracePeriod,
 	}, nil
 }
 
@@ -124,6 +126,12 @@ func (k *Watcher) runOnCollectors(store cache.Store, fn func(collectors map[stri
 		if pod.Spec.NodeName == "" {
 			continue
 		}
+
+		// pod healthiness check will always be disabled if CollectorNotReadyGracePeriod is set to 0 * time.Second
+		if k.isPodUnhealthy(pod, k.collectorNotReadyGracePeriod) {
+			continue
+		}
+
 		collectorMap[pod.Name] = allocation.NewCollector(pod.Name, pod.Spec.NodeName)
 	}
 	collectorsDiscovered.Set(float64(len(collectorMap)))
@@ -133,3 +141,25 @@ func (k *Watcher) runOnCollectors(store cache.Store, fn func(collectors map[stri
 func (k *Watcher) Close() {
 	close(k.close)
 }
+
+func (k *Watcher) isPodUnhealthy(pod *v1.Pod, collectorNotReadyGracePeriod time.Duration) bool {
+	if collectorNotReadyGracePeriod == 0*time.Second {
+		return false
+	}
+
+	isPodUnhealthy := false
+	timeNow := time.Now()
+	// stop assigning targets to a non-Running pod that has lasted for a specific period
+	if pod.Status.Phase != v1.PodRunning &&
+		pod.Status.StartTime != nil &&
+		(timeNow.Sub(pod.Status.StartTime.Time) > collectorNotReadyGracePeriod) {
+		isPodUnhealthy = true
+	}
+	// stop assigning targets to a non-Ready pod that has lasted for a specific period
+	for _, condition := range pod.Status.Conditions {
+		if condition.Type == v1.PodReady && condition.Status != v1.ConditionTrue && (timeNow.Sub(condition.LastTransitionTime.Time) > collectorNotReadyGracePeriod) {
+			isPodUnhealthy = true
+		}
+	}
+	return isPodUnhealthy
+}
diff --git a/cmd/otel-allocator/internal/config/config.go b/cmd/otel-allocator/internal/config/config.go
index 296ece16..9991f094 100644
--- a/cmd/otel-allocator/internal/config/config.go
+++ b/cmd/otel-allocator/internal/config/config.go
@@ -30,26 +30,28 @@ import (
 )
 
 const (
-	DefaultResyncTime                        = 5 * time.Minute
-	DefaultConfigFilePath     string         = "/conf/targetallocator.yaml"
-	DefaultCRScrapeInterval   model.Duration = model.Duration(time.Second * 30)
-	DefaultAllocationStrategy                = "consistent-hashing"
-	DefaultFilterStrategy                    = "relabel-config"
+	DefaultResyncTime                                  = 5 * time.Minute
+	DefaultConfigFilePath               string         = "/conf/targetallocator.yaml"
+	DefaultCRScrapeInterval             model.Duration = model.Duration(time.Second * 30)
+	DefaultAllocationStrategy                          = "consistent-hashing"
+	DefaultFilterStrategy                              = "relabel-config"
+	DefaultCollectorNotReadyGracePeriod                = 0 * time.Second
 )
 
 type Config struct {
-	ListenAddr                 string                `yaml:"listen_addr,omitempty"`
-	KubeConfigFilePath         string                `yaml:"kube_config_file_path,omitempty"`
-	ClusterConfig              *rest.Config          `yaml:"-"`
-	RootLogger                 logr.Logger           `yaml:"-"`
-	CollectorSelector          *metav1.LabelSelector `yaml:"collector_selector,omitempty"`
-	CollectorNamespace         string                `yaml:"collector_namespace,omitempty"`
-	PromConfig                 *promconfig.Config    `yaml:"config"`
-	AllocationStrategy         string                `yaml:"allocation_strategy,omitempty"`
-	AllocationFallbackStrategy string                `yaml:"allocation_fallback_strategy,omitempty"`
-	FilterStrategy             string                `yaml:"filter_strategy,omitempty"`
-	PrometheusCR               PrometheusCRConfig    `yaml:"prometheus_cr,omitempty"`
-	HTTPS                      HTTPSServerConfig     `yaml:"https,omitempty"`
+	ListenAddr                   string                `yaml:"listen_addr,omitempty"`
+	KubeConfigFilePath           string                `yaml:"kube_config_file_path,omitempty"`
+	ClusterConfig                *rest.Config          `yaml:"-"`
+	RootLogger                   logr.Logger           `yaml:"-"`
+	CollectorSelector            *metav1.LabelSelector `yaml:"collector_selector,omitempty"`
+	CollectorNamespace           string                `yaml:"collector_namespace,omitempty"`
+	PromConfig                   *promconfig.Config    `yaml:"config"`
+	AllocationStrategy           string                `yaml:"allocation_strategy,omitempty"`
+	AllocationFallbackStrategy   string                `yaml:"allocation_fallback_strategy,omitempty"`
+	FilterStrategy               string                `yaml:"filter_strategy,omitempty"`
+	PrometheusCR                 PrometheusCRConfig    `yaml:"prometheus_cr,omitempty"`
+	HTTPS                        HTTPSServerConfig     `yaml:"https,omitempty"`
+	CollectorNotReadyGracePeriod time.Duration         `yaml:"collector_not_ready_grace_period,omitempty"`
 }
 
 type PrometheusCRConfig struct {
@@ -299,6 +301,7 @@ func CreateDefaultConfig() Config {
 			ScrapeConfigNamespaceSelector:   &metav1.LabelSelector{},
 			ProbeNamespaceSelector:          &metav1.LabelSelector{},
 		},
+		CollectorNotReadyGracePeriod: DefaultCollectorNotReadyGracePeriod,
 	}
 }
 
diff --git a/cmd/otel-allocator/main.go b/cmd/otel-allocator/main.go
index b287a799..e782bb28 100644
--- a/cmd/otel-allocator/main.go
+++ b/cmd/otel-allocator/main.go
@@ -96,7 +96,7 @@ func main() {
 	discoveryManager = discovery.NewManager(discoveryCtx, gokitlog.NewNopLogger(), prometheus.DefaultRegisterer, sdMetrics)
 
 	targetDiscoverer = target.NewDiscoverer(log, discoveryManager, allocatorPrehook, srv, allocator.SetTargets)
-	collectorWatcher, collectorWatcherErr := collector.NewCollectorWatcher(log, cfg.ClusterConfig)
+	collectorWatcher, collectorWatcherErr := collector.NewCollectorWatcher(log, cfg.ClusterConfig, cfg.CollectorNotReadyGracePeriod)
 	if collectorWatcherErr != nil {
 		setupLog.Error(collectorWatcherErr, "Unable to initialize collector watcher")
 		os.Exit(1)



Test patch:
diff --git a/cmd/otel-allocator/internal/collector/collector_test.go b/cmd/otel-allocator/internal/collector/collector_test.go
index 906b5a30..15de3792 100644
--- a/cmd/otel-allocator/internal/collector/collector_test.go
+++ b/cmd/otel-allocator/internal/collector/collector_test.go
@@ -29,12 +29,13 @@ var labelSelector = metav1.LabelSelector{
 	MatchLabels: labelMap,
 }
 
-func getTestPodWatcher() Watcher {
+func getTestPodWatcher(collectorNotReadyGracePeriod time.Duration) Watcher {
 	podWatcher := Watcher{
-		k8sClient:         fake.NewSimpleClientset(),
-		close:             make(chan struct{}),
-		log:               logger,
-		minUpdateInterval: time.Millisecond,
+		k8sClient:                    fake.NewSimpleClientset(),
+		close:                        make(chan struct{}),
+		log:                          logger,
+		minUpdateInterval:            time.Millisecond,
+		collectorNotReadyGracePeriod: collectorNotReadyGracePeriod,
 	}
 	return podWatcher
 }
@@ -49,14 +50,51 @@ func pod(name string) *v1.Pod {
 		Spec: v1.PodSpec{
 			NodeName: "test-node",
 		},
+		Status: v1.PodStatus{Phase: v1.PodRunning, Conditions: []v1.PodCondition{{
+			Type:   v1.PodReady,
+			Status: v1.ConditionTrue,
+		}}},
+	}
+}
+
+func podWithPodPhaseAndStartTime(name string, podPhase v1.PodPhase, startTime time.Time) *v1.Pod {
+	return &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "test-ns",
+			Labels:    labelMap,
+		},
+		Spec: v1.PodSpec{
+			NodeName: "test-node",
+		},
+		Status: v1.PodStatus{Phase: podPhase, StartTime: &metav1.Time{Time: startTime}},
+	}
+}
+
+func podWithPodReadyConditionStatusAndLastTransitionTime(name string, podConditionStatus v1.ConditionStatus, lastTransitionTime time.Time) *v1.Pod {
+	return &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "test-ns",
+			Labels:    labelMap,
+		},
+		Spec: v1.PodSpec{
+			NodeName: "test-node",
+		},
+		Status: v1.PodStatus{Conditions: []v1.PodCondition{{
+			Type:               v1.PodReady,
+			Status:             podConditionStatus,
+			LastTransitionTime: metav1.Time{Time: lastTransitionTime},
+		}}},
 	}
 }
 
 func Test_runWatch(t *testing.T) {
 	namespace := "test-ns"
 	type args struct {
-		kubeFn       func(t *testing.T, podWatcher Watcher)
-		collectorMap map[string]*allocation.Collector
+		collectorNotReadyGracePeriod time.Duration
+		kubeFn                       func(t *testing.T, podWatcher Watcher)
+		collectorMap                 map[string]*allocation.Collector
 	}
 	tests := []struct {
 		name string
@@ -66,6 +104,7 @@ func Test_runWatch(t *testing.T) {
 		{
 			name: "pod add",
 			args: args{
+				collectorNotReadyGracePeriod: 0 * time.Second,
 				kubeFn: func(t *testing.T, podWatcher Watcher) {
 					for _, k := range []string{"test-pod1", "test-pod2", "test-pod3"} {
 						p := pod(k)
@@ -93,6 +132,7 @@ func Test_runWatch(t *testing.T) {
 		{
 			name: "pod delete",
 			args: args{
+				collectorNotReadyGracePeriod: 0 * time.Second,
 				kubeFn: func(t *testing.T, podWatcher Watcher) {
 					for _, k := range []string{"test-pod2", "test-pod3"} {
 						err := podWatcher.k8sClient.CoreV1().Pods(namespace).Delete(context.Background(), k, metav1.DeleteOptions{})
@@ -124,7 +164,7 @@ func Test_runWatch(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			podWatcher := getTestPodWatcher()
+			podWatcher := getTestPodWatcher(tt.args.collectorNotReadyGracePeriod)
 			defer func() {
 				close(podWatcher.close)
 			}()
@@ -157,9 +197,254 @@ func Test_runWatch(t *testing.T) {
 	}
 }
 
+func Test_gracePeriodWithNonRunningPodPhase(t *testing.T) {
+	namespace := "test-ns"
+	type args struct {
+		collectorNotReadyGracePeriod time.Duration
+		collectorMap                 map[string]*allocation.Collector
+	}
+	tests := []struct {
+		name string
+		args args
+		want map[string]*allocation.Collector
+	}{
+		{
+			name: "collector healthiness check disabled",
+			args: args{
+				collectorNotReadyGracePeriod: 0 * time.Second,
+				collectorMap: map[string]*allocation.Collector{
+					"test-pod-running": {
+						Name:     "test-pod-running",
+						NodeName: "test-node",
+					},
+					"test-pod-unknown-within-grace-period": {
+						Name:     "test-pod-unknown-within-grace-period",
+						NodeName: "test-node",
+					},
+					"test-pod-pending-over-grace-period": {
+						Name:     "test-pod-pending-over-grace-period",
+						NodeName: "test-node",
+					},
+				},
+			},
+			want: map[string]*allocation.Collector{
+				"test-pod-running": {
+					Name:     "test-pod-running",
+					NodeName: "test-node",
+				},
+				"test-pod-unknown-within-grace-period": {
+					Name:     "test-pod-unknown-within-grace-period",
+					NodeName: "test-node",
+				},
+				"test-pod-pending-over-grace-period": {
+					Name:     "test-pod-pending-over-grace-period",
+					NodeName: "test-node",
+				},
+			},
+		},
+		{
+			name: "collector healthiness check enabled",
+			args: args{
+				collectorNotReadyGracePeriod: 30 * time.Second,
+				collectorMap: map[string]*allocation.Collector{
+					"test-pod-running": {
+						Name:     "test-pod-running",
+						NodeName: "test-node",
+					},
+					"test-pod-unknown-within-grace-period": {
+						Name:     "test-pod-unknown-within-grace-period",
+						NodeName: "test-node",
+					},
+					"test-pod-pending-over-grace-period": {
+						Name:     "test-pod-pending-over-grace-period",
+						NodeName: "test-node",
+					},
+				},
+			},
+			want: map[string]*allocation.Collector{
+				"test-pod-running": {
+					Name:     "test-pod-running",
+					NodeName: "test-node",
+				},
+				"test-pod-unknown-within-grace-period": {
+					Name:     "test-pod-unknown-within-grace-period",
+					NodeName: "test-node",
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			timeNow := time.Now()
+			podWatcher := getTestPodWatcher(tt.args.collectorNotReadyGracePeriod)
+			defer func() {
+				close(podWatcher.close)
+			}()
+			var actual map[string]*allocation.Collector
+			mapMutex := sync.Mutex{}
+			for _, k := range tt.args.collectorMap {
+				var p *v1.Pod
+				switch k.Name {
+				case "test-pod-running":
+					p = podWithPodPhaseAndStartTime(k.Name, v1.PodRunning, timeNow)
+				case "test-pod-unknown-within-grace-period":
+					p = podWithPodPhaseAndStartTime(k.Name, v1.PodUnknown,
+						timeNow.Add(-1*podWatcher.collectorNotReadyGracePeriod).Add(podWatcher.collectorNotReadyGracePeriod/2))
+				case "test-pod-pending-over-grace-period":
+					p = podWithPodPhaseAndStartTime(k.Name, v1.PodPending,
+						timeNow.Add(-1*podWatcher.collectorNotReadyGracePeriod).Add(-podWatcher.collectorNotReadyGracePeriod/2))
+				}
+				_, err := podWatcher.k8sClient.CoreV1().Pods("test-ns").Create(context.Background(), p, metav1.CreateOptions{})
+				assert.NoError(t, err)
+			}
+
+			go func(podWatcher Watcher) {
+				err := podWatcher.Watch(namespace, &labelSelector, func(colMap map[string]*allocation.Collector) {
+					mapMutex.Lock()
+					defer mapMutex.Unlock()
+					actual = colMap
+				})
+				require.NoError(t, err)
+			}(podWatcher)
+
+			assert.EventuallyWithT(t, func(collect *assert.CollectT) {
+				mapMutex.Lock()
+				defer mapMutex.Unlock()
+				assert.Len(collect, actual, len(tt.want))
+				assert.Equal(collect, actual, tt.want)
+				assert.Equal(collect, testutil.ToFloat64(collectorsDiscovered), float64(len(actual)))
+			}, time.Second*3, time.Millisecond)
+		})
+	}
+}
+
+func Test_gracePeriodWithNonReadyPodCondition(t *testing.T) {
+	namespace := "test-ns"
+	type args struct {
+		collectorNotReadyGracePeriod time.Duration
+		collectorMap                 map[string]*allocation.Collector
+	}
+
+	tests := []struct {
+		name string
+		args args
+		want map[string]*allocation.Collector
+	}{
+		{
+			name: "collector healthiness check disabled",
+			args: args{
+				collectorNotReadyGracePeriod: 0 * time.Second,
+				collectorMap: map[string]*allocation.Collector{
+					"test-pod-ready": {
+						Name:     "test-pod-ready",
+						NodeName: "test-node",
+					},
+					"test-pod-non-ready-within-grace-period": {
+						Name:     "test-pod-non-ready-within-grace-period",
+						NodeName: "test-node",
+					},
+					"test-pod-non-ready-over-grace-period": {
+						Name:     "test-pod-non-ready-over-grace-period",
+						NodeName: "test-node",
+					},
+				},
+			},
+			want: map[string]*allocation.Collector{
+				"test-pod-ready": {
+					Name:     "test-pod-ready",
+					NodeName: "test-node",
+				},
+				"test-pod-non-ready-within-grace-period": {
+					Name:     "test-pod-non-ready-within-grace-period",
+					NodeName: "test-node",
+				},
+				"test-pod-non-ready-over-grace-period": {
+					Name:     "test-pod-non-ready-over-grace-period",
+					NodeName: "test-node",
+				},
+			},
+		},
+		{
+			name: "collector healthiness check enabled",
+			args: args{
+				collectorNotReadyGracePeriod: 30 * time.Second,
+				collectorMap: map[string]*allocation.Collector{
+					"test-pod-ready": {
+						Name:     "test-pod-ready",
+						NodeName: "test-node",
+					},
+					"test-pod-non-ready-within-grace-period": {
+						Name:     "test-pod-non-ready-within-grace-period",
+						NodeName: "test-node",
+					},
+					"test-pod-non-ready-over-grace-period": {
+						Name:     "test-pod-non-ready-over-grace-period",
+						NodeName: "test-node",
+					},
+				},
+			},
+			want: map[string]*allocation.Collector{
+				"test-pod-ready": {
+					Name:     "test-pod-ready",
+					NodeName: "test-node",
+				},
+				"test-pod-non-ready-within-grace-period": {
+					Name:     "test-pod-non-ready-within-grace-period",
+					NodeName: "test-node",
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			timeNow := time.Now()
+			podWatcher := getTestPodWatcher(tt.args.collectorNotReadyGracePeriod)
+			defer func() {
+				close(podWatcher.close)
+			}()
+			var actual map[string]*allocation.Collector
+			mapMutex := sync.Mutex{}
+			for _, k := range tt.args.collectorMap {
+				var p *v1.Pod
+				switch k.Name {
+				case "test-pod-ready":
+					p = podWithPodReadyConditionStatusAndLastTransitionTime(k.Name, v1.ConditionTrue, timeNow)
+				case "test-pod-non-ready-within-grace-period":
+					p = podWithPodReadyConditionStatusAndLastTransitionTime(k.Name, v1.ConditionFalse,
+						timeNow.Add(-1*podWatcher.collectorNotReadyGracePeriod).Add(podWatcher.collectorNotReadyGracePeriod/2))
+				case "test-pod-non-ready-over-grace-period":
+					p = podWithPodReadyConditionStatusAndLastTransitionTime(k.Name, v1.ConditionFalse,
+						timeNow.Add(-1*podWatcher.collectorNotReadyGracePeriod).Add(-podWatcher.collectorNotReadyGracePeriod/2))
+				}
+				_, err := podWatcher.k8sClient.CoreV1().Pods("test-ns").Create(context.Background(), p, metav1.CreateOptions{})
+				assert.NoError(t, err)
+			}
+
+			go func(podWatcher Watcher) {
+				err := podWatcher.Watch(namespace, &labelSelector, func(colMap map[string]*allocation.Collector) {
+					mapMutex.Lock()
+					defer mapMutex.Unlock()
+					actual = colMap
+				})
+				require.NoError(t, err)
+			}(podWatcher)
+
+			assert.EventuallyWithT(t, func(collect *assert.CollectT) {
+				mapMutex.Lock()
+				defer mapMutex.Unlock()
+				assert.Len(collect, actual, len(tt.want))
+				assert.Equal(collect, actual, tt.want)
+				assert.Equal(collect, testutil.ToFloat64(collectorsDiscovered), float64(len(actual)))
+			}, time.Second*3, time.Millisecond)
+		})
+	}
+}
+
 // this tests runWatch in the case of watcher channel closing.
 func Test_closeChannel(t *testing.T) {
-	podWatcher := getTestPodWatcher()
+	podWatcher := getTestPodWatcher(0 * time.Second)
 
 	var wg sync.WaitGroup
 	wg.Add(1)