123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611 |
- package endpoint
- import (
- "context"
- "fmt"
- "reflect"
- "strconv"
- "time"
- v1 "k8s.io/api/core/v1"
- apiequality "k8s.io/apimachinery/pkg/api/equality"
- "k8s.io/apimachinery/pkg/api/errors"
- metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
- "k8s.io/apimachinery/pkg/labels"
- utilruntime "k8s.io/apimachinery/pkg/util/runtime"
- "k8s.io/apimachinery/pkg/util/wait"
- utilfeature "k8s.io/apiserver/pkg/util/feature"
- coreinformers "k8s.io/client-go/informers/core/v1"
- clientset "k8s.io/client-go/kubernetes"
- "k8s.io/client-go/kubernetes/scheme"
- v1core "k8s.io/client-go/kubernetes/typed/core/v1"
- corelisters "k8s.io/client-go/listers/core/v1"
- "k8s.io/client-go/tools/cache"
- "k8s.io/client-go/tools/leaderelection/resourcelock"
- "k8s.io/client-go/tools/record"
- "k8s.io/client-go/util/workqueue"
- "k8s.io/component-base/metrics/prometheus/ratelimiter"
- "k8s.io/klog"
- "k8s.io/kubernetes/pkg/api/v1/endpoints"
- podutil "k8s.io/kubernetes/pkg/api/v1/pod"
- api "k8s.io/kubernetes/pkg/apis/core"
- helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
- "k8s.io/kubernetes/pkg/controller"
- endpointutil "k8s.io/kubernetes/pkg/controller/util/endpoint"
- "k8s.io/kubernetes/pkg/features"
- utillabels "k8s.io/kubernetes/pkg/util/labels"
- utilnet "k8s.io/utils/net"
- )
- const (
-
-
-
-
-
- maxRetries = 15
-
-
-
-
-
-
-
-
-
-
-
-
- TolerateUnreadyEndpointsAnnotation = "service.alpha.kubernetes.io/tolerate-unready-endpoints"
- )
- func NewEndpointController(podInformer coreinformers.PodInformer, serviceInformer coreinformers.ServiceInformer,
- endpointsInformer coreinformers.EndpointsInformer, client clientset.Interface, endpointUpdatesBatchPeriod time.Duration) *EndpointController {
- broadcaster := record.NewBroadcaster()
- broadcaster.StartLogging(klog.Infof)
- broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: client.CoreV1().Events("")})
- recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "endpoint-controller"})
- if client != nil && client.CoreV1().RESTClient().GetRateLimiter() != nil {
- ratelimiter.RegisterMetricAndTrackRateLimiterUsage("endpoint_controller", client.CoreV1().RESTClient().GetRateLimiter())
- }
- e := &EndpointController{
- client: client,
- queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "endpoint"),
- workerLoopPeriod: time.Second,
- }
- serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
- AddFunc: e.onServiceUpdate,
- UpdateFunc: func(old, cur interface{}) {
- e.onServiceUpdate(cur)
- },
- DeleteFunc: e.onServiceDelete,
- })
- e.serviceLister = serviceInformer.Lister()
- e.servicesSynced = serviceInformer.Informer().HasSynced
- podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
- AddFunc: e.addPod,
- UpdateFunc: e.updatePod,
- DeleteFunc: e.deletePod,
- })
- e.podLister = podInformer.Lister()
- e.podsSynced = podInformer.Informer().HasSynced
- e.endpointsLister = endpointsInformer.Lister()
- e.endpointsSynced = endpointsInformer.Informer().HasSynced
- e.triggerTimeTracker = endpointutil.NewTriggerTimeTracker()
- e.eventBroadcaster = broadcaster
- e.eventRecorder = recorder
- e.endpointUpdatesBatchPeriod = endpointUpdatesBatchPeriod
- e.serviceSelectorCache = endpointutil.NewServiceSelectorCache()
- return e
- }
- type EndpointController struct {
- client clientset.Interface
- eventBroadcaster record.EventBroadcaster
- eventRecorder record.EventRecorder
-
-
- serviceLister corelisters.ServiceLister
-
-
- servicesSynced cache.InformerSynced
-
-
- podLister corelisters.PodLister
-
-
- podsSynced cache.InformerSynced
-
-
- endpointsLister corelisters.EndpointsLister
-
-
- endpointsSynced cache.InformerSynced
-
-
-
-
-
- queue workqueue.RateLimitingInterface
-
- workerLoopPeriod time.Duration
-
-
- triggerTimeTracker *endpointutil.TriggerTimeTracker
- endpointUpdatesBatchPeriod time.Duration
-
-
- serviceSelectorCache *endpointutil.ServiceSelectorCache
- }
- func (e *EndpointController) Run(workers int, stopCh <-chan struct{}) {
- defer utilruntime.HandleCrash()
- defer e.queue.ShutDown()
- klog.Infof("Starting endpoint controller")
- defer klog.Infof("Shutting down endpoint controller")
- if !cache.WaitForNamedCacheSync("endpoint", stopCh, e.podsSynced, e.servicesSynced, e.endpointsSynced) {
- return
- }
- for i := 0; i < workers; i++ {
- go wait.Until(e.worker, e.workerLoopPeriod, stopCh)
- }
- go func() {
- defer utilruntime.HandleCrash()
- e.checkLeftoverEndpoints()
- }()
- <-stopCh
- }
- func (e *EndpointController) addPod(obj interface{}) {
- pod := obj.(*v1.Pod)
- services, err := e.serviceSelectorCache.GetPodServiceMemberships(e.serviceLister, pod)
- if err != nil {
- utilruntime.HandleError(fmt.Errorf("Unable to get pod %s/%s's service memberships: %v", pod.Namespace, pod.Name, err))
- return
- }
- for key := range services {
- e.queue.AddAfter(key, e.endpointUpdatesBatchPeriod)
- }
- }
- func podToEndpointAddressForService(svc *v1.Service, pod *v1.Pod) (*v1.EndpointAddress, error) {
- if !utilfeature.DefaultFeatureGate.Enabled(features.IPv6DualStack) {
- return podToEndpointAddress(pod), nil
- }
-
-
-
- ipv6ClusterIP := utilnet.IsIPv6String(svc.Spec.ClusterIP)
- for _, podIP := range pod.Status.PodIPs {
- ipv6PodIP := utilnet.IsIPv6String(podIP.IP)
-
-
-
- if ipv6ClusterIP == ipv6PodIP {
- return &v1.EndpointAddress{
- IP: podIP.IP,
- NodeName: &pod.Spec.NodeName,
- TargetRef: &v1.ObjectReference{
- Kind: "Pod",
- Namespace: pod.ObjectMeta.Namespace,
- Name: pod.ObjectMeta.Name,
- UID: pod.ObjectMeta.UID,
- ResourceVersion: pod.ObjectMeta.ResourceVersion,
- }}, nil
- }
- }
- return nil, fmt.Errorf("failed to find a matching endpoint for service %v", svc.Name)
- }
- func podToEndpointAddress(pod *v1.Pod) *v1.EndpointAddress {
- return &v1.EndpointAddress{
- IP: pod.Status.PodIP,
- NodeName: &pod.Spec.NodeName,
- TargetRef: &v1.ObjectReference{
- Kind: "Pod",
- Namespace: pod.ObjectMeta.Namespace,
- Name: pod.ObjectMeta.Name,
- UID: pod.ObjectMeta.UID,
- ResourceVersion: pod.ObjectMeta.ResourceVersion,
- }}
- }
- func endpointChanged(pod1, pod2 *v1.Pod) bool {
- endpointAddress1 := podToEndpointAddress(pod1)
- endpointAddress2 := podToEndpointAddress(pod2)
- endpointAddress1.TargetRef.ResourceVersion = ""
- endpointAddress2.TargetRef.ResourceVersion = ""
- return !reflect.DeepEqual(endpointAddress1, endpointAddress2)
- }
- func (e *EndpointController) updatePod(old, cur interface{}) {
- services := endpointutil.GetServicesToUpdateOnPodChange(e.serviceLister, e.serviceSelectorCache, old, cur, endpointChanged)
- for key := range services {
- e.queue.AddAfter(key, e.endpointUpdatesBatchPeriod)
- }
- }
- func (e *EndpointController) deletePod(obj interface{}) {
- pod := endpointutil.GetPodFromDeleteAction(obj)
- if pod != nil {
- e.addPod(pod)
- }
- }
- func (e *EndpointController) onServiceUpdate(obj interface{}) {
- key, err := controller.KeyFunc(obj)
- if err != nil {
- utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %+v: %v", obj, err))
- return
- }
- _ = e.serviceSelectorCache.Update(key, obj.(*v1.Service).Spec.Selector)
- e.queue.Add(key)
- }
- func (e *EndpointController) onServiceDelete(obj interface{}) {
- key, err := controller.KeyFunc(obj)
- if err != nil {
- utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %+v: %v", obj, err))
- return
- }
- e.serviceSelectorCache.Delete(key)
- e.queue.Add(key)
- }
- func (e *EndpointController) worker() {
- for e.processNextWorkItem() {
- }
- }
- func (e *EndpointController) processNextWorkItem() bool {
- eKey, quit := e.queue.Get()
- if quit {
- return false
- }
- defer e.queue.Done(eKey)
- err := e.syncService(eKey.(string))
- e.handleErr(err, eKey)
- return true
- }
- func (e *EndpointController) handleErr(err error, key interface{}) {
- if err == nil {
- e.queue.Forget(key)
- return
- }
- if e.queue.NumRequeues(key) < maxRetries {
- klog.V(2).Infof("Error syncing endpoints for service %q, retrying. Error: %v", key, err)
- e.queue.AddRateLimited(key)
- return
- }
- klog.Warningf("Dropping service %q out of the queue: %v", key, err)
- e.queue.Forget(key)
- utilruntime.HandleError(err)
- }
- func (e *EndpointController) syncService(key string) error {
- startTime := time.Now()
- defer func() {
- klog.V(4).Infof("Finished syncing service %q endpoints. (%v)", key, time.Since(startTime))
- }()
- namespace, name, err := cache.SplitMetaNamespaceKey(key)
- if err != nil {
- return err
- }
- service, err := e.serviceLister.Services(namespace).Get(name)
- if err != nil {
- if !errors.IsNotFound(err) {
- return err
- }
-
-
-
-
-
- err = e.client.CoreV1().Endpoints(namespace).Delete(context.TODO(), name, nil)
- if err != nil && !errors.IsNotFound(err) {
- return err
- }
- e.triggerTimeTracker.DeleteService(namespace, name)
- return nil
- }
- if service.Spec.Selector == nil {
-
-
- return nil
- }
- klog.V(5).Infof("About to update endpoints for service %q", key)
- pods, err := e.podLister.Pods(service.Namespace).List(labels.Set(service.Spec.Selector).AsSelectorPreValidated())
- if err != nil {
-
-
- return err
- }
-
- tolerateUnreadyEndpoints := service.Spec.PublishNotReadyAddresses
- if v, ok := service.Annotations[TolerateUnreadyEndpointsAnnotation]; ok {
- b, err := strconv.ParseBool(v)
- if err == nil {
- tolerateUnreadyEndpoints = b
- } else {
- utilruntime.HandleError(fmt.Errorf("Failed to parse annotation %v: %v", TolerateUnreadyEndpointsAnnotation, err))
- }
- }
-
-
-
- endpointsLastChangeTriggerTime := e.triggerTimeTracker.
- ComputeEndpointLastChangeTriggerTime(namespace, service, pods)
- subsets := []v1.EndpointSubset{}
- var totalReadyEps int
- var totalNotReadyEps int
- for _, pod := range pods {
- if len(pod.Status.PodIP) == 0 {
- klog.V(5).Infof("Failed to find an IP for pod %s/%s", pod.Namespace, pod.Name)
- continue
- }
- if !tolerateUnreadyEndpoints && pod.DeletionTimestamp != nil {
- klog.V(5).Infof("Pod is being deleted %s/%s", pod.Namespace, pod.Name)
- continue
- }
- ep, err := podToEndpointAddressForService(service, pod)
- if err != nil {
-
-
- klog.V(2).Infof("failed to find endpoint for service:%v with ClusterIP:%v on pod:%v with error:%v", service.Name, service.Spec.ClusterIP, pod.Name, err)
- continue
- }
- epa := *ep
- if endpointutil.ShouldSetHostname(pod, service) {
- epa.Hostname = pod.Spec.Hostname
- }
-
- if len(service.Spec.Ports) == 0 {
- if service.Spec.ClusterIP == api.ClusterIPNone {
- subsets, totalReadyEps, totalNotReadyEps = addEndpointSubset(subsets, pod, epa, nil, tolerateUnreadyEndpoints)
-
- }
- } else {
- for i := range service.Spec.Ports {
- servicePort := &service.Spec.Ports[i]
- portName := servicePort.Name
- portProto := servicePort.Protocol
- portNum, err := podutil.FindPort(pod, servicePort)
- if err != nil {
- klog.V(4).Infof("Failed to find port for service %s/%s: %v", service.Namespace, service.Name, err)
- continue
- }
- var readyEps, notReadyEps int
- epp := &v1.EndpointPort{Name: portName, Port: int32(portNum), Protocol: portProto}
- subsets, readyEps, notReadyEps = addEndpointSubset(subsets, pod, epa, epp, tolerateUnreadyEndpoints)
- totalReadyEps = totalReadyEps + readyEps
- totalNotReadyEps = totalNotReadyEps + notReadyEps
- }
- }
- }
- subsets = endpoints.RepackSubsets(subsets)
-
- currentEndpoints, err := e.endpointsLister.Endpoints(service.Namespace).Get(service.Name)
- if err != nil {
- if errors.IsNotFound(err) {
- currentEndpoints = &v1.Endpoints{
- ObjectMeta: metav1.ObjectMeta{
- Name: service.Name,
- Labels: service.Labels,
- },
- }
- } else {
- return err
- }
- }
- createEndpoints := len(currentEndpoints.ResourceVersion) == 0
- if !createEndpoints &&
- apiequality.Semantic.DeepEqual(currentEndpoints.Subsets, subsets) &&
- apiequality.Semantic.DeepEqual(currentEndpoints.Labels, service.Labels) {
- klog.V(5).Infof("endpoints are equal for %s/%s, skipping update", service.Namespace, service.Name)
- return nil
- }
- newEndpoints := currentEndpoints.DeepCopy()
- newEndpoints.Subsets = subsets
- newEndpoints.Labels = service.Labels
- if newEndpoints.Annotations == nil {
- newEndpoints.Annotations = make(map[string]string)
- }
- if !endpointsLastChangeTriggerTime.IsZero() {
- newEndpoints.Annotations[v1.EndpointsLastChangeTriggerTime] =
- endpointsLastChangeTriggerTime.Format(time.RFC3339Nano)
- } else {
- delete(newEndpoints.Annotations, v1.EndpointsLastChangeTriggerTime)
- }
- if newEndpoints.Labels == nil {
- newEndpoints.Labels = make(map[string]string)
- }
- if !helper.IsServiceIPSet(service) {
- newEndpoints.Labels = utillabels.CloneAndAddLabel(newEndpoints.Labels, v1.IsHeadlessService, "")
- } else {
- newEndpoints.Labels = utillabels.CloneAndRemoveLabel(newEndpoints.Labels, v1.IsHeadlessService)
- }
- klog.V(4).Infof("Update endpoints for %v/%v, ready: %d not ready: %d", service.Namespace, service.Name, totalReadyEps, totalNotReadyEps)
- if createEndpoints {
-
- _, err = e.client.CoreV1().Endpoints(service.Namespace).Create(context.TODO(), newEndpoints, metav1.CreateOptions{})
- } else {
-
- _, err = e.client.CoreV1().Endpoints(service.Namespace).Update(context.TODO(), newEndpoints, metav1.UpdateOptions{})
- }
- if err != nil {
- if createEndpoints && errors.IsForbidden(err) {
-
-
-
-
- klog.V(5).Infof("Forbidden from creating endpoints: %v", err)
-
- if errors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
- return nil
- }
- }
- if createEndpoints {
- e.eventRecorder.Eventf(newEndpoints, v1.EventTypeWarning, "FailedToCreateEndpoint", "Failed to create endpoint for service %v/%v: %v", service.Namespace, service.Name, err)
- } else {
- e.eventRecorder.Eventf(newEndpoints, v1.EventTypeWarning, "FailedToUpdateEndpoint", "Failed to update endpoint %v/%v: %v", service.Namespace, service.Name, err)
- }
- return err
- }
- return nil
- }
- func (e *EndpointController) checkLeftoverEndpoints() {
- list, err := e.endpointsLister.List(labels.Everything())
- if err != nil {
- utilruntime.HandleError(fmt.Errorf("Unable to list endpoints (%v); orphaned endpoints will not be cleaned up. (They're pretty harmless, but you can restart this component if you want another attempt made.)", err))
- return
- }
- for _, ep := range list {
- if _, ok := ep.Annotations[resourcelock.LeaderElectionRecordAnnotationKey]; ok {
-
-
-
-
-
- continue
- }
- key, err := controller.KeyFunc(ep)
- if err != nil {
- utilruntime.HandleError(fmt.Errorf("Unable to get key for endpoint %#v", ep))
- continue
- }
- e.queue.Add(key)
- }
- }
- func addEndpointSubset(subsets []v1.EndpointSubset, pod *v1.Pod, epa v1.EndpointAddress,
- epp *v1.EndpointPort, tolerateUnreadyEndpoints bool) ([]v1.EndpointSubset, int, int) {
- var readyEps int
- var notReadyEps int
- ports := []v1.EndpointPort{}
- if epp != nil {
- ports = append(ports, *epp)
- }
- if tolerateUnreadyEndpoints || podutil.IsPodReady(pod) {
- subsets = append(subsets, v1.EndpointSubset{
- Addresses: []v1.EndpointAddress{epa},
- Ports: ports,
- })
- readyEps++
- } else if shouldPodBeInEndpoints(pod) {
- klog.V(5).Infof("Pod is out of service: %s/%s", pod.Namespace, pod.Name)
- subsets = append(subsets, v1.EndpointSubset{
- NotReadyAddresses: []v1.EndpointAddress{epa},
- Ports: ports,
- })
- notReadyEps++
- }
- return subsets, readyEps, notReadyEps
- }
- func shouldPodBeInEndpoints(pod *v1.Pod) bool {
- switch pod.Spec.RestartPolicy {
- case v1.RestartPolicyNever:
- return pod.Status.Phase != v1.PodFailed && pod.Status.Phase != v1.PodSucceeded
- case v1.RestartPolicyOnFailure:
- return pod.Status.Phase != v1.PodSucceeded
- default:
- return true
- }
- }
|