Revision 3e306193515f9ed51da70cba87d41720dac17d49 authored by Julian Wiedmann on 30 May 2024, 10:02:25 UTC, committed by Julian Wiedmann on 14 June 2024, 04:25:14 UTC
Declutter the maps.h header, and reduce the usage of HAVE_ENCAP. Signed-off-by: Julian Wiedmann <jwi@isovalent.com>
1 parent dfb6b94
pod.go
// SPDX-License-Identifier: Apache-2.0
// Copyright Authors of Cilium
package watchers
import (
"context"
"fmt"
"sync"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/client-go/tools/cache"
operatorOption "github.com/cilium/cilium/operator/option"
k8sClient "github.com/cilium/cilium/pkg/k8s/client"
"github.com/cilium/cilium/pkg/k8s/informer"
slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
slim_metav1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/apis/meta/v1"
k8sUtils "github.com/cilium/cilium/pkg/k8s/utils"
)
const PodNodeNameIndex = "pod-node"
var (
// PodStore has a minimal copy of all pods running in the cluster.
// Warning: The pods stored in the cache are not intended to be used for Update
// operations in k8s as some of its fields are not populated.
PodStore cache.Store
// PodStoreSynced is closed once the PodStore is synced with k8s.
PodStoreSynced = make(chan struct{})
// UnmanagedPodStore has a minimal copy of the unmanaged pods running
// in the cluster.
// Warning: The pods stored in the cache are not intended to be used for Update
// operations in k8s as some of its fields are not populated.
UnmanagedPodStore cache.Store
// UnmanagedPodStoreSynced is closed once the UnmanagedKubeDNSPodStore is synced
// with k8s.
UnmanagedPodStoreSynced = make(chan struct{})
)
// podNodeNameIndexFunc indexes pods by node name
func podNodeNameIndexFunc(obj interface{}) ([]string, error) {
pod := obj.(*slim_corev1.Pod)
if pod.Spec.NodeName != "" {
return []string{pod.Spec.NodeName}, nil
}
return []string{}, nil
}
func PodsInit(ctx context.Context, wg *sync.WaitGroup, clientset k8sClient.Clientset) {
var podInformer cache.Controller
PodStore = cache.NewIndexer(cache.DeletionHandlingMetaNamespaceKeyFunc, cache.Indexers{
PodNodeNameIndex: podNodeNameIndexFunc,
})
podInformer = informer.NewInformerWithStore(
k8sUtils.ListerWatcherWithFields(
k8sUtils.ListerWatcherFromTyped[*slim_corev1.PodList](clientset.Slim().CoreV1().Pods("")),
fields.Everything()),
&slim_corev1.Pod{},
0,
cache.ResourceEventHandlerFuncs{},
transformToPod,
PodStore,
)
wg.Add(1)
go func() {
defer wg.Done()
podInformer.Run(ctx.Done())
}()
cache.WaitForCacheSync(ctx.Done(), podInformer.HasSynced)
}
// transformToPod stores a minimal version of the pod as it is only intended
// for it to check if a pod is running in the cluster or not. The stored pod
// should not be used to update an existing pod in the kubernetes cluster.
// If the given obj can't be cast into either Pod nor DeletedFinalStateUnknown,
// an error is returned.
func transformToPod(obj interface{}) (interface{}, error) {
switch concreteObj := obj.(type) {
case *slim_corev1.Pod:
p := &slim_corev1.Pod{
TypeMeta: concreteObj.TypeMeta,
ObjectMeta: slim_metav1.ObjectMeta{
Name: concreteObj.Name,
Namespace: concreteObj.Namespace,
ResourceVersion: concreteObj.ResourceVersion,
},
Spec: slim_corev1.PodSpec{
NodeName: concreteObj.Spec.NodeName,
},
Status: slim_corev1.PodStatus{
Phase: concreteObj.Status.Phase,
},
}
*concreteObj = slim_corev1.Pod{}
return p, nil
case cache.DeletedFinalStateUnknown:
pod, ok := concreteObj.Obj.(*slim_corev1.Pod)
if !ok {
return nil, fmt.Errorf("unknown object type %T", concreteObj.Obj)
}
dfsu := cache.DeletedFinalStateUnknown{
Key: concreteObj.Key,
Obj: &slim_corev1.Pod{
TypeMeta: pod.TypeMeta,
ObjectMeta: slim_metav1.ObjectMeta{
Name: pod.Name,
Namespace: pod.Namespace,
ResourceVersion: pod.ResourceVersion,
},
Spec: slim_corev1.PodSpec{
NodeName: pod.Spec.NodeName,
},
Status: slim_corev1.PodStatus{
Phase: pod.Status.Phase,
},
},
}
// Small GC optimization
*pod = slim_corev1.Pod{}
return dfsu, nil
default:
return nil, fmt.Errorf("unknown object type %T", concreteObj)
}
}
func UnmanagedPodsInit(ctx context.Context, wg *sync.WaitGroup, clientset k8sClient.Clientset) {
var unmanagedPodInformer cache.Controller
UnmanagedPodStore, unmanagedPodInformer = informer.NewInformer(
k8sUtils.ListerWatcherWithModifier(
k8sUtils.ListerWatcherFromTyped[*slim_corev1.PodList](clientset.Slim().CoreV1().Pods("")),
func(options *metav1.ListOptions) {
options.FieldSelector = "status.phase=Running"
options.LabelSelector = operatorOption.Config.PodRestartSelector
}),
&slim_corev1.Pod{},
0,
cache.ResourceEventHandlerFuncs{},
TransformToUnmanagedPod,
)
wg.Add(1)
go func() {
defer wg.Done()
unmanagedPodInformer.Run(ctx.Done())
}()
cache.WaitForCacheSync(ctx.Done(), unmanagedPodInformer.HasSynced)
}
func TransformToUnmanagedPod(obj interface{}) (interface{}, error) {
switch concreteObj := obj.(type) {
case *slim_corev1.Pod:
p := &slim_corev1.Pod{
TypeMeta: concreteObj.TypeMeta,
ObjectMeta: slim_metav1.ObjectMeta{
Name: concreteObj.Name,
Namespace: concreteObj.Namespace,
ResourceVersion: concreteObj.ResourceVersion,
},
Spec: slim_corev1.PodSpec{
HostNetwork: concreteObj.Spec.HostNetwork,
},
Status: slim_corev1.PodStatus{
StartTime: concreteObj.Status.StartTime,
},
}
*concreteObj = slim_corev1.Pod{}
return p, nil
case cache.DeletedFinalStateUnknown:
pod, ok := concreteObj.Obj.(*slim_corev1.Pod)
if !ok {
return nil, fmt.Errorf("unknown object type %T", concreteObj.Obj)
}
dfsu := cache.DeletedFinalStateUnknown{
Key: concreteObj.Key,
Obj: &slim_corev1.Pod{
TypeMeta: pod.TypeMeta,
ObjectMeta: slim_metav1.ObjectMeta{
Name: pod.Name,
Namespace: pod.Namespace,
ResourceVersion: pod.ResourceVersion,
},
Spec: slim_corev1.PodSpec{
HostNetwork: pod.Spec.HostNetwork,
},
Status: slim_corev1.PodStatus{
StartTime: pod.Status.StartTime,
},
},
}
// Small GC optimization
*pod = slim_corev1.Pod{}
return dfsu, nil
default:
return nil, fmt.Errorf("unknown object type %T", concreteObj)
}
}
![swh spinner](/static/img/swh-spinner.gif)
Computing file changes ...