emansije/kube-metrics-adapter
1
0
Fork 0
forked from yusheng-guo/kube-metrics-adapter
Code Pull Requests Activity

Merge pull request #312 from adutchak-x/feature/stop-processing-non-runing-pods

Browse Source 
Feature/stop processing non runing pods
This commit is contained in:
Jonathan Juares Beber
2021-05-06 10:35:17 +02:00
committed by GitHub
parent 726c1e691d e16bacb24e
commit 7044e79b08
2 changed files with 59 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
pkg/collector/pod_collector.go
View File

@ -99,15 +99,18 @@ 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.Debugf("Skipping metrics collection for pod %s/%s because it is being terminated (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++
c.logger.Warnf("Skipping metrics collection for pod %s/%s because it's status is not Ready.", pod.Namespace, pod.Name)
c.logger.Debugf("Skipping metrics collection for pod %s/%s because it's status is not Ready.", pod.Namespace, pod.Name)
}
}

56
pkg/collector/pod_collector_test.go
View File

@ -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,49 @@ 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 +225,7 @@ 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 +240,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)
}