KubeVirt Domain Informer 原理

func NewController(
    recorder record.EventRecorder,
    clientset kubecli.KubevirtClient,
    host string,
    migrationIpAddress string,
    virtShareDir string,
    virtPrivateDir string,
    kubeletPodsDir string,
    vmiSourceInformer cache.SharedIndexInformer,
    vmiTargetInformer cache.SharedIndexInformer,
    domainInformer cache.SharedInformer,
   // a lot of parameters here
) (*VirtualMachineController, error) {

    queue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "virt-handler-vm")

    c := &VirtualMachineController{
        Queue:                       queue,
        recorder:                    recorder,
        clientset:                   clientset,
        host:                        host,
        migrationIpAddress:          migrationIpAddress,
        virtShareDir:                virtShareDir,
        vmiSourceInformer:           vmiSourceInformer,
        vmiTargetInformer:           vmiTargetInformer,
        domainInformer:              domainInformer,
    }

    _, err = domainInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    c.addDomainFunc,
        DeleteFunc: c.deleteDomainFunc,
        UpdateFunc: c.updateDomainFunc,
    })
    if err != nil {
        return nil, err
    }
    // a lot of code here
}

func (app *virtHandlerApp) Run() {
    // a lot of code here
    domainSharedInformer, err := virtcache.NewSharedInformer(app.VirtShareDir, int(app.WatchdogTimeoutDuration.Seconds()), recorder, vmiSourceInformer.GetStore(), time.Duration(app.domainResyncPeriodSeconds)*time.Second)
    if err != nil {
        panic(err)
    }
    // a lot of code here
}

func (f *kubeInformerFactory) VMI() cache.SharedIndexInformer {
    return f.getInformer("vmiInformer", func() cache.SharedIndexInformer {
        lw := cache.NewListWatchFromClient(f.restClient, "virtualmachineinstances", k8sv1.NamespaceAll, fields.Everything())
        return cache.NewSharedIndexInformer(lw, &kubev1.VirtualMachineInstance{}, f.defaultResync, GetVMIInformerIndexers())
    })
}

func NewSharedInformer(virtShareDir string, watchdogTimeout int, recorder record.EventRecorder, vmiStore cache.Store, resyncPeriod time.Duration) (cache.SharedInformer, error) {
    lw := newListWatchFromNotify(virtShareDir, watchdogTimeout, recorder, vmiStore, resyncPeriod)
    informer := cache.NewSharedInformer(lw, &api.Domain{}, 0)
    return informer, nil
}

type DomainWatcher struct {
    lock                     sync.Mutex
    wg                       sync.WaitGroup
    stopChan                 chan struct{}
    eventChan                chan watch.Event
    backgroundWatcherStarted bool
    virtShareDir             string
    watchdogTimeout          int
    recorder                 record.EventRecorder
    vmiStore                 cache.Store
    resyncPeriod             time.Duration

    watchDogLock        sync.Mutex
    unresponsiveSockets map[string]int64
}

// Lister is any object that knows how to perform an initial list.
type Lister interface {
    // List should return a list type object; the Items field will be extracted, and the
    // ResourceVersion field will be used to start the watch in the right place.
    List(options metav1.ListOptions) (runtime.Object, error)
}

// Watcher is any object that knows how to start a watch on a resource.
type Watcher interface {
    // Watch should begin a watch at the specified version.
    Watch(options metav1.ListOptions) (watch.Interface, error)
}

// ListerWatcher is any object that knows how to perform an initial list and start a watch on a resource.
type ListerWatcher interface {
    Lister
    Watcher
}

func (d *DomainWatcher) List(_ k8sv1.ListOptions) (runtime.Object, error) {

    log.Log.V(3).Info("Synchronizing domains")
    err := d.startBackground()
    if err != nil {
        return nil, err
    }

    domains, err := d.listAllKnownDomains()
    if err != nil {
        return nil, err
    }

    list := api.DomainList{
        Items: []api.Domain{},
    }

    for _, domain := range domains {
        list.Items = append(list.Items, *domain)
    }
    return &list, nil
}

func (d *DomainWatcher) listAllKnownDomains() ([]*api.Domain, error) {
    var domains []*api.Domain

    socketFiles, err := listSockets()
    if err != nil {
        return nil, err
    }
    for _, socketFile := range socketFiles {

        exists, err := diskutils.FileExists(socketFile)
        if err != nil {
            log.Log.Reason(err).Error("failed access cmd client socket")
            continue
        }

        if !exists {
            record, recordExists := findGhostRecordBySocket(socketFile)
            if recordExists {
                domain := api.NewMinimalDomainWithNS(record.Namespace, record.Name)
                domain.ObjectMeta.UID = record.UID
                now := k8sv1.Now()
                domain.ObjectMeta.DeletionTimestamp = &now
                log.Log.Object(domain).Warning("detected stale domain from ghost record")
                domains = append(domains, domain)
            }
            continue
        }

        log.Log.V(3).Infof("List domains from sock %s", socketFile)
        client, err := cmdclient.NewClient(socketFile)
        if err != nil {
            log.Log.Reason(err).Error("failed to connect to cmd client socket")
            // Ignore failure to connect to client.
            // These are all local connections via unix socket.
            // A failure to connect means there's nothing on the other
            // end listening.
            continue
        }
        defer client.Close()

        domain, exists, err := client.GetDomain()
        if err != nil {
            log.Log.Reason(err).Error("failed to list domains on cmd client socket")
            // Failure to get domain list means that client
            // was unable to contact libvirt. As soon as the connection
            // is restored on the client's end, a domain notification will
            // be sent.
            continue
        }
        if exists == true {
            domains = append(domains, domain)
        }
    }
    return domains, nil
}

$ kubectl get vmi ecs-rocky8-10 -n ns-demo
NAME            AGE   PHASE     IP              NODENAME   READY
ecs-rocky8-10   11d   Running   192.168.100.4   node173    True

$ kubectl get po -n ns-demo | grep ecs-rocky8-10
virt-launcher-ecs-rocky8-10-bxknl            1/1     Running   0               11d

$ kubectl get po virt-launcher-ecs-rocky8-10-bxknl -n ns-demo -o jsonpath='{.metadata.uid}'
69e41c74-7e47-45f7-a4e4-9c193efcf5f8

$ ll /var/lib/kubelet/pods/69e41c74-7e47-45f7-a4e4-9c193efcf5f8/volumes/kubernetes.io~empty-dir/
total 8
drwxrwsrwx  2 root qemu    6 Jun 16 09:09 container-disks
drwxrwsrwx  5 root qemu   67 Jun 17 11:32 ephemeral-disks
drwxrwsrwx  2 root qemu    6 Jun 16 09:09 hotplug-disks
drwxrwsrwx  5 root qemu 4096 Jun 27 10:51 libvirt-runtime
drwxrwsrwx 10 root qemu 4096 Jun 17 11:32 private
drwxrwsrwx  5 root qemu   96 Jun 20 10:25 public
drwxrwsrwx  2 root qemu   27 Jun 17 11:32 sockets
drwxrwsrwx  2 root qemu    6 Jun 16 09:09 vhostuser-sockets

$ stat /var/lib/kubelet/pods/69e41c74-7e47-45f7-a4e4-9c193efcf5f8/volumes/kubernetes.io~empty-dir/sockets/launcher-sock
  File: /var/lib/kubelet/pods/69e41c74-7e47-45f7-a4e4-9c193efcf5f8/volumes/kubernetes.io~empty-dir/sockets/launcher-sock
  Size: 0          Blocks: 0          IO Block: 4096   socket
Device: fd00h/64768d Inode: 3267213     Links: 1
Access: (0755/srwxr-xr-x)  Uid: (  107/    qemu)   Gid: (  107/    qemu)
Access: 2024-06-26 11:32:45.034447042 +0800
Modify: 2024-06-17 11:32:21.331893299 +0800
Change: 2024-06-17 11:32:21.342893055 +0800
 Birth: 2024-06-17 11:32:21.331893299 +0800

# virt-handler-r64rw 与 virt-launcher Pod 同节点
$ kubectl exec -it virt-handler-r64rw -n kubevirt -- ls -al /var/lib/kubelet/pods/69e41c74-7e47-45f7-a4e4-9c193efcf5f8/volumes/kubernetes.io~empty-dir/sockets/launcher-sock
Defaulted container "virt-handler" out of: virt-handler, virt-launcher (init)
srwxr-xr-x 1 qemu qemu 0 Jun 17 03:32 /var/lib/kubelet/pods/69e41c74-7e47-45f7-a4e4-9c193efcf5f8/volumes/kubernetes.io~empty-dir/sockets/launcher-sock

// Interface can be implemented by anything that knows how to watch and report changes.
type Interface interface {
    // Stop stops watching. Will close the channel returned by ResultChan(). Releases
    // any resources used by the watch.
    Stop()

    // ResultChan returns a chan which will receive all the events. If an error occurs
    // or Stop() is called, the implementation will close this channel and
    // release any resources used by the watch.
    ResultChan() <-chan Event
}

func (d *DomainWatcher) Stop() {
    d.lock.Lock()
    defer d.lock.Unlock()

    if d.backgroundWatcherStarted == false {
        return
    }
    close(d.stopChan)
    d.wg.Wait()
    d.backgroundWatcherStarted = false
    close(d.eventChan)
}

func (d *DomainWatcher) ResultChan() <-chan watch.Event {
    return d.eventChan
}

func (d *DomainWatcher) startBackground() error {
    // a lot of code here
    go func() {
        defer d.wg.Done()

        resyncTicker := time.NewTicker(d.resyncPeriod)
        resyncTickerChan := resyncTicker.C
        defer resyncTicker.Stop()

        // Divide the watchdogTimeout by 3 for our ticker.
        // This ensures we always have at least 2 response failures
        // in a row before we mark the socket as unavailable (which results in shutdown of VMI)
        expiredWatchdogTicker := time.NewTicker(time.Duration((d.watchdogTimeout/3)+1) * time.Second)
        defer expiredWatchdogTicker.Stop()

        expiredWatchdogTickerChan := expiredWatchdogTicker.C

        srvErr := make(chan error)
        go func() {
            defer close(srvErr)
            err := notifyserver.RunServer(d.virtShareDir, d.stopChan, d.eventChan, d.recorder, d.vmiStore)
            srvErr <- err
        }()

        for {
            select {
            case <-resyncTickerChan:
                d.handleResync()
            case <-expiredWatchdogTickerChan:
                d.handleStaleWatchdogFiles()
                d.handleStaleSocketConnections()
            case err := <-srvErr:
                if err != nil {
                    log.Log.Reason(err).Errorf("Unexpected err encountered with Domain Notify aggregation server")
                }

                // server exitted so this goroutine is done.
                return
            }
        }
    }()

    d.backgroundWatcherStarted = true
    return nil
}

func (d *DomainWatcher) handleResync() {
    socketFiles, err := listSockets()
    if err != nil {
        log.Log.Reason(err).Error("failed to list sockets")
        return
    }

    log.Log.Infof("resyncing virt-launcher domains")
    for _, socket := range socketFiles {
        client, err := cmdclient.NewClient(socket)
        if err != nil {
            log.Log.Reason(err).Error("failed to connect to cmd client socket during resync")
            // Ignore failure to connect to client.
            // These are all local connections via unix socket.
            // A failure to connect means there's nothing on the other
            // end listening.
            continue
        }
        defer client.Close()

        domain, exists, err := client.GetDomain()
        if err != nil {
            // this resync is best effort only.
            log.Log.Reason(err).Errorf("unable to retrieve domain at socket %s during resync", socket)
            continue
        } else if !exists {
            // nothing to sync if it doesn't exist
            continue
        }

        d.eventChan <- watch.Event{Type: watch.Modified, Object: domain}
    }
}

    switch domain.Status.Status {
    case api.Shutoff, api.Crashed:
        switch domain.Status.Reason {
        case api.ReasonCrashed, api.ReasonPanicked:
            return v1.Failed, nil
        case api.ReasonDestroyed:
            // When ACPI is available, the domain was tried to be shutdown,
            // and destroyed means that the domain was destroyed after the graceperiod expired.
            // Without ACPI a destroyed domain is ok.
            if isACPIEnabled(vmi, domain) {
                return v1.Failed, nil
            }
            return v1.Succeeded, nil
        case api.ReasonShutdown, api.ReasonSaved, api.ReasonFromSnapshot:
            return v1.Succeeded, nil
        case api.ReasonMigrated:
            // if the domain migrated, we no longer know the phase.
            return vmi.Status.Phase, nil
        }
    case api.Running, api.Paused, api.Blocked, api.PMSuspended:
        return v1.Running, nil
    }