如何分析SuperEdge 拓扑算法

# A Service that prefers node local, zonal, then regional endpoints but falls back to cluster wide endpoints.apiVersion: v1kind: Servicemetadata:    name: my-servicespec:    selector:        app: my-app    ports:        - protocol: TCP            port: 80            targetPort: 9376    topologyKeys:        - "kubernetes.io/hostname"    - "topology.kubernetes.io/zone"     - "topology.kubernetes.io/region"    - "*"

# A Service that only prefers node zone1al endpoints.apiVersion: v1kind: Servicemetadata:    annotations:        topologyKeys: '["zone1"]'    labels:        superedge.io/grid-selector: servicegrid-demo    name: servicegrid-demo-svcspec:    ports:    - port: 80        protocol: TCP        targetPort: 8080    selector:        appGrid: echo

# step1: labels edge nodes$ kubectl  get nodesNAME    STATUS   ROLES    AGE   VERSIONnode0   Ready    <none>   16d   v1.16.7node1    Ready    <none>   16d   v1.16.7node2    Ready    <none>   16d   v1.16.7# nodeunit1(nodegroup and servicegroup zone1)$ kubectl --kubeconfig config label nodes node0 zone1=nodeunit1  # nodeunit2(nodegroup and servicegroup zone1)$ kubectl --kubeconfig config label nodes node1 zone1=nodeunit2$ kubectl --kubeconfig config label nodes node2 zone1=nodeunit2...# step3: deploy echo ServiceGrid$ cat <<EOF | kubectl --kubeconfig config apply -f -apiVersion: superedge.io/v1kind: ServiceGridmetadata:    name: servicegrid-demo    namespace: defaultspec:    gridUniqKey: zone1    template:        selector:            appGrid: echo        ports:        - protocol: TCP            port: 80            targetPort: 8080EOFservicegrid.superedge.io/servicegrid-demo created# note that there is only one relevant service generated$ kubectl  get svcNAME                TYPE        CLUSTER-IP        EXTERNAL-IP   PORT(S)   AGEkubernetes          ClusterIP   192.168.0.1       <none>        443/TCP   16dservicegrid-demo-svc   ClusterIP   192.168.6.139     <none>        80/TCP    10m    # step4: access servicegrid-demo-svc(service topology and closed-looped)# execute on node0$ curl 192.168.6.139|grep "node name"        node name:      node0# execute on node1 and node2$ curl 192.168.6.139|grep "node name"        node name:      node2$ curl 192.168.6.139|grep "node name"       node name:      node1

func (sgc *ServiceGridController) syncServiceGrid(key string) error {        startTime := time.Now()        klog.V(4).Infof("Started syncing service grid %q (%v)", key, startTime)        defer func() {              klog.V(4).Infof("Finished syncing service grid %q (%v)", key, time.Since(startTime))        }()            namespace, name, err := cache.SplitMetaNamespaceKey(key)        if err != nil {              return err        }            sg, err := sgc.svcGridLister.ServiceGrids(namespace).Get(name)        if errors.IsNotFound(err) {                      klog.V(2).Infof("service grid %v has been deleted", key)                return nil        }        if err != nil {                return err        }            if sg.Spec.GridUniqKey == "" {               sgc.eventRecorder.Eventf(sg, corev1.EventTypeWarning, "Empty", "This service grid has an empty grid key")                return nil        }            // get service workload list of this grid        svcList, err := sgc.getServiceForGrid(sg)        if err != nil {                return err        }            if sg.DeletionTimestamp != nil {                return nil        }            // sync service grid relevant services workload        return sgc.reconcile(sg, svcList)    }    func (sgc *ServiceGridController) getServiceForGrid(sg *crdv1.ServiceGrid) ([]*corev1.Service, error) {      svcList, err := sgc.svcLister.Services(sg.Namespace).List(labels.Everything())      if err != nil {            return nil, err      }        labelSelector, err := common.GetDefaultSelector(sg.Name)      if err != nil {              return nil, err      }          canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error)  {                 fresh, err :=   sgc.crdClient.SuperedgeV1().ServiceGrids(sg.Namespace).Get(context.TODO(), sg.Name, metav1.GetOptions{})              if err != nil {                    return nil, err              }              if fresh.UID != sg.UID {                      return nil, fmt.Errorf("orignal service grid %v/%v is gone: got uid %v, wanted %v", sg.Namespace,                                sg.Name, fresh.UID, sg.UID)              }              return fresh, nil        })              cm := controller.NewServiceControllerRefManager(sgc.svcClient, sg, labelSelector, util.ControllerKind, canAdoptFunc)        return cm.ClaimService(svcList)}    func (sgc *ServiceGridController) reconcile(g *crdv1.ServiceGrid, svcList []*corev1.Service) error {        var (                adds    []*corev1.Service             updates []*corev1.Service             deletes []*corev1.Service        )            sgTargetSvcName := util.GetServiceName(g)        isExistingSvc := false        for _, svc := range svcList {             if svc.Name == sgTargetSvcName {                        isExistingSvc = true                 template := util.KeepConsistence(g, svc)                        if !apiequality.Semantic.DeepEqual(template, svc) {                           updates = append(updates, template)                        }                } else {                        deletes = append(deletes, svc)                }        }            if !isExistingSvc {                adds = append(adds, util.CreateService(g))        }            return sgc.syncService(adds, updates, deletes)}func CreateService(sg *crdv1.ServiceGrid) *corev1.Service {        svc := &corev1.Service{                ObjectMeta: metav1.ObjectMeta{                      Name:      GetServiceName(sg),                      Namespace: sg.Namespace,             // Append existed ServiceGrid labels to service to be created            Labels: func() map[string]string {                              if sg.Labels != nil {                   newLabels := sg.Labels                                      newLabels[common.GridSelectorName] = sg.Name                               newLabels[common.GridSelectorUniqKeyName] = sg.Spec.GridUniqKey                                      return newLabels                           } else {                                 return map[string]string{                                              common.GridSelectorName:        sg.Name,                                   common.GridSelectorUniqKeyName: sg.Spec.GridUniqKey,               }                         }                  }(),                  Annotations: make(map[string]string),           },            Spec: sg.Spec.Template,     }        keys := make([]string, 1)      keys[0] = sg.Spec.GridUniqKey      keyData, _ := json.Marshal(keys)       svc.Annotations[common.TopologyAnnotationsKey] = string(keyData)        return svc}

apiVersion: v1kind: Servicemetadata:    annotations:        topologyKeys: '["zone1"]'    creationTimestamp: "2021-03-03T07:33:30Z"    labels:        superedge.io/grid-selector: servicegrid-demo      name: servicegrid-demo-svc      namespace: default      ownerReferences:      - apiVersion: superedge.io/v1          blockOwnerDeletion: true          controller: true          kind: ServiceGrid          name: servicegrid-demo          uid: 78c74d3c-72ac-4e68-8c79-f1396af5a581      resourceVersion: "127987090"      selfLink: /api/v1/namespaces/default/services/servicegrid-demo-svc      uid: 8130ba7b-c27e-4c3a-8ceb-4f6dd0178dfcspec:      clusterIP: 192.168.161.1      ports:      - port: 80        protocol: TCP        targetPort: 8080    selector:        appGrid: echo    sessionAffinity: None    type: ClusterIPstatus:    loadBalancer: {}

kube-proxy -> application-grid-wrapper -> lite-apiserver -> kube-apiserver

func (s *interceptorServer) Run(debug bool, bindAddress string, insecure bool, caFile, certFile, keyFile string) error {        ...        klog.Infof("Start to run interceptor server")        /* filter         */        server := &http.Server{Addr: bindAddress, Handler: s.buildFilterChains(debug)}            if insecure {                return server.ListenAndServe()        }        ...        server.TLSConfig = tlsConfig        return server.ListenAndServeTLS("", "")}func (s *interceptorServer) buildFilterChains(debug bool) http.Handler {       handler := http.Handler(http.NewServeMux())            handler = s.interceptEndpointsRequest(handler)        handler = s.interceptServiceRequest(handler)        handler = s.interceptEventRequest(handler)        handler = s.interceptNodeRequest(handler)        handler = s.logger(handler)          if debug {              handler = s.debugger(handler)     }        return handler}

func (s *interceptorServer) interceptEventRequest(handler http.Handler) http.Handler {     return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {           if r.Method != http.MethodPost || !strings.HasSuffix(r.URL.Path, "/events") {                   handler.ServeHTTP(w, r)              return           }                targetURL, _ := url.Parse(s.restConfig.Host)           reverseProxy := httputil.NewSingleHostReverseProxy(targetURL)           reverseProxy.Transport, _ = rest.TransportFor(s.restConfig)           reverseProxy.ServeHTTP(w, r)     })  }

type storageCache struct {        // hostName is the nodeName of node which application-grid-wrapper deploys on        hostName         string        wrapperInCluster bool            // mu lock protect the following map structure        mu           sync.RWMutex        servicesMap  map[types.NamespacedName]*serviceContainer        endpointsMap map[types.NamespacedName]*endpointsContainer        nodesMap     map[types.NamespacedName]*nodeContainer            // service watch channel        serviceChan chan<- watch.Event       // endpoints watch channel        endpointsChan chan<- watch.Event}...func NewStorageCache(hostName string, wrapperInCluster bool, serviceNotifier, endpointsNotifier chan watch.Event) *storageCache {       msc := &storageCache{               hostName:         hostName,          wrapperInCluster: wrapperInCluster,               servicesMap:      make(map[types.NamespacedName]*serviceContainer),        endpointsMap:     make(map[types.NamespacedName]*endpointsContainer),               nodesMap:         make(map[types.NamespacedName]*nodeContainer),         serviceChan:      serviceNotifier,               endpointsChan:    endpointsNotifier,        }            return msc}...func (s *interceptorServer) Run(debug bool, bindAddress string, insecure bool, caFile, certFile, keyFile string) error {        ...        if err := s.setupInformers(ctx.Done()); err != nil {                return err        }         klog.Infof("Start to run interceptor server")       /* filter        */      server := &http.Server{Addr: bindAddress, Handler: s.buildFilterChains(debug)}      ...      return server.ListenAndServeTLS("", "")}func (s *interceptorServer) setupInformers(stop <-chan struct{}) error {       klog.Infof("Start to run service and endpoints informers")        noProxyName, err := labels.NewRequirement(apis.LabelServiceProxyName, selection.DoesNotExist, nil)        if err != nil {                klog.Errorf("can't parse proxy label, %v", err)                return err        }            noHeadlessEndpoints, err := labels.NewRequirement(v1.IsHeadlessService, selection.DoesNotExist, nil)        if err != nil {                klog.Errorf("can't parse headless label, %v", err)                return err        }                labelSelector := labels.NewSelector()        labelSelector = labelSelector.Add(*noProxyName, *noHeadlessEndpoints)         resyncPeriod := time.Minute * 5       client := kubernetes.NewForConfigOrDie(s.restConfig)        nodeInformerFactory := informers.NewSharedInformerFactory(client, resyncPeriod)        informerFactory := informers.NewSharedInformerFactoryWithOptions(client, resyncPeriod,              informers.WithTweakListOptions(func(options *metav1.ListOptions) {                    options.LabelSelector = labelSelector.String()               }))                 nodeInformer := nodeInformerFactory.Core().V1().Nodes().Informer()        serviceInformer := informerFactory.Core().V1().Services().Informer()        endpointsInformer := informerFactory.Core().V1().Endpoints().Informer()           /*        */        nodeInformer.AddEventHandlerWithResyncPeriod(s.cache.NodeEventHandler(), resyncPeriod)        serviceInformer.AddEventHandlerWithResyncPeriod(s.cache.ServiceEventHandler(), resyncPeriod)           endpointsInformer.AddEventHandlerWithResyncPeriod(s.cache.EndpointsEventHandler(), resyncPeriod)            go nodeInformer.Run(stop)        go serviceInformer.Run(stop)        go endpointsInformer.Run(stop)            if !cache.WaitForNamedCacheSync("node", stop,                nodeInformer.HasSynced,                serviceInformer.HasSynced,             endpointsInformer.HasSynced) {              return fmt.Errorf("can't sync informers")        }            return nil}func (sc *storageCache) NodeEventHandler() cache.ResourceEventHandler {        return &nodeHandler{cache: sc}}func (sc *storageCache) ServiceEventHandler() cache.ResourceEventHandler {        return &serviceHandler{cache: sc}}func (sc *storageCache) EndpointsEventHandler() cache.ResourceEventHandler {        return &endpointsHandler{cache: sc}}

func (nh *nodeHandler) add(node *v1.Node) {        sc := nh.cache            sc.mu.Lock()            nodeKey := types.NamespacedName{Namespace: node.Namespace, Name: node.Name}        klog.Infof("Adding node %v", nodeKey)        sc.nodesMap[nodeKey] = &nodeContainer{                node:   node,                labels: node.Labels,        }        // update endpoints        changedEps := sc.rebuildEndpointsMap()            sc.mu.Unlock()            for _, eps := range changedEps {         sc.endpointsChan <- eps            }}func (nh *nodeHandler) update(node *v1.Node) {        sc := nh.cache            sc.mu.Lock()            nodeKey := types.NamespacedName{Namespace: node.Namespace, Name: node.Name}        klog.Infof("Updating node %v", nodeKey)        nodeContainer, found := sc.nodesMap[nodeKey]        if !found {                sc.mu.Unlock()                klog.Errorf("Updating non-existed node %v", nodeKey)                return        }            nodeContainer.node = node        // return directly when labels of node stay unchanged        if reflect.DeepEqual(node.Labels, nodeContainer.labels) {                sc.mu.Unlock()                return        }        nodeContainer.labels = node.Labels            // update endpoints        changedEps := sc.rebuildEndpointsMap()            sc.mu.Unlock()            for _, eps := range changedEps {                sc.endpointsChan <- eps      }}...

// rebuildEndpointsMap updates all endpoints stored in storageCache.endpointsMap dynamically and constructs relevant modified eventsfunc (sc *storageCache) rebuildEndpointsMap() []watch.Event {       evts := make([]watch.Event, 0)       for name, endpointsContainer := range sc.endpointsMap {               newEps := pruneEndpoints(sc.hostName, sc.nodesMap, sc.servicesMap, endpointsContainer.endpoints, sc.wrapperInCluster)           if apiequality.Semantic.DeepEqual(newEps, endpointsContainer.modified) {                        continue                }                sc.endpointsMap[name].modified = newEps                evts = append(evts, watch.Event{                        Type:   watch.Modified,                  Object: newEps,            })       }       return evts}

// pruneEndpoints filters endpoints using serviceTopology rules combined by services topologyKeys and node labelsfunc pruneEndpoints(hostName string,    nodes map[types.NamespacedName]*nodeContainer,        services map[types.NamespacedName]*serviceContainer,        eps *v1.Endpoints, wrapperInCluster bool) *v1.Endpoints {           epsKey := types.NamespacedName{Namespace: eps.Namespace, Name: eps.Name}          if wrapperInCluster {               eps = genLocalEndpoints(eps)     }           // dangling endpoints        svc, ok := services[epsKey]     if !ok {                klog.V(4).Infof("Dangling endpoints %s, %+#v", eps.Name, eps.Subsets)                return eps        }               // normal service        if len(svc.keys) == 0 {             klog.V(4).Infof("Normal endpoints %s, %+#v", eps.Name, eps.Subsets)            return eps       }            // topology endpoints        newEps := eps.DeepCopy()        for si := range newEps.Subsets {         subnet := &newEps.Subsets[si]        subnet.Addresses = filterConcernedAddresses(svc.keys, hostName, nodes, subnet.Addresses)                subnet.NotReadyAddresses = filterConcernedAddresses(svc.keys, hostName, nodes, subnet.NotReadyAddresses)        }        klog.V(4).Infof("Topology endpoints %s: subnets from %+#v to %+#v", eps.Name, eps.Subsets, newEps.Subsets)         return newEps}// filterConcernedAddresses aims to filter out endpoints addresses within the same node unitfunc filterConcernedAddresses(topologyKeys []string, hostName string, nodes map[types.NamespacedName]*nodeContainer,        addresses []v1.EndpointAddress) []v1.EndpointAddress {        hostNode, found := nodes[types.NamespacedName{Name: hostName}]        if !found {                return nil        }            filteredEndpointAddresses := make([]v1.EndpointAddress, 0)        for i := range addresses {               addr := addresses[i]                if nodeName := addr.NodeName; nodeName != nil {                         epsNode, found := nodes[types.NamespacedName{Name: *nodeName}]                        if !found {                              continue                         }                         if hasIntersectionLabel(topologyKeys, hostNode.labels, epsNode.labels) {                                 filteredEndpointAddresses = append(filteredEndpointAddresses, addr)                         }                }        }            return filteredEndpointAddresses }func hasIntersectionLabel(keys []string, n1, n2 map[string]string) bool {     if n1 == nil || n2 == nil {              return false        }            for _, key := range keys {               val1, v1found := n1[key]             val2, v2found := n2[key]                    if v1found && v2found && val1 == val2 {                    return true                }        }            return false   }

apiVersion: v1kind: Endpointsmetadata:    annotations:        superedge.io/local-endpoint: 127.0.0.1        superedge.io/local-port: "51003"   name: kubernetes    namespace: defaultsubsets:- addresses:   - ip: 172.31.0.60    ports:    - name: https      port: xxx      protocol: TCP

func genLocalEndpoints(eps *v1.Endpoints) *v1.Endpoints {        if eps.Namespace != metav1.NamespaceDefault || eps.Name != MasterEndpointName {                return eps        }            klog.V(4).Infof("begin to gen local ep %v", eps)        ipAddress, e := eps.Annotations[EdgeLocalEndpoint]        if !e {                return eps        }                portStr, e := eps.Annotations[EdgeLocalPort]        if !e {                return eps        }            klog.V(4).Infof("get local endpoint %s:%s", ipAddress, portStr)        port, err := strconv.ParseInt(portStr, 10, 32)        if err != nil {                klog.Errorf("parse int %s err %v", portStr, err)                return eps        }          ip := net.ParseIP(ipAddress)       if ip == nil {                klog.Warningf("parse ip %s nil", ipAddress)                return eps        }            nep := eps.DeepCopy()        nep.Subsets = []v1.EndpointSubset{               {                       Addresses: []v1.EndpointAddress{                              {                                       IP: ipAddress,                   },                        },                        Ports: []v1.EndpointPort{                                 {                                         Protocol: v1.ProtocolTCP,                                         Port:     int32(port),                                        Name:     "https",                                },                      },               },       }            klog.V(4).Infof("gen new endpoint complete %v", nep)          return nep}

func getTopologyKeys(objectMeta *metav1.ObjectMeta) []string {       if !hasTopologyKey(objectMeta) {         return nil        }           var keys []string       keyData := objectMeta.Annotations[TopologyAnnotationsKey]        if err := json.Unmarshal([]byte(keyData), &keys); err != nil {                klog.Errorf("can't parse topology keys %s, %v", keyData, err)               return nil        }            return keys}

// filterConcernedAddresses aims to filter out endpoints addresses within the same node unitfunc filterConcernedAddresses(topologyKeys []string, hostName string, nodes map[types.NamespacedName]*nodeContainer,        addresses []v1.EndpointAddress) []v1.EndpointAddress {        hostNode, found := nodes[types.NamespacedName{Name: hostName}]        if !found {                return nil       }            filteredEndpointAddresses := make([]v1.EndpointAddress, 0)        for i := range addresses {               addr := addresses[i]                if nodeName := addr.NodeName; nodeName != nil {                        epsNode, found := nodes[types.NamespacedName{Name: *nodeName}]               if !found {                                continue                        }                       if hasIntersectionLabel(topologyKeys, hostNode.labels, epsNode.labels) {                                filteredEndpointAddresses = append(filteredEndpointAddresses, addr)                        }               }       }            return filteredEndpointAddresses}    func hasIntersectionLabel(keys []string, n1, n2 map[string]string) bool {           if n1 == nil || n2 == nil {                   return false            }              for _, key := range keys {                  val1, v1found := n1[key]                  val2, v2found := n2[key]                 if v1found && v2found && val1 == val2 {                     return true               }       }         return false}

func (nh *nodeHandler) add(node *v1.Node) {        sc := nh.cache           sc.mu.Lock()            nodeKey := types.NamespacedName{Namespace: node.Namespace, Name: node.Name}        klog.Infof("Adding node %v", nodeKey)        sc.nodesMap[nodeKey] = &nodeContainer{                node:   node,                labels: node.Labels,        }        // update endpoints        changedEps := sc.rebuildEndpointsMap()            sc.mu.Unlock()            for _, eps := range changedEps {         sc.endpointsChan <- eps        }}// rebuildEndpointsMap updates all endpoints stored in storageCache.endpointsMap dynamically and constructs relevant modified eventsfunc (sc *storageCache) rebuildEndpointsMap() []watch.Event {        evts := make([]watch.Event, 0)      for name, endpointsContainer := range sc.endpointsMap {                newEps := pruneEndpoints(sc.hostName, sc.nodesMap, sc.servicesMap, endpointsContainer.endpoints, sc.wrapperInCluster)                if apiequality.Semantic.DeepEqual(newEps, endpointsContainer.modified) {                        continue                }                sc.endpointsMap[name].modified = newEps                evts = append(evts, watch.Event{                        Type:   watch.Modified,               Object: newEps,                })        }        return evts}

func (sh *serviceHandler) update(service *v1.Service) {        sc := sh.cache           sc.mu.Lock()        serviceKey := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}        klog.Infof("Updating service %v", serviceKey)        newTopologyKeys := getTopologyKeys(&service.ObjectMeta)        serviceContainer, found := sc.servicesMap[serviceKey]        if !found {                sc.mu.Unlock()                klog.Errorf("update non-existed service, %v", serviceKey)                return        }          sc.serviceChan <- watch.Event{           Type:   watch.Modified,               Object: service,        }            serviceContainer.svc = service       // return directly when topologyKeys of service stay unchanged        if reflect.DeepEqual(serviceContainer.keys, newTopologyKeys) {                sc.mu.Unlock()                return        }            serviceContainer.keys = newTopologyKeys            // update endpoints        changedEps := sc.rebuildEndpointsMap()        sc.mu.Unlock()            for _, eps := range changedEps {         sc.endpointsChan <- eps        }}

func (eh *endpointsHandler) update(endpoints *v1.Endpoints) {        sc := eh.cache            sc.mu.Lock()       endpointsKey := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}        klog.Infof("Updating endpoints %v", endpointsKey)            endpointsContainer, found := sc.endpointsMap[endpointsKey]        if !found {                sc.mu.Unlock()               klog.Errorf("Updating non-existed endpoints %v", endpointsKey)            return        }        endpointsContainer.endpoints = endpoints        newEps := pruneEndpoints(sc.hostName, sc.nodesMap, sc.servicesMap, endpoints, sc.wrapperInCluster)        changed := !apiequality.Semantic.DeepEqual(endpointsContainer.modified, newEps)        if changed {                endpointsContainer.modified = newEps       }        sc.mu.Unlock()            if changed {               sc.endpointsChan <- watch.Event{                        Type:   watch.Modified,               Object: newEps,               }       }}

func (s *interceptorServer) interceptEndpointsRequest(handler http.Handler) http.Handler {        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {         if r.Method != http.MethodGet || !strings.HasPrefix(r.URL.Path, "/api/v1/endpoints") {                        handler.ServeHTTP(w, r)               return               }                        queries := r.URL.Query()             acceptType := r.Header.Get("Accept")                info, found := s.parseAccept(acceptType, s.mediaSerializer)               if !found {                        klog.Errorf("can't find %s serializer", acceptType)                       w.WriteHeader(http.StatusBadRequest)                        return                }                        encoder := scheme.Codecs.EncoderForVersion(info.Serializer, v1.SchemeGroupVersion)           // list request                if queries.Get("watch") == "" {                       w.Header().Set("Content-Type", info.MediaType)                        allEndpoints := s.cache.GetEndpoints()                       epsItems := make([]v1.Endpoints, 0, len(allEndpoints))                    for _, eps := range allEndpoints {                                epsItems = append(epsItems, *eps)                        }                                    epsList := &v1.EndpointsList{                                Items: epsItems,                  }                                  err := encoder.Encode(epsList, w)                      if err != nil {                           klog.Errorf("can't marshal endpoints list, %v", err)                      w.WriteHeader(http.StatusInternalServerError)                             return                      }                              return             }                      // watch request             timeoutSecondsStr := r.URL.Query().Get("timeoutSeconds")         timeout := time.Minute              if timeoutSecondsStr != "" {            timeout, _ = time.ParseDuration(fmt.Sprintf("%ss", timeoutSecondsStr))             }                      timer := time.NewTimer(timeout)            defer timer.Stop()                   flusher, ok := w.(http.Flusher)            if !ok {                      klog.Errorf("unable to start watch - can't get http.Flusher: %#v", w)                      w.WriteHeader(http.StatusMethodNotAllowed)                      return                }                    e := restclientwatch.NewEncoder(                       streaming.NewEncoder(info.StreamSerializer.Framer.NewFrameWriter(w),                        scheme.Codecs.EncoderForVersion(info.StreamSerializer, v1.SchemeGroupVersion)),                      encoder)              if info.MediaType == runtime.ContentTypeProtobuf {                   w.Header().Set("Content-Type",     runtime.ContentTypeProtobuf+";stream=watch")             } else {                     w.Header().Set("Content-Type", runtime.ContentTypeJSON)              }              w.Header().Set("Transfer-Encoding", "chunked")              w.WriteHeader(http.StatusOK)         flusher.Flush()              for {                      select {                      case <-r.Context().Done():               return                      case <-timer.C:                          return                     case evt := <-s.endpointsWatchCh:                             klog.V(4).Infof("Send endpoint watch event: %+#v", evt)                   err := e.Encode(&evt)                 if err != nil {                       klog.Errorf("can't encode watch event, %v", err)                           return                          }                                   if len(s.endpointsWatchCh) == 0 {                                  flusher.Flush()                     }                     }               }      })}

func (sc *storageCache) GetEndpoints() []*v1.Endpoints {        sc.mu.RLock()       defer sc.mu.RUnlock()           epList := make([]*v1.Endpoints, 0, len(sc.endpointsMap))        for _, v := range sc.endpointsMap {                epList = append(epList, v.modified)        }        return epList}