Skip to content

Commit 111df9f

Browse files
valerian-rochemloiseleur
valerian-roche
and
mloiseleur
authored
feat(source): use transformers in pod informers to reduce memory footprint (#5596)
* feat: use transformers in pod informers to reduce memory footprint Add a transformer to the pods informer of the pod and service sources. Refs: #5595 Signed-off-by: Valerian Roche <[email protected]> * Do not use transformer when fqdnTemplate is set * Update source/pod_test.go Co-authored-by: Michel Loiseleur <[email protected]> --------- Signed-off-by: Valerian Roche <[email protected]> Co-authored-by: Michel Loiseleur <[email protected]>
1 parent 5805ea0 commit 111df9f

File tree

4 files changed

+327
-0
lines changed

4 files changed

+327
-0
lines changed

‎source/pod.go‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import (
2424

2525
log "github.com/sirupsen/logrus"
2626
corev1 "k8s.io/api/core/v1"
27+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
2728
"k8s.io/apimachinery/pkg/labels"
2829

2930
kubeinformers "k8s.io/client-go/informers"
@@ -76,6 +77,40 @@ func NewPodSource(
7677
}
7778

7879
_, _ = podInformer.Informer().AddEventHandler(informers.DefaultEventHandler())
80+
81+
if fqdnTemplate == "" {
82+
// Transformer is used to reduce the memory usage of the informer.
83+
// The pod informer will otherwise store a full in-memory, go-typed copy of all pod schemas in the cluster.
84+
// If watchList is not used it will not prevent memory bursts on the initial informer sync.
85+
// When fqdnTemplate is used the entire pod needs to be provided to the rendering call, but the informer itself becomes unneeded.
86+
podInformer.Informer().SetTransform(func(i interface{}) (interface{}, error) {
87+
pod, ok := i.(*corev1.Pod)
88+
if !ok {
89+
return nil, fmt.Errorf("object is not a pod")
90+
}
91+
if pod.UID == "" {
92+
// Pod was already transformed and we must be idempotent.
93+
return pod, nil
94+
}
95+
return &corev1.Pod{
96+
ObjectMeta: metav1.ObjectMeta{
97+
// Name/namespace must always be kept for the informer to work.
98+
Name: pod.Name,
99+
Namespace: pod.Namespace,
100+
// Used by the controller. This includes non-external-dns prefixed annotations.
101+
Annotations: pod.Annotations,
102+
},
103+
Spec: corev1.PodSpec{
104+
HostNetwork: pod.Spec.HostNetwork,
105+
NodeName: pod.Spec.NodeName,
106+
},
107+
Status: corev1.PodStatus{
108+
PodIP: pod.Status.PodIP,
109+
},
110+
}, nil
111+
})
112+
}
113+
79114
_, _ = nodeInformer.Informer().AddEventHandler(informers.DefaultEventHandler())
80115

81116
informerFactory.Start(ctx.Done())

‎source/pod_test.go‎

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"github.com/stretchr/testify/mock"
2828
"github.com/stretchr/testify/require"
2929
corev1 "k8s.io/api/core/v1"
30+
v1 "k8s.io/api/core/v1"
3031
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3132
corev1lister "k8s.io/client-go/listers/core/v1"
3233
"k8s.io/client-go/tools/cache"
@@ -997,3 +998,147 @@ func nodesFixturesIPv4() []*corev1.Node {
997998
},
998999
}
9991000
}
1001+
1002+
func TestPodTransformerInPodSource(t *testing.T) {
1003+
t.Run("transformer set", func(t *testing.T) {
1004+
ctx := t.Context()
1005+
fakeClient := fake.NewClientset()
1006+
1007+
pod := &v1.Pod{
1008+
Spec: v1.PodSpec{
1009+
Containers: []v1.Container{{
1010+
Name: "test",
1011+
}},
1012+
Hostname: "test-hostname",
1013+
NodeName: "test-node",
1014+
HostNetwork: true,
1015+
},
1016+
ObjectMeta: metav1.ObjectMeta{
1017+
Namespace: "test-ns",
1018+
Name: "test-name",
1019+
Labels: map[string]string{
1020+
"label1": "value1",
1021+
"label2": "value2",
1022+
"label3": "value3",
1023+
},
1024+
Annotations: map[string]string{
1025+
"user-annotation": "value",
1026+
"external-dns.alpha.kubernetes.io/hostname": "test-hostname",
1027+
"external-dns.alpha.kubernetes.io/random": "value",
1028+
"other/annotation": "value",
1029+
},
1030+
UID: "someuid",
1031+
},
1032+
Status: v1.PodStatus{
1033+
PodIP: "127.0.0.1",
1034+
HostIP: "127.0.0.2",
1035+
Conditions: []v1.PodCondition{{
1036+
Type: v1.PodReady,
1037+
Status: v1.ConditionTrue,
1038+
}, {
1039+
Type: v1.ContainersReady,
1040+
Status: v1.ConditionFalse,
1041+
}},
1042+
},
1043+
}
1044+
1045+
_, err := fakeClient.CoreV1().Pods(pod.Namespace).Create(context.Background(), pod, metav1.CreateOptions{})
1046+
require.NoError(t, err)
1047+
1048+
// Should not error when creating the source
1049+
src, err := NewPodSource(ctx, fakeClient, "", "", false, "", "", false, "", nil)
1050+
require.NoError(t, err)
1051+
ps, ok := src.(*podSource)
1052+
require.True(t, ok)
1053+
1054+
retrieved, err := ps.podInformer.Lister().Pods("test-ns").Get("test-name")
1055+
require.NoError(t, err)
1056+
1057+
// Metadata
1058+
assert.Equal(t, "test-name", retrieved.Name)
1059+
assert.Equal(t, "test-ns", retrieved.Namespace)
1060+
assert.Empty(t, retrieved.UID)
1061+
assert.Empty(t, retrieved.Labels)
1062+
// Filtered
1063+
assert.Equal(t, map[string]string{
1064+
"user-annotation": "value",
1065+
"external-dns.alpha.kubernetes.io/hostname": "test-hostname",
1066+
"external-dns.alpha.kubernetes.io/random": "value",
1067+
"other/annotation": "value",
1068+
}, retrieved.Annotations)
1069+
1070+
// Spec
1071+
assert.Empty(t, retrieved.Spec.Containers)
1072+
assert.Empty(t, retrieved.Spec.Hostname)
1073+
assert.Equal(t, "test-node", retrieved.Spec.NodeName)
1074+
assert.True(t, retrieved.Spec.HostNetwork)
1075+
1076+
// Status
1077+
assert.Empty(t, retrieved.Status.ContainerStatuses)
1078+
assert.Empty(t, retrieved.Status.InitContainerStatuses)
1079+
assert.Empty(t, retrieved.Status.HostIP)
1080+
assert.Equal(t, "127.0.0.1", retrieved.Status.PodIP)
1081+
assert.Empty(t, retrieved.Status.Conditions)
1082+
})
1083+
1084+
t.Run("transformer is not used when fqdnTemplate is set", func(t *testing.T) {
1085+
ctx := t.Context()
1086+
fakeClient := fake.NewClientset()
1087+
1088+
pod := &v1.Pod{
1089+
Spec: v1.PodSpec{
1090+
Containers: []v1.Container{{
1091+
Name: "test",
1092+
}},
1093+
Hostname: "test-hostname",
1094+
NodeName: "test-node",
1095+
HostNetwork: true,
1096+
},
1097+
ObjectMeta: metav1.ObjectMeta{
1098+
Namespace: "test-ns",
1099+
Name: "test-name",
1100+
Labels: map[string]string{
1101+
"label1": "value1",
1102+
"label2": "value2",
1103+
"label3": "value3",
1104+
},
1105+
Annotations: map[string]string{
1106+
"user-annotation": "value",
1107+
"external-dns.alpha.kubernetes.io/hostname": "test-hostname",
1108+
"external-dns.alpha.kubernetes.io/random": "value",
1109+
"other/annotation": "value",
1110+
},
1111+
UID: "someuid",
1112+
},
1113+
Status: v1.PodStatus{
1114+
PodIP: "127.0.0.1",
1115+
HostIP: "127.0.0.2",
1116+
Conditions: []v1.PodCondition{{
1117+
Type: v1.PodReady,
1118+
Status: v1.ConditionTrue,
1119+
}, {
1120+
Type: v1.ContainersReady,
1121+
Status: v1.ConditionFalse,
1122+
}},
1123+
},
1124+
}
1125+
1126+
_, err := fakeClient.CoreV1().Pods(pod.Namespace).Create(context.Background(), pod, metav1.CreateOptions{})
1127+
require.NoError(t, err)
1128+
1129+
// Should not error when creating the source
1130+
src, err := NewPodSource(ctx, fakeClient, "", "", false, "", "template", false, "", nil)
1131+
require.NoError(t, err)
1132+
ps, ok := src.(*podSource)
1133+
require.True(t, ok)
1134+
1135+
retrieved, err := ps.podInformer.Lister().Pods("test-ns").Get("test-name")
1136+
require.NoError(t, err)
1137+
1138+
// Metadata
1139+
assert.Equal(t, "test-name", retrieved.Name)
1140+
assert.Equal(t, "test-ns", retrieved.Namespace)
1141+
assert.NotEmpty(t, retrieved.UID)
1142+
assert.NotEmpty(t, retrieved.Labels)
1143+
})
1144+
}

‎source/service.go‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import (
2929
log "github.com/sirupsen/logrus"
3030
v1 "k8s.io/api/core/v1"
3131
discoveryv1 "k8s.io/api/discovery/v1"
32+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3233
"k8s.io/apimachinery/pkg/labels"
3334
"k8s.io/apimachinery/pkg/types"
3435
kubeinformers "k8s.io/client-go/informers"
@@ -134,6 +135,48 @@ func NewServiceSource(ctx context.Context, kubeClient kubernetes.Interface, name
134135
if err != nil {
135136
return nil, err
136137
}
138+
139+
// Transformer is used to reduce the memory usage of the informer.
140+
// The pod informer will otherwise store a full in-memory, go-typed copy of all pod schemas in the cluster.
141+
// If watchList is not used it will not prevent memory bursts on the initial informer sync.
142+
podInformer.Informer().SetTransform(func(i interface{}) (interface{}, error) {
143+
pod, ok := i.(*v1.Pod)
144+
if !ok {
145+
return nil, fmt.Errorf("object is not a pod")
146+
}
147+
if pod.UID == "" {
148+
// Pod was already transformed and we must be idempotent.
149+
return pod, nil
150+
}
151+
152+
// All pod level annotations we're interested in start with a common prefix
153+
podAnnotations := map[string]string{}
154+
for key, value := range pod.Annotations {
155+
if strings.HasPrefix(key, annotations.AnnotationKeyPrefix) {
156+
podAnnotations[key] = value
157+
}
158+
}
159+
return &v1.Pod{
160+
ObjectMeta: metav1.ObjectMeta{
161+
// Name/namespace must always be kept for the informer to work.
162+
Name: pod.Name,
163+
Namespace: pod.Namespace,
164+
// Used to match services.
165+
Labels: pod.Labels,
166+
Annotations: podAnnotations,
167+
DeletionTimestamp: pod.DeletionTimestamp,
168+
},
169+
Spec: v1.PodSpec{
170+
Hostname: pod.Spec.Hostname,
171+
NodeName: pod.Spec.NodeName,
172+
},
173+
Status: v1.PodStatus{
174+
HostIP: pod.Status.HostIP,
175+
Phase: pod.Status.Phase,
176+
Conditions: pod.Status.Conditions,
177+
},
178+
}, nil
179+
})
137180
}
138181

139182
var nodeInformer coreinformers.NodeInformer

‎source/service_test.go‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4702,6 +4702,110 @@ func TestEndpointSlicesIndexer(t *testing.T) {
47024702
})
47034703
}
47044704

4705+
func TestPodTransformerInServiceSource(t *testing.T) {
4706+
ctx := t.Context()
4707+
fakeClient := fake.NewClientset()
4708+
4709+
pod := &v1.Pod{
4710+
Spec: v1.PodSpec{
4711+
Containers: []v1.Container{{
4712+
Name: "test",
4713+
}},
4714+
Hostname: "test-hostname",
4715+
NodeName: "test-node",
4716+
},
4717+
ObjectMeta: metav1.ObjectMeta{
4718+
Namespace: "test-ns",
4719+
Name: "test-name",
4720+
Labels: map[string]string{
4721+
"label1": "value1",
4722+
"label2": "value2",
4723+
"label3": "value3",
4724+
},
4725+
Annotations: map[string]string{
4726+
"user-annotation": "value",
4727+
"external-dns.alpha.kubernetes.io/hostname": "test-hostname",
4728+
"external-dns.alpha.kubernetes.io/random": "value",
4729+
"other/annotation": "value",
4730+
},
4731+
UID: "someuid",
4732+
},
4733+
Status: v1.PodStatus{
4734+
PodIP: "127.0.0.1",
4735+
HostIP: "127.0.0.2",
4736+
Conditions: []v1.PodCondition{{
4737+
Type: v1.PodReady,
4738+
Status: v1.ConditionTrue,
4739+
}, {
4740+
Type: v1.ContainersReady,
4741+
Status: v1.ConditionFalse,
4742+
}},
4743+
},
4744+
}
4745+
4746+
_, err := fakeClient.CoreV1().Pods(pod.Namespace).Create(context.Background(), pod, metav1.CreateOptions{})
4747+
require.NoError(t, err)
4748+
4749+
// Should not error when creating the source
4750+
src, err := NewServiceSource(
4751+
ctx,
4752+
fakeClient,
4753+
"",
4754+
"",
4755+
"{{.Name}}",
4756+
false,
4757+
"",
4758+
false,
4759+
false,
4760+
false,
4761+
[]string{},
4762+
false,
4763+
labels.Everything(),
4764+
false,
4765+
false,
4766+
false,
4767+
)
4768+
require.NoError(t, err)
4769+
ss, ok := src.(*serviceSource)
4770+
require.True(t, ok)
4771+
4772+
retrieved, err := ss.podInformer.Lister().Pods("test-ns").Get("test-name")
4773+
require.NoError(t, err)
4774+
4775+
// Metadata
4776+
assert.Equal(t, "test-name", retrieved.Name)
4777+
assert.Equal(t, "test-ns", retrieved.Namespace)
4778+
assert.Empty(t, retrieved.UID)
4779+
assert.Equal(t, map[string]string{
4780+
"label1": "value1",
4781+
"label2": "value2",
4782+
"label3": "value3",
4783+
}, retrieved.Labels)
4784+
// Filtered
4785+
assert.Equal(t, map[string]string{
4786+
"external-dns.alpha.kubernetes.io/hostname": "test-hostname",
4787+
"external-dns.alpha.kubernetes.io/random": "value",
4788+
}, retrieved.Annotations)
4789+
4790+
// Spec
4791+
assert.Empty(t, retrieved.Spec.Containers)
4792+
assert.Equal(t, "test-hostname", retrieved.Spec.Hostname)
4793+
assert.Equal(t, "test-node", retrieved.Spec.NodeName)
4794+
4795+
// Status
4796+
assert.Empty(t, retrieved.Status.ContainerStatuses)
4797+
assert.Empty(t, retrieved.Status.InitContainerStatuses)
4798+
assert.Equal(t, "127.0.0.2", retrieved.Status.HostIP)
4799+
assert.Empty(t, retrieved.Status.PodIP)
4800+
assert.ElementsMatch(t, []v1.PodCondition{{
4801+
Type: v1.PodReady,
4802+
Status: v1.ConditionTrue,
4803+
}, {
4804+
Type: v1.ContainersReady,
4805+
Status: v1.ConditionFalse,
4806+
}}, retrieved.Status.Conditions)
4807+
}
4808+
47054809
// createTestServicesByType creates the requested number of services per type in the given namespace.
47064810
func createTestServicesByType(namespace string, typeCounts map[v1.ServiceType]int) []*v1.Service {
47074811
var services []*v1.Service

0 commit comments

Comments
 (0)