default via 169.254.1.1 dev eth0169.254.1.1 dev eth0 scope link

169.254.1.1 dev eth0 lladdr ee:ee:ee:ee:ee:ee STALE

172.17.8.2 dev veth1 scope link172.17.8.3 dev veth2 scope link172.17.9.0/24 via 192.168.100.101 dev eth0 proto bird onlink172.17.10.0/24 via 192.168.100.102 dev eth0 proto bird onlink

172.17.9.2 dev veth1 scope link172.17.9.3 dev veth2 scope link172.17.8.0/24 via 192.168.100.100 dev eth0 proto bird onlink172.17.10.0/24 via 192.168.100.102 dev eth0 proto bird onlink

curl --header "Content-Type: application/json-patch+json" \--request PATCH \--data '[{"op": "add", "path": "/status/capacity/example.com~1dongle", "value": "4"}]' \http://localhost:8001/api/v1/nodes/<your-node-name>/status

apiVersion: v1kind: Podmetadata:  name: extended-resource-demospec:  containers:  - name: extended-resource-demo-ctr    image: nginx    resources:      requests:        example.com/dongle: 3      limits:        example.com/dongle: 3

tc qdisc add dev eth0 root handle 1: htb default 12

tc class add dev eth0 parent 1: classid 1:1 htb rate 100kbps ceil 100kbps

tc class add dev eth0 parent 1:1 classid 1:10 htb rate 30kbps ceil 100kbpstc class add dev eth0 parent 1:1 classid 1:11 htb rate 10kbps ceil 100kbpstc class add dev eth0 parent 1:1 classid 1:12 htb rate 60kbps ceil 100kbps

tc qdisc add dev eth0 parent 1:10 handle 20: pfifo limit 5tc qdisc add dev eth0 parent 1:11 handle 30: pfifo limit 5tc qdisc add dev eth0 parent 1:12 handle 40: sfq perturb 10

tc filter add dev eth0 protocol ip parent 1:0 prio 1 u32 match ip src 1.2.3.4 match ip dport 80 0xffff flowid 1:10tc filter add dev eth0 protocol ip parent 1:0 prio 1 u32 match ip src 1.2.3.4 flowid 1:11

tc filter add dev eth0 protocol ip parent 1:0 prio 1 handle 1: cgroup

mkdir /sys/fs/cgroup/net_cls/amkdir /sys/fs/cgroup/net_cls/b

echo 0x00010010 > /sys/fs/cgroup/net_cls/a/net_cls.classidecho 0x00010011 > /sys/fs/cgroup/net_cls/b/net_cls.classid

// grpcServices are all the grpc services provided by cri containerd.type grpcServices interface {  runtime.RuntimeServiceServer  runtime.ImageServiceServer}
// CRIService is the interface implement CRI remote service server.type CRIService interface {  Run() error  // io.Closer is used by containerd to gracefully stop cri service.  io.Closer  plugin.Service  grpcServices}

// criService implements CRIService.type criService struct {  // config contains all configurations.  config criconfig.Config  // imageFSPath is the path to image filesystem.  imageFSPath string  // os is an interface for all required os operations.  os osinterface.OS  // sandboxStore stores all resources associated with sandboxes.  sandboxStore *sandboxstore.Store  // sandboxNameIndex stores all sandbox names and make sure each name  // is unique.  sandboxNameIndex *registrar.Registrar  // containerStore stores all resources associated with containers.  containerStore *containerstore.Store  // containerNameIndex stores all container names and make sure each  // name is unique.  containerNameIndex *registrar.Registrar  // imageStore stores all resources associated with images.  imageStore *imagestore.Store  // snapshotStore stores information of all snapshots.  snapshotStore *snapshotstore.Store  // netPlugin is used to setup and teardown network when run/stop pod sandbox.  netPlugin cni.CNI  // client is an instance of the containerd client  client *containerd.Client......}

restStorage := LegacyRESTStorage{}......podTemplateStorage, err := podtemplatestore.NewREST(restOptionsGetter)......nodeStorage, err := nodestore.NewStorage(restOptionsGetter, c.KubeletClientConfig, c.ProxyTransport)......podStorage, err := podstore.NewStorage(    restOptionsGetter,    nodeStorage.KubeletConnectionInfo,    c.ProxyTransport,    podDisruptionClient,  )......restStorageMap := map[string]rest.Storage{    "pods":             podStorage.Pod,......    "podTemplates": podTemplateStorage,......    "services":        serviceRest,......    "endpoints": endpointsStorage,......    "nodes":        nodeStorage.Node,......}

--etcd-servers-overrides=/events#https://0.example.com:2381;https://1.example.com:2381;https://2.example.com:2381

func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {  controllers := map[string]InitFunc{}  controllers["endpoint"] = startEndpointController  controllers["replicationcontroller"] = startReplicationController......  controllers["daemonset"] = startDaemonSetController......  controllers["deployment"] = startDeploymentController  controllers["replicaset"] = startReplicaSetController  controllers["horizontalpodautoscaling"] = startHPAController......  controllers["statefulset"] = startStatefulSetController......  controllers["service"] = startServiceController......  return controllers}

go replicaset.NewReplicaSetController(    ctx.InformerFactory.Apps().V1().ReplicaSets(),    ctx.InformerFactory.Core().V1().Pods(),    ctx.ClientBuilder.ClientOrDie("replicaset-controller"),    replicaset.BurstReplicas,  ).Run(int(ctx.ComponentConfig.ReplicaSetController.ConcurrentRSSyncs), ctx.Stop)

type PodInformer interface {  Informer() cache.SharedIndexInformer  Lister() v1.PodLister}
func (f *podInformer) Informer() cache.SharedIndexInformer {  return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)}
func (f *podInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {  return NewFilteredPodInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)}
func NewFilteredPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {  return cache.NewSharedIndexInformer(    &cache.ListWatch{      ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {        return client.CoreV1().Pods(namespace).List(context.TODO(), options)      },      WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {        return client.CoreV1().Pods(namespace).Watch(context.TODO(), options)      },    },    &corev1.Pod{},    resyncPeriod,    indexers,  )}
func NewSharedIndexInformer(lw ListerWatcher, exampleObject runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer {  realClock := &clock.RealClock{}  sharedIndexInformer := &sharedIndexInformer{    processor:                       &sharedProcessor{clock: realClock},    indexer:                         NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers),    listerWatcher:                   lw,    objectType:                      exampleObject,    resyncCheckPeriod:               defaultEventHandlerResyncPeriod,    defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,    cacheMutationDetector:           NewCacheMutationDetector(fmt.Sprintf("%T", exampleObject)),    clock:                           realClock,  }  return sharedIndexInformer}

rsc := &ReplicaSetController{    GroupVersionKind: gvk,    kubeClient:       kubeClient,    podControl:       podControl,    burstReplicas:    burstReplicas,    expectations:     controller.NewUIDTrackingControllerExpectations(controller.NewControllerExpectations()),    queue:            workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), queueName),}
rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{    AddFunc:    rsc.addRS,    UpdateFunc: rsc.updateRS,    DeleteFunc: rsc.deleteRS,})
podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{    AddFunc: rsc.addPod,    // This invokes the ReplicaSet for every pod change, eg: host assignment. Though this might seem like    // overkill the most frequent pod update is status, and the associated ReplicaSet will only list from    // local storage, so it should be ok.    UpdateFunc: rsc.updatePod,    DeleteFunc: rsc.deletePod,})
rsc.syncHandler = rsc.syncReplicaSet

// When a pod is deleted, enqueue the replica set that manages the pod and update its expectations.// obj could be an *v1.Pod, or a DeletionFinalStateUnknown marker item.func (rsc *ReplicaSetController) deletePod(obj interface{}) {  pod, ok := obj.(*v1.Pod)......  controllerRef := metav1.GetControllerOf(pod)......  rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)......  rsKey, err := controller.KeyFunc(rs)......
  rsc.expectations.DeletionObserved(rsKey, controller.PodKey(pod))  rsc.queue.Add(rsKey)}

// Run begins watching and syncing.func (rsc *ReplicaSetController) Run(workers int, stopCh <-chan struct{}) {......  for i := ; i < workers; i++ {    go wait.Until(rsc.worker, time.Second, stopCh)  }......}
func (rsc *ReplicaSetController) worker() {  for rsc.processNextWorkItem() {  }}
func (rsc *ReplicaSetController) processNextWorkItem() bool {  key, quit := rsc.queue.Get()......  err := rsc.syncHandler(key.(string))......  return true}

func (rsc *ReplicaSetController) syncReplicaSet(key string) error {  namespace, name, err := cache.SplitMetaNamespaceKey(key)  rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)  allPods, err := rsc.podLister.Pods(rs.Namespace).List(labels.Everything())  filteredPods := controller.FilterActivePods(allPods)  filteredPods, err = rsc.claimPods(rs, selector, filteredPods)  manageReplicasErr = rsc.manageReplicas(filteredPods, rs)  newStatus := calculateStatus(rs, filteredPods, manageReplicasErr)  updatedRS, err := updateReplicaSetStatus(rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace), rs, newStatus)}

sched, err := scheduler.New(cc.Client,    cc.InformerFactory,    cc.PodInformer,    recorderFactory,    ctx.Done(),    scheduler.WithProfiles(cc.ComponentConfig.Profiles...),    scheduler.WithAlgorithmSource(cc.ComponentConfig.AlgorithmSource),    scheduler.WithPreemptionDisabled(cc.ComponentConfig.DisablePreemption),    scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),    scheduler.WithBindTimeoutSeconds(cc.ComponentConfig.BindTimeoutSeconds),    scheduler.WithFrameworkOutOfTreeRegistry(outOfTreeRegistry),    scheduler.WithPodMaxBackoffSeconds(cc.ComponentConfig.PodMaxBackoffSeconds),    scheduler.WithPodInitialBackoffSeconds(cc.ComponentConfig.PodInitialBackoffSeconds),    scheduler.WithExtenders(cc.ComponentConfig.Extenders...),  )

schedulerCache := internalcache.New(30*time.Second, stopEverything)

  volumeBinder := scheduling.NewVolumeBinder(    client,    informerFactory.Core().V1().Nodes(),    informerFactory.Storage().V1().CSINodes(),    informerFactory.Core().V1().PersistentVolumeClaims(),    informerFactory.Core().V1().PersistentVolumes(),    informerFactory.Storage().V1().StorageClasses(),    time.Duration(options.bindTimeoutSeconds)*time.Second,  )

  podQueue := internalqueue.NewSchedulingQueue(    lessFn,    internalqueue.WithPodInitialBackoffDuration(time.Duration(c.podInitialBackoffSeconds)*time.Second),    internalqueue.WithPodMaxBackoffDuration(time.Duration(c.podMaxBackoffSeconds)*time.Second),  )

  algo := core.NewGenericScheduler(    c.schedulerCache,    podQueue,    c.nodeInfoSnapshot,    extenders,    c.informerFactory.Core().V1().PersistentVolumeClaims().Lister(),    GetPodDisruptionBudgetLister(c.informerFactory),    c.disablePreemption,    c.percentageOfNodesToScore,    c.enableNonPreempting,  )

return &Scheduler{    SchedulerCache:  c.schedulerCache,    Algorithm:       algo,    Profiles:        profiles,    NextPod:         internalqueue.MakeNextPodFunc(podQueue),    Error:           MakeDefaultErrorFunc(c.client, podQueue, c.schedulerCache),    StopEverything:  c.StopEverything,    VolumeBinder:    c.volumeBinder,    SchedulingQueue: podQueue,  }

func addAllEventHandlers(  sched *Scheduler,  informerFactory informers.SharedInformerFactory,  podInformer coreinformers.PodInformer,) {  // scheduled pod cache  podInformer.Informer().AddEventHandler(    cache.FilteringResourceEventHandler{      FilterFunc: func(obj interface{}) bool {        switch t := obj.(type) {        case *v1.Pod:          return assignedPod(t)        }      },      Handler: cache.ResourceEventHandlerFuncs{        AddFunc:    sched.addPodToCache,        UpdateFunc: sched.updatePodInCache,        DeleteFunc: sched.deletePodFromCache,      },    },  )  // unscheduled pod queue  podInformer.Informer().AddEventHandler(    cache.FilteringResourceEventHandler{      FilterFunc: func(obj interface{}) bool {        switch t := obj.(type) {        case *v1.Pod:          return !assignedPod(t) && responsibleForPod(t, sched.Profiles)        }      },      Handler: cache.ResourceEventHandlerFuncs{        AddFunc:    sched.addPodToSchedulingQueue,        UpdateFunc: sched.updatePodInSchedulingQueue,        DeleteFunc: sched.deletePodFromSchedulingQueue,      },    },  )
  informerFactory.Core().V1().Nodes().Informer().AddEventHandler(    cache.ResourceEventHandlerFuncs{      AddFunc:    sched.addNodeToCache,      UpdateFunc: sched.updateNodeInCache,      DeleteFunc: sched.deleteNodeFromCache,    },  )
  // On add and delete of PVs, it will affect equivalence cache items  // related to persistent volume  informerFactory.Core().V1().PersistentVolumes().Informer().AddEventHandler(    cache.ResourceEventHandlerFuncs{      // MaxPDVolumeCountPredicate: since it relies on the counts of PV.      AddFunc:    sched.onPvAdd,      UpdateFunc: sched.onPvUpdate,    },  )
  // This is for MaxPDVolumeCountPredicate: add/delete PVC will affect counts of PV when it is bound.  informerFactory.Core().V1().PersistentVolumeClaims().Informer().AddEventHandler(    cache.ResourceEventHandlerFuncs{      AddFunc:    sched.onPvcAdd,      UpdateFunc: sched.onPvcUpdate,    },  )......}

func (sched *Scheduler) Run(ctx context.Context) {  if !cache.WaitForCacheSync(ctx.Done(), sched.scheduledPodsHasSynced) {    return  }  sched.SchedulingQueue.Run()  wait.UntilWithContext(ctx, sched.scheduleOne, )  sched.SchedulingQueue.Close()}

podInfo := sched.NextPod()

scheduleResult, err := sched.Algorithm.Schedule(schedulingCycleCtx, prof, state, pod)

allBound, err := sched.VolumeBinder.AssumePodVolumes(assumedPod, scheduleResult.SuggestedHost)

// assume modifies `assumedPod` by setting NodeName=scheduleResult.SuggestedHost  err = sched.assume(assumedPod, scheduleResult.SuggestedHost)

// bind the pod to its host asynchronously (we can do this b/c of the assumption step above).  go func() {  ......    err := sched.bindVolumes(assumedPod)  ......    err := sched.bind(bindingCycleCtx, prof, assumedPod, scheduleResult.SuggestedHost, state)  }

// Schedule tries to schedule the given pod to one of the nodes in the node list.// If it succeeds, it will return the name of the node.// If it fails, it will return a FitError error with reasons.func (g *genericScheduler) Schedule(ctx context.Context, prof *profile.Profile, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {    ......    filteredNodes, filteredNodesStatuses, err := g.findNodesThatFitPod(ctx, prof, state, pod)    ......    priorityList, err := g.prioritizeNodes(ctx, prof, state, pod, filteredNodes)    ......    host, err := g.selectHost(priorityList)    ......    return ScheduleResult{        SuggestedHost:  host,        EvaluatedNodes: len(filteredNodes) + len(filteredNodesStatuses),        FeasibleNodes:  len(filteredNodes),    }, err}

// Unmarshal the network config, and perform validationconf := *.NetConf{}// Determine which node name to use.nodename := utils.DetermineNodename(conf)// Extract WEP identifiers such as pod name, pod namespace (for k8s), containerID, IfName.wepIDs, err := utils.GetIdentifiers(args, nodename)calicoClient, err := utils.CreateClient(conf)// Calculate the workload name prefix from the WEP specific identifiers// for the given orchestrator.wepPrefix, err := wepIDs.CalculateWorkloadEndpointName(true)  // Check if there's an existing endpoint by listing the existing endpoints based on the WEP name prefix.endpoints, err := calicoClient.WorkloadEndpoints().List(ctx, options.ListOptions{Name: wepPrefix, Namespace: wepIDs.Namespace, Prefix: true})var endpoint *api.WorkloadEndpointfor _, ep := range endpoints.Items {  match, err := wepIDs.WorkloadEndpointIdentifiers.NameMatches(ep.Name)  if match {    endpoint = &ep    // Assign the WEP name to wepIDs' WEPName field.    wepIDs.WEPName = endpoint.Name    // Put the endpoint name from the matched WEP in the identifiers.    wepIDs.Endpoint = ep.Spec.Endpoint    break  }}// If we don't find a match from the existing WorkloadEndpoints then we calculate// the WEP name with the IfName passed in so we can create the WorkloadEndpoint later in the process.if endpoint == nil {    wepIDs.Endpoint = args.IfName    wepIDs.WEPName, err = wepIDs.CalculateWorkloadEndpointName(false)}

// 1) Run the IPAM plugin and make sure there's an IP address returned.ipamResult, err := ipam.ExecAdd(conf.IPAM.Type, args.StdinData)

// 2) Create the endpoint objectendpoint = api.NewWorkloadEndpoint()endpoint.Name = wepIDs.WEPNameendpoint.Namespace = wepIDs.Namespaceendpoint.Spec.Endpoint = wepIDs.Endpointendpoint.Spec.Node = wepIDs.Nodeendpoint.Spec.Orchestrator = wepIDs.Orchestratorendpoint.Spec.ContainerID = wepIDs.ContainerIDendpoint.Labels = labelsendpoint.Spec.Profiles = []string{profileID}

// 3) Set up the vethd, err := dataplane.GetDataplane(conf, logger)// Select the first 11 characters of the containerID for the host veth.desiredVethName := "cali" + args.ContainerID[:utils.Min(11, len(args.ContainerID))]hostVethName, contVethMac, err := d.DoNetworking(args, result, desiredVethName, utils.DefaultRoutes, endpoint, map[string]string{})endpoint.Spec.MAC = contVethMacendpoint.Spec.InterfaceName = hostVethName

veth := &netlink.Veth{      LinkAttrs: netlink.LinkAttrs{        Name:  contVethName,        Flags: net.FlagUp,        MTU:   d.mtu,      },      PeerName: hostVethName, }  netlink.LinkAdd(veth)

hostVeth, err := netlink.LinkByName(hostVethName) netlink.LinkSetHardwareAddr(hostVeth, mac)netlink.LinkSetUp(hostVeth)

// Fetch the MAC from the container Veth. This is needed by Calico.contVethMAC = contVeth.Attrs().HardwareAddr.String()// Add a connected route to a dummy next hop so that a default route can be setgw := net.IPv4(169, 254, 1, 1)gwNet := &net.IPNet{IP: gw, Mask: net.CIDRMask(32, 32)}err := netlink.RouteAdd(    &netlink.Route{      LinkIndex: contVeth.Attrs().Index,      Scope:     netlink.SCOPE_LINK,      Dst:       gwNet,    },)for _, r := range routes {   ip.AddRoute(r, gw, contVeth)} // Now add the IPs to the container side of the veth.for _, addr := range result.IPs {    netlink.AddrAdd(contVeth, &netlink.Addr{IPNet: &addr.Address})}d.configureContainerSysctls(hasIPv4, hasIPv6)

// Now that the everything has been successfully set up in the container, move the "host" end of the// veth into the host namespace.netlink.LinkSetNsFd(hostVeth, int(hostNS.Fd()))d.configureSysctls(hostVethName, hasIPv4, hasIPv6)// Moving a veth between namespaces always leaves it in the "DOWN" state. Set it back to "UP" now that we're// back in the host namespace.hostVeth, err := netlink.LinkByName(hostVethName)netlink.LinkSetUp(hostVeth)// Now that the host side of the veth is moved, state set to UP, and configured with sysctls, we can add the routes to it in the host namespace.SetupRoutes(hostVeth, result)

apiVersion: v1kind: Servicemetadata:  labels:    helm.sh/chart: ingress-nginx-2.4.0    app.kubernetes.io/name: ingress-nginx    app.kubernetes.io/instance: ingress-nginx    app.kubernetes.io/version: 0.33.0    app.kubernetes.io/managed-by: Helm    app.kubernetes.io/component: controller  name: ingress-nginx-controller  namespace: ingress-nginxspec:  type: NodePort  ports:    - name: http      port: 80      protocol: TCP      targetPort: http    - name: https      port: 443      protocol: TCP      targetPort: https  selector:    app.kubernetes.io/name: ingress-nginx    app.kubernetes.io/instance: ingress-nginx    app.kubernetes.io/component: controller---apiVersion: apps/v1kind: Deploymentmetadata:  labels:    helm.sh/chart: ingress-nginx-2.4.0    app.kubernetes.io/name: ingress-nginx    app.kubernetes.io/instance: ingress-nginx    app.kubernetes.io/version: 0.33.0    app.kubernetes.io/managed-by: Helm    app.kubernetes.io/component: controller  name: ingress-nginx-controller  namespace: ingress-nginxspec:  selector:    matchLabels:      app.kubernetes.io/name: ingress-nginx      app.kubernetes.io/instance: ingress-nginx      app.kubernetes.io/component: controller  revisionHistoryLimit: 10  minReadySeconds:   template:    metadata:      labels:        app.kubernetes.io/name: ingress-nginx        app.kubernetes.io/instance: ingress-nginx        app.kubernetes.io/component: controller    spec:      dnsPolicy: ClusterFirst      containers:        - name: controller          image: quay.io/kubernetes-ingress-controller/nginx-ingress-controller:0.33.0          imagePullPolicy: IfNotPresent          args:            - /nginx-ingress-controller            - --election-id=ingress-controller-leader            - --ingress-class=nginx            - --configmap=ingress-nginx/ingress-nginx-controller            - --validating-webhook=:8443            - --validating-webhook-certificate=/usr/local/certificates/cert            - --validating-webhook-key=/usr/local/certificates/key            runAsUser: 101            allowPrivilegeEscalation: true          env:            - name: POD_NAME              valueFrom:                fieldRef:                  fieldPath: metadata.name            - name: POD_NAMESPACE              valueFrom:                fieldRef:                  fieldPath: metadata.namespace          ports:            - name: http              containerPort: 80              protocol: TCP            - name: https              containerPort: 443              protocol: TCP            - name: webhook              containerPort: 8443              protocol: TCP          volumeMounts:            - name: webhook-cert              mountPath: /usr/local/certificates/              readOnly: true          resources:            requests:              cpu: 100m              memory: 90Mi---

kubeClient, err := createApiserverClient(conf.APIServerHost, conf.RootCAFile, conf.KubeConfigFile)conf.Client = kubeClientngx := controller.NewNGINXController(conf, mc)go ngx.Start()

  n := &NGINXController{    isIPV6Enabled: ing_net.IsIPv6Enabled(),    resolver:        h,    cfg:             config,    syncRateLimiter: flowcontrol.NewTokenBucketRateLimiter(config.SyncRateLimit, 1),    recorder: eventBroadcaster.NewRecorder(scheme.Scheme, apiv1.EventSource{      Component: "nginx-ingress-controller",    }),    stopCh:   make(chan struct{}),    updateCh: channels.NewRingChannel(1024),    ngxErrCh: make(chan error),    stopLock: &sync.Mutex{},    runningConfig: new(ingress.Configuration),    Proxy: &TCPProxy{},    metricCollector: mc,    command: NewNginxCommand(),  }

  n.store = store.New(    config.Namespace,    config.ConfigMapName,    config.TCPConfigMapName,    config.UDPConfigMapName,    config.DefaultSSLCertificate,    config.ResyncPeriod,    config.Client,    n.updateCh,    pod,    config.DisableCatchAll)

  store := &k8sStore{    informers:             &Informer{},    listers:               &Lister{},    sslStore:              NewSSLCertTracker(),    updateCh:              updateCh,    backendConfig:         ngx_config.NewDefault(),    syncSecretMu:          &sync.Mutex{},    backendConfigMu:       &sync.RWMutex{},    secretIngressMap:      NewObjectRefMap(),    defaultSSLCertificate: defaultSSLCertificate,    pod:                   pod,  }

// create informers factory, enable and assign required informers  infFactory := informers.NewSharedInformerFactoryWithOptions(client, resyncPeriod,    informers.WithNamespace(namespace),    informers.WithTweakListOptions(tweakListOptionsFunc))store.informers.Ingress = infFactory.Networking().V1beta1().Ingresses().Informer()store.informers.Endpoint = infFactory.Core().V1().Endpoints().Informer()store.informers.Service = infFactory.Core().V1().Services().Informer()

  ingEventHandler := cache.ResourceEventHandlerFuncs{    AddFunc: func(obj interface{}) {      ing, _ := toIngress(obj)      store.syncIngress(ing)      store.updateSecretIngressMap(ing)      store.syncSecrets(ing)      updateCh.In() <- Event{        Type: CreateEvent,        Obj:  obj,      }    },    DeleteFunc: ingDeleteHandler,    UpdateFunc: func(old, cur interface{}) {......    },  }

n.syncQueue = task.NewTaskQueue(n.syncIngress)

// Start starts a new NGINX master process running in the foreground.func (n *NGINXController) Start() {  klog.Info("Starting NGINX Ingress controller")  n.store.Run(n.stopCh)......  cmd := n.command.ExecCommand()......  klog.Info("Starting NGINX process")  n.start(cmd)  go n.syncQueue.Run(time.Second, n.stopCh)......  for {    select {......    case event := <-n.updateCh.Out():      if evt, ok := event.(store.Event); ok {        klog.V(3).Infof("Event %v received - object %v", evt.Type, evt.Obj)        n.syncQueue.EnqueueSkippableTask(evt.Obj)      } else {        klog.Warningf("Unexpected event type received %T", event)      }......  }}

// OnUpdate is called by the synchronization loop whenever configuration// changes were detected. The received backend Configuration is merged with the// configuration ConfigMap before generating the final configuration file.// Returns nil in case the backend was successfully reloaded.func (n *NGINXController) OnUpdate(ingressCfg ingress.Configuration) error {  cfg := n.store.GetBackendConfiguration()  cfg.Resolver = n.resolver
  content, err := n.generateTemplate(cfg, ingressCfg)......  err = n.testTemplate(content)......  err = ioutil.WriteFile(cfgPath, content, file.ReadWriteByUser)......  o, err := n.command.ExecCommand("-s", "reload").CombinedOutput()......}

// NewIdentityServer initialize a identity server for rbd CSI driverfunc NewIdentityServer(d *csicommon.CSIDriver) *IdentityServer {  return &IdentityServer{    DefaultIdentityServer: csicommon.NewDefaultIdentityServer(d),  }}
// NewControllerServer initialize a controller server for rbd CSI driverfunc NewControllerServer(d *csicommon.CSIDriver, cachePersister util.CachePersister) *ControllerServer {  return &ControllerServer{    DefaultControllerServer: csicommon.NewDefaultControllerServer(d),    MetadataStore:           cachePersister,    VolumeLocks:             util.NewVolumeLocks(),    SnapshotLocks:           util.NewVolumeLocks(),  }}
// NewNodeServer initialize a node server for rbd CSI driver.func NewNodeServer(d *csicommon.CSIDriver, t string, topology map[string]string) (*NodeServer, error) {  mounter := mount.New("")  return &NodeServer{    DefaultNodeServer: csicommon.NewDefaultNodeServer(d, t, topology),    mounter:           mounter,    VolumeLocks:       util.NewVolumeLocks(),  }, nil}

// CreateVolume creates the volume in backendfunc (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {......  err = rbdVol.Connect(cr)......  rbdSnap, err := cs.checkSnapshotSource(ctx, req, cr)......  err = reserveVol(ctx, rbdVol, rbdSnap, cr)......  err = createBackingImage(ctx, cr, rbdVol, rbdSnap)......  volumeContext := req.GetParameters()  volumeContext["pool"] = rbdVol.Pool  volumeContext["journalPool"] = rbdVol.JournalPool  volumeContext["imageName"] = rbdVol.RbdImageName  volume := &csi.Volume{    VolumeId:      rbdVol.VolID,    CapacityBytes: rbdVol.VolSize,    VolumeContext: volumeContext,    ContentSource: req.GetVolumeContentSource(),  }  return &csi.CreateVolumeResponse{Volume: volume}, nil}

// NodePublishVolume mounts the volume mounted to the device path to the target// pathfunc (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {......  targetPath := req.GetTargetPath()  isBlock := req.GetVolumeCapability().GetBlock() != nil  stagingPath := req.GetStagingTargetPath()  volID := req.GetVolumeId()  stagingPath += "/" + volID......  // Check if that target path exists properly  notMnt, err := ns.createTargetMountPath(ctx, targetPath, isBlock)......  // Publish Path  err = ns.mountVolume(ctx, stagingPath, req)......  return &csi.NodePublishVolumeResponse{}, nil}

apiVersion: storage.k8s.io/v1kind: StorageClassmetadata:  name: standardprovisioner: kubernetes.io/aws-ebsparameters:  type: gp2reclaimPolicy: RetainallowVolumeExpansion: truemountOptions:  - debugvolumeBindingMode: Immediate

kind: StorageClassapiVersion: storage.k8s.io/v1metadata:  name: fast-storageprovisioner: csi-driver.example.comparameters:  type: pd-ssd  csi.storage.k8s.io/provisioner-secret-name: mysecret  csi.storage.k8s.io/provisioner-secret-namespace: mynamespace

apiVersion: v1kind: PersistentVolumemetadata:  name: pv0003spec:  capacity:    storage: 5Gi  volumeMode: Filesystem  accessModes:    - ReadWriteOnce  persistentVolumeReclaimPolicy: Recycle  storageClassName: slow  mountOptions:    - hard    - nfsvers=4.1  nfs:    path: /tmp    server: 172.17.0.2

etcdctl snapshot --endpoints=$ETCD_SERVERS --cacert=/var/lib/etcd/cert/ca.pem --cert=/var/lib/etcd/cert/etcd-client.pem --key=/var/lib/etcd/cert/etcd-client-key.pem save /var/lib/etcd_backup/backup_$(date "+%Y%m%d%H%M%S").db

set -xexport ETCD_NAME=$(cat /usr/lib/systemd/system/etcd.service|grep ExecStart|grep -Eo "name.*-name-[0-9].*--client"|awk '{print $2}')export ETCD_CLUSTER=$(cat /usr/lib/systemd/system/etcd.service|grep ExecStart|grep -Eo "initial-cluster.*--initial"|awk '{print $2}')export ETCD_INITIAL_CLUSTER_TOKEN=$(cat /usr/lib/systemd/system/etcd.service|grep ExecStart|grep -Eo "initial-cluster-token.*"|awk '{print $2}')export ETCD_INITIAL_ADVERTISE_PEER_URLS=$(cat /usr/lib/systemd/system/etcd.service|grep ExecStart|grep -Eo "initial-advertise-peer-urls.*--listen-peer"|awk '{print $2}')ETCDCTL_API=3 etcdctl snapshot --cacert=/var/lib/etcd/cert/ca.pem --cert=/var/lib/etcd/cert/etcd-client.pem --key=/var/lib/etcd/cert/etcd-client-key.pem  restore /var/lib/etcd_backup/backup_xxx.db \  --name $ETCD_NAME \  --data-dir /var/lib/etcd/data.etcd \  --initial-cluster $ETCD_CLUSTER \  --initial-cluster-token $ETCD_INITIAL_CLUSTER_TOKEN \  --initial-advertise-peer-urls $ETCD_INITIAL_ADVERTISE_PEER_URLSchown -R etcd:etcd /var/lib/etcd/data.etcd

NAME    STATUS   ROLES    AGE     VERSION   LABELSnode1   Ready    <none>   4m26s   v1.16.   node=node1,zone=zoneAnode2   Ready    <none>   3m58s   v1.16.   node=node2,zone=zoneAnode3   Ready    <none>   3m17s   v1.16.   node=node3,zone=zoneBnode4   Ready    <none>   2m43s   v1.16.   node=node4,zone=zoneB

kind: PodapiVersion: v1metadata:  name: mypod  labels:    foo: barspec:  topologySpreadConstraints:  - maxSkew: 1    topologyKey: zone    whenUnsatisfiable: DoNotSchedule    labelSelector:      matchLabels:        foo: bar  - maxSkew: 1    topologyKey: node    whenUnsatisfiable: DoNotSchedule    labelSelector:      matchLabels:        foo: bar  containers:  - name: pause    image: k8s.gcr.io/pause:3.1

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"    - "*"

c := controller.New(cfg)err := c.Start()

func (c *Controller) Start() error {  err := c.initResource()......  c.run()......}

func (c *Controller) initResource() error {  if c.Config.CreateCRD {    err := c.initCRD()  }  return nil}
func (c *Controller) initCRD() error {  err := k8sutil.CreateCRD(c.KubeExtCli, api.EtcdClusterCRDName, api.EtcdClusterResourceKind, api.EtcdClusterResourcePlural, "etcd")  return k8sutil.WaitCRDReady(c.KubeExtCli, api.EtcdClusterCRDName)}

func (c *Controller) run() {......  source := cache.NewListWatchFromClient(    c.Config.EtcdCRCli.EtcdV1beta2().RESTClient(),    api.EtcdClusterResourcePlural,    ns,    fields.Everything())......  _, informer := cache.NewIndexerInformer(source, &api.EtcdCluster{}, , cache.ResourceEventHandlerFuncs{    AddFunc:    c.onAddEtcdClus,    UpdateFunc: c.onUpdateEtcdClus,    DeleteFunc: c.onDeleteEtcdClus,  }, cache.Indexers{})......  informer.Run(ctx.Done())}

// handleClusterEvent returns true if cluster is ignored (not managed) by this instance.func (c *Controller) handleClusterEvent(event *Event) (bool, error) {  clus := event.Object......  switch event.Type {  case kwatch.Added:......    nc := cluster.New(c.makeClusterConfig(), clus)    c.clusters[getNamespacedName(clus)] = nc......  case kwatch.Modified:......    c.clusters[getNamespacedName(clus)].Update(clus)......  case kwatch.Deleted:    c.clusters[getNamespacedName(clus)].Delete()    delete(c.clusters, getNamespacedName(clus))......  }}

func New(config Config, cl *api.EtcdCluster) *Cluster {  c := &Cluster{    logger:    lg,    config:    config,    cluster:   cl,    eventCh:   make(chan *clusterEvent, 100),    stopCh:    make(chan struct{}),    status:    *(cl.Status.DeepCopy()),    eventsCli: config.KubeCli.Core().Events(cl.Namespace),  }  go func() {    if err := c.setup(); err != nil {......    }    c.run()  }()  return c}

func (c *Cluster) setup() error {......  if shouldCreateCluster {    return c.create()  }......}
func (c *Cluster) create() error {  c.status.SetPhase(api.ClusterPhaseCreating)  if err := c.updateCRStatus(); err != nil {......  }  return c.prepareSeedMember()}
func (c *Cluster) updateCRStatus() error {  if reflect.DeepEqual(c.cluster.Status, c.status) {    return nil  }  newCluster := c.cluster  newCluster.Status = c.status  newCluster, err := c.config.EtcdCRCli.EtcdV1beta2().EtcdClusters(c.cluster.Namespace).Update(c.cluster)  if err != nil {    return fmt.Errorf("failed to update CR status: %v", err)  }  c.cluster = newCluster  return nil}

func (c *Cluster) prepareSeedMember() error {  err := c.bootstrap()......}// bootstrap creates the seed etcd member for a new cluster.func (c *Cluster) bootstrap() error {  return c.startSeedMember()}
func (c *Cluster) startSeedMember() error {  m := &etcdutil.Member{    Name:         k8sutil.UniqueMemberName(c.cluster.Name),    Namespace:    c.cluster.Namespace,    SecurePeer:   c.isSecurePeer(),    SecureClient: c.isSecureClient(),  }  ms := etcdutil.NewMemberSet(m)  if err := c.createPod(ms, m, "new"); err != nil {    return fmt.Errorf("failed to create seed member (%s): %v", m.Name, err)  }  c.members = ms
  _, err := c.eventsCli.Create(k8sutil.NewMemberAddEvent(m.Name, c.cluster))  return nil}

func (c *Cluster) run() {  c.status.ServiceName = k8sutil.ClientServiceName(c.cluster.Name)  c.status.ClientPort = k8sutil.EtcdClientPort  c.status.SetPhase(api.ClusterPhaseRunning)  for {    select {......    case <-time.After(reconcileInterval):      start := time.Now()......      running, pending, err := c.pollPods()......      // On controller restore, we could have "members == nil"      if rerr != nil || c.members == nil {        rerr = c.updateMembers(podsToMemberSet(running, c.isSecureClient()))      }      rerr = c.reconcile(running)......      c.updateMemberStatus(running)......    }......  }}

// reconcile reconciles cluster current state to desired state specified by spec.// - it tries to reconcile the cluster to desired size.// - if the cluster needs for upgrade, it tries to upgrade old member one by one.func (c *Cluster) reconcile(pods []*v1.Pod) error {......  sp := c.cluster.Spec  running := podsToMemberSet(pods, c.isSecureClient())  if !running.IsEqual(c.members) || c.members.Size() != sp.Size {    return c.reconcileMembers(running)  }  c.status.ClearCondition(api.ClusterConditionScaling)  if needUpgrade(pods, sp) {    c.status.UpgradeVersionTo(sp.Version)    m := pickOneOldMember(pods, sp.Version)    return c.upgradeOneMember(m.Name)  }  c.status.ClearCondition(api.ClusterConditionUpgrading)  c.status.SetVersion(sp.Version)  c.status.SetReadyCondition()  return nil}
// reconcileMembers reconciles// - running pods on k8s and cluster membership// - cluster membership and expected size of etcd cluster// Steps:// 1. Remove all pods from running set that does not belong to member set.// 2. L consist of remaining pods of runnings// 3. If L = members, the current state matches the membership state. END.// 4. If len(L) < len(members)/2 + 1, return quorum lost error.// 5. Add one missing member. END.func (c *Cluster) reconcileMembers(running etcdutil.MemberSet) error {  unknownMembers := running.Diff(c.members)  if unknownMembers.Size() >  {    c.logger.Infof("removing unexpected pods: %v", unknownMembers)    for _, m := range unknownMembers {      if err := c.removePod(m.Name); err != nil {        return err      }    }  }  L := running.Diff(unknownMembers)  if L.Size() == c.members.Size() {    return c.resize()  }  if L.Size() < c.members.Size()/2+1 {    return ErrLostQuorum  }  c.logger.Infof("removing one dead member")  // remove dead members that doesn't have any running pods before doing resizing.  return c.removeDeadMember(c.members.Diff(L).PickOne())}