diff --git a/pkg/collector/pod_collector.go b/pkg/collector/pod_collector.go index e5cb2db..d4317ee 100644 --- a/pkg/collector/pod_collector.go +++ b/pkg/collector/pod_collector.go @@ -99,11 +99,14 @@ func (c *PodCollector) GetMetrics() ([]CollectedMetric, error) { isPodReady, podReadyAge := GetPodReadyAge(pod) if isPodReady { - if podReadyAge >= c.minPodReadyAge { - go c.getPodMetric(pod, ch, errCh) - } else { + if pod.DeletionTimestamp != nil { + skippedPodsCount++ + c.logger.Warnf("Skipping metrics collection for pod %s/%s because it is seems to be scheduled for termination (DeletionTimestamp: %s)", pod.Namespace, pod.Name, pod.DeletionTimestamp) + } else if podReadyAge < c.minPodReadyAge { skippedPodsCount++ c.logger.Warnf("Skipping metrics collection for pod %s/%s because it's ready age is %s and min-pod-ready-age is set to %s", pod.Namespace, pod.Name, podReadyAge, c.minPodReadyAge) + } else { + go c.getPodMetric(pod, ch, errCh) } } else { skippedPodsCount++ diff --git a/pkg/collector/pod_collector_test.go b/pkg/collector/pod_collector_test.go index c7bbe43..d35c10f 100644 --- a/pkg/collector/pod_collector_test.go +++ b/pkg/collector/pod_collector_test.go @@ -48,7 +48,8 @@ func TestPodCollector(t *testing.T) { lastReadyTransitionTimeTimestamp := v1.NewTime(time.Now().Add(time.Duration(-30) * time.Second)) minPodReadyAge := time.Duration(0 * time.Second) podCondition := corev1.PodCondition{Type: corev1.PodReady, Status: corev1.ConditionTrue, LastTransitionTime: lastReadyTransitionTimeTimestamp} - makeTestPods(t, host, port, "test-metric", client, 5, podCondition) + podDeletionTimestamp := time.Time{} + makeTestPods(t, host, port, "test-metric", client, 5, podCondition,podDeletionTimestamp) testHPA := makeTestHPA(t, client) testConfig := makeTestConfig(port, minPodReadyAge) collector, err := plugin.NewCollector(testHPA, testConfig, testInterval) @@ -87,7 +88,8 @@ func TestPodCollectorWithMinPodReadyAge(t *testing.T) { // Pods that are not older that 60 seconds (all in this case) should not be processed minPodReadyAge := time.Duration(60 * time.Second) podCondition := corev1.PodCondition{Type: corev1.PodReady, Status: corev1.ConditionTrue, LastTransitionTime: lastReadyTransitionTimeTimestamp} - makeTestPods(t, host, port, "test-metric", client, 5, podCondition) + podDeletionTimestamp := time.Time{} + makeTestPods(t, host, port, "test-metric", client, 5, podCondition, podDeletionTimestamp) testHPA := makeTestHPA(t, client) testConfig := makeTestConfig(port, minPodReadyAge) collector, err := plugin.NewCollector(testHPA, testConfig, testInterval) @@ -123,9 +125,50 @@ func TestPodCollectorWithPodCondition(t *testing.T) { host, port, metricsHandler := makeTestHTTPServer(t, tc.metrics) lastScheduledTransitionTimeTimestamp := v1.NewTime(time.Now().Add(time.Duration(-30) * time.Second)) minPodReadyAge := time.Duration(0 * time.Second) + podDeletionTimestamp := time.Time{} //Pods in state corev1.PodReady == corev1.ConditionFalse should not be processed podCondition := corev1.PodCondition{Type: corev1.PodReady, Status: corev1.ConditionFalse, LastTransitionTime: lastScheduledTransitionTimeTimestamp} - makeTestPods(t, host, port, "test-metric", client, 5, podCondition) + makeTestPods(t, host, port, "test-metric", client, 5, podCondition, podDeletionTimestamp) + testHPA := makeTestHPA(t, client) + testConfig := makeTestConfig(port, minPodReadyAge) + collector, err := plugin.NewCollector(testHPA, testConfig, testInterval) + require.NoError(t, err) + metrics, err := collector.GetMetrics() + require.NoError(t, err) + require.Equal(t, len(metrics), int(metricsHandler.calledCounter)) + var values []int64 + for _, m := range metrics { + values = append(values, m.Custom.Value.Value()) + } + require.ElementsMatch(t, tc.result, values) + }) + } +} + + +func TestPodCollectorWithPodTerminatingCondition(t *testing.T) { + for _, tc := range []struct { + name string + metrics [][]int64 + result []int64 + }{ + { + name: "simple-with-pod-condition", + metrics: [][]int64{{1}, {3}, {8}, {5}, {2}}, + result: []int64{}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + client := fake.NewSimpleClientset() + plugin := NewPodCollectorPlugin(client) + makeTestDeployment(t, client) + host, port, metricsHandler := makeTestHTTPServer(t, tc.metrics) + lastScheduledTransitionTimeTimestamp := v1.NewTime(time.Now().Add(time.Duration(-30) * time.Second)) + minPodReadyAge := time.Duration(0 * time.Second) + //Pods with podDeletionTimestamp should not be processed + podDeletionTimestamp := time.Now() + podCondition := corev1.PodCondition{Type: corev1.PodReady, Status: corev1.ConditionTrue, LastTransitionTime: lastScheduledTransitionTimeTimestamp} + makeTestPods(t, host, port, "test-metric", client, 5, podCondition, podDeletionTimestamp) testHPA := makeTestHPA(t, client) testConfig := makeTestConfig(port, minPodReadyAge) collector, err := plugin.NewCollector(testHPA, testConfig, testInterval) @@ -183,7 +226,9 @@ func makeTestConfig(port string, minPodReadyAge time.Duration) *MetricConfig { } } -func makeTestPods(t *testing.T, testServer string, metricName string, port string, client kubernetes.Interface, replicas int, podCondition corev1.PodCondition) { +func makeTestPods(t *testing.T, testServer string, metricName string, port string, client kubernetes.Interface, replicas int, podCondition corev1.PodCondition, podDeletionTimestamp time.Time) { + + for i := 0; i < replicas; i++ { testPod := &corev1.Pod{ ObjectMeta: v1.ObjectMeta{ @@ -198,6 +243,12 @@ func makeTestPods(t *testing.T, testServer string, metricName string, port strin Conditions: []corev1.PodCondition{podCondition}, }, } + + if podDeletionTimestamp.IsZero(){ + testPod.ObjectMeta.DeletionTimestamp = nil + } else { + testPod.ObjectMeta.DeletionTimestamp = &v1.Time{Time: podDeletionTimestamp} + } _, err := client.CoreV1().Pods(testNamespace).Create(context.TODO(), testPod, v1.CreateOptions{}) require.NoError(t, err) }