Merge pull request kubevirt#4686 from davidvossel/launcher-updates-v1

Automated VMI Workload Component Updates

api/openapi-spec/swagger.json

@@ -10290,6 +10290,10 @@
       "description": "Specifies if kubevirt can be deleted if workloads are still present. This is mainly a precaution to avoid accidental data loss",
       "type": "string"
      },
+     "workloadUpdateStrategy": {
+      "description": "WorkloadUpdateStrategy defines at the cluster level how to handle automated workload updates",
+      "$ref": "#/definitions/v1.KubeVirtWorkloadUpdateStrategy"
+     },
      "workloads": {
       "description": "selectors and tolerations that should apply to KubeVirt workloads",
       "$ref": "#/definitions/v1.ComponentConfig"
@@ -10322,6 +10326,10 @@
      "operatorVersion": {
       "type": "string"
      },
+     "outdatedVirtualMachineInstanceWorkloads": {
+      "type": "integer",
+      "format": "int32"
+     },
      "phase": {
       "type": "string"
      },
@@ -10339,6 +10347,29 @@
      }
     }
    },
+   "v1.KubeVirtWorkloadUpdateStrategy": {
+    "description": "KubeVirtWorkloadUpdateStrategy defines options related to updating a KubeVirt install",
+    "type": "object",
+    "properties": {
+     "batchEvictionInterval": {
+      "description": "BatchEvictionInterval Represents the interval to wait before issuing the next batch of shutdowns\n\nDefaults to 1 minute",
+      "$ref": "#/definitions/k8s.io.apimachinery.pkg.apis.meta.v1.Duration"
+     },
+     "batchEvictionSize": {
+      "description": "BatchEvictionSize Represents the number of VMIs that can be forced updated per the BatchShutdownInteral interval\n\nDefaults to 10",
+      "type": "integer",
+      "format": "int32"
+     },
+     "workloadUpdateMethods": {
+      "description": "WorkloadUpdateMethods defines the methods that can be used to disrupt workloads during automated workload updates. When multiple methods are present, the least disruptive method takes precedence over more disruptive methods. For example if both LiveMigrate and Shutdown methods are listed, only VMs which are not live migratable will be restarted/shutdown\n\nAn empty list defaults to no automated workload updating",
+      "type": "array",
+      "items": {
+       "type": "string"
+      },
+      "x-kubernetes-list-type": "atomic"
+     }
+    }
+   },
    "v1.LogVerbosity": {
     "description": "LogVerbosity sets log verbosity level of  various components",
     "type": "object",
@@ -11798,6 +11829,10 @@
        "$ref": "#/definitions/v1.VirtualMachineInstanceNetworkInterface"
       }
      },
+     "launcherContainerImageVersion": {
+      "description": "LauncherContainerImageVersion indicates what container image is currently active for the vmi.",
+      "type": "string"
+     },
      "migrationMethod": {
       "description": "Represents the method using which the vmi can be migrated: live migration or block migration",
       "type": "string"

docs/metrics.md

@@ -132,3 +132,7 @@ A design proposal and its implementation history can be seen [here](https://docs
 #### HELP leading_virt_controller Indication for an operating virt-controller.
 ## ready_virt_controller
 #### HELP ready_virt_controller Indication for a virt-controller that is ready to take the lead.
+
+ # Other Metrics 
+## kubevirt_vmi_outdated_count
+#### HELP kubevirt_vmi_outdated_count Indication for the number of VirtualMachineInstance workloads that are not running within the most up-to-date version of the virt-launcher environment.

hack/prom-rule-ci/rule-spec-dumper.go

@@ -26,7 +26,7 @@ func main() {
 
 	targetFile := os.Args[1]
 
-	promRuleSpec := components.NewPrometheusRuleSpec("ci")
+	promRuleSpec := components.NewPrometheusRuleSpec("ci", true)
 	b, err := json.Marshal(promRuleSpec)
 	if err != nil {
 		panic(err)

manifests/generated/kubevirt-cr.yaml.in

@@ -9,3 +9,4 @@ spec:
   configuration: {}
   customizeComponents: {}
   imagePullPolicy: {{.ImagePullPolicy}}
+  workloadUpdateStrategy: {}

manifests/generated/kv-resource.yaml

@@ -636,6 +636,22 @@ spec:
             uninstallStrategy:
               description: Specifies if kubevirt can be deleted if workloads are still present. This is mainly a precaution to avoid accidental data loss
               type: string
+            workloadUpdateStrategy:
+              description: WorkloadUpdateStrategy defines at the cluster level how to handle automated workload updates
+              properties:
+                batchEvictionInterval:
+                  description: "BatchEvictionInterval Represents the interval to wait before issuing the next batch of shutdowns \n Defaults to 1 minute"
+                  type: string
+                batchEvictionSize:
+                  description: "BatchEvictionSize Represents the number of VMIs that can be forced updated per the BatchShutdownInteral interval \n Defaults to 10"
+                  type: integer
+                workloadUpdateMethods:
+                  description: "WorkloadUpdateMethods defines the methods that can be used to disrupt workloads during automated workload updates. When multiple methods are present, the least disruptive method takes precedence over more disruptive methods. For example if both LiveMigrate and Shutdown methods are listed, only VMs which are not live migratable will be restarted/shutdown \n An empty list defaults to no automated workload updating"
+                  items:
+                    type: string
+                  type: array
+                  x-kubernetes-list-type: atomic
+              type: object
             workloads:
               description: selectors and tolerations that should apply to KubeVirt workloads
               properties:
@@ -1052,6 +1068,8 @@ spec:
               type: string
             operatorVersion:
               type: string
+            outdatedVirtualMachineInstanceWorkloads:
+              type: integer
             phase:
               description: KubeVirtPhase is a label for the phase of a KubeVirt deployment at the current time.
               type: string

manifests/generated/operator-csv.yaml.in

@@ -465,6 +465,12 @@ spec:
           - pods/finalizers
           verbs:
           - update
+        - apiGroups:
+          - ""
+          resources:
+          - pods/eviction
+          verbs:
+          - create
         - apiGroups:
           - ""
           resources:

manifests/generated/rbac-operator.authorization.k8s.yaml.in

@@ -367,6 +367,12 @@ rules:
   - pods/finalizers
   verbs:
   - update
+- apiGroups:
+  - ""
+  resources:
+  - pods/eviction
+  verbs:
+  - create
 - apiGroups:
   - ""
   resources:

pkg/controller/controller.go

@@ -250,3 +250,41 @@ func ApplyVolumeRequestOnVMISpec(vmiSpec *v1.VirtualMachineInstanceSpec, request
 
 	return vmiSpec
 }
+
+func CurrentVMIPod(vmi *v1.VirtualMachineInstance, podInformer cache.SharedIndexInformer) (*k8sv1.Pod, error) {
+
+	// current pod is the most recent pod created on the current VMI node
+	// OR the most recent pod created if no VMI node is set.
+
+	// Get all pods from the namespace
+	objs, err := podInformer.GetIndexer().ByIndex(cache.NamespaceIndex, vmi.Namespace)
+	if err != nil {
+		return nil, err
+	}
+	pods := []*k8sv1.Pod{}
+	for _, obj := range objs {
+		pod := obj.(*k8sv1.Pod)
+		pods = append(pods, pod)
+	}
+
+	var curPod *k8sv1.Pod = nil
+	for _, pod := range pods {
+		if !IsControlledBy(pod, vmi) {
+			continue
+		}
+
+		if vmi.Status.NodeName != "" &&
+			vmi.Status.NodeName != pod.Spec.NodeName {
+			// This pod isn't scheduled to the current node.
+			// This can occur during the initial migration phases when
+			// a new target node is being prepared for the VMI.
+			continue
+		}
+
+		if curPod == nil || curPod.CreationTimestamp.Before(&pod.CreationTimestamp) {
+			curPod = pod
+		}
+	}
+
+	return curPod, nil
+}

pkg/util/migrations/migrations.go

@@ -7,7 +7,7 @@ import (
 	v1 "kubevirt.io/client-go/api/v1"
 )
 
-func ListUnfinishedMigrations(informer cache.SharedIndexInformer) ([]*v1.VirtualMachineInstanceMigration, error) {
+func ListUnfinishedMigrations(informer cache.SharedIndexInformer) []*v1.VirtualMachineInstanceMigration {
 	objs := informer.GetStore().List()
 	migrations := []*v1.VirtualMachineInstanceMigration{}
 	for _, obj := range objs {
@@ -16,7 +16,7 @@ func ListUnfinishedMigrations(informer cache.SharedIndexInformer) ([]*v1.Virtual
 			migrations = append(migrations, migration)
 		}
 	}
-	return migrations, nil
+	return migrations
 }
 
 func FilterRunningMigrations(migrations []v1.VirtualMachineInstanceMigration) []v1.VirtualMachineInstanceMigration {

pkg/util/status/status.go

@@ -130,6 +130,13 @@ func (u *updater) patchUnstructured(obj runtime.Object, patchType types.PatchTyp
 			return "", "", err
 		}
 		return oldObj.ResourceVersion, newObj.ResourceVersion, nil
+	case *v1.KubeVirt:
+		oldObj := obj.(*v1.KubeVirt)
+		newObj, err := u.cli.KubeVirt(a.GetNamespace()).Patch(a.GetName(), patchType, data)
+		if err != nil {
+			return "", "", err
+		}
+		return oldObj.ResourceVersion, newObj.ResourceVersion, nil
 	default:
 		panic("Unknown object")
 	}
@@ -144,6 +151,9 @@ func (u *updater) patchStatusUnstructured(obj runtime.Object, patchType types.Pa
 	case *v1.VirtualMachine:
 		_, err = u.cli.VirtualMachine(a.GetNamespace()).PatchStatus(a.GetName(), patchType, data)
 		return err
+	case *v1.KubeVirt:
+		_, err = u.cli.KubeVirt(a.GetNamespace()).PatchStatus(a.GetName(), patchType, data)
+		return err
 	default:
 		panic("Unknown object")
 	}
@@ -275,6 +285,10 @@ func (v *KVStatusUpdater) UpdateStatus(kv *v1.KubeVirt) error {
 	return v.updater.update(kv)
 }
 
+func (v *KVStatusUpdater) PatchStatus(kv *v1.KubeVirt, pt types.PatchType, data []byte) error {
+	return v.updater.patch(kv, pt, data)
+}
+
 func NewKubeVirtStatusUpdater(cli kubecli.KubevirtClient) *KVStatusUpdater {
 	return &KVStatusUpdater{
 		updater: updater{

pkg/virt-controller/services/template.go

@@ -96,6 +96,7 @@ type TemplateService interface {
 	RenderLaunchManifest(*v1.VirtualMachineInstance) (*k8sv1.Pod, error)
 	RenderHotplugAttachmentPodTemplate(volume *v1.Volume, ownerPod *k8sv1.Pod, vmi *v1.VirtualMachineInstance, pvcName string, isBlock, tempPod bool) (*k8sv1.Pod, error)
 	RenderLaunchManifestNoVm(*v1.VirtualMachineInstance) (*k8sv1.Pod, error)
+	GetLauncherImage() string
 }
 
 type templateService struct {
@@ -336,6 +337,11 @@ func requestResource(resources *k8sv1.ResourceRequirements, resourceName string)
 		resources.Requests[name] = unitQuantity
 	}
 }
+
+func (t *templateService) GetLauncherImage() string {
+	return t.launcherImage
+}
+
 func (t *templateService) RenderLaunchManifestNoVm(vmi *v1.VirtualMachineInstance) (*k8sv1.Pod, error) {
 	return t.renderLaunchManifest(vmi, true)
 }

pkg/virt-controller/watch/BUILD.bazel

@@ -31,6 +31,7 @@ go_library(
         "//pkg/virt-controller/watch/drain/disruptionbudget:go_default_library",
         "//pkg/virt-controller/watch/drain/evacuation:go_default_library",
         "//pkg/virt-controller/watch/snapshot:go_default_library",
+        "//pkg/virt-controller/watch/workload-updater:go_default_library",
         "//staging/src/kubevirt.io/client-go/api/v1:go_default_library",
         "//staging/src/kubevirt.io/client-go/apis/snapshot/v1alpha1:go_default_library",
         "//staging/src/kubevirt.io/client-go/kubecli:go_default_library",

pkg/virt-controller/watch/application.go

@@ -59,6 +59,7 @@ import (
 	"kubevirt.io/kubevirt/pkg/virt-controller/watch/drain/disruptionbudget"
 	"kubevirt.io/kubevirt/pkg/virt-controller/watch/drain/evacuation"
 	"kubevirt.io/kubevirt/pkg/virt-controller/watch/snapshot"
+	workloadupdater "kubevirt.io/kubevirt/pkg/virt-controller/watch/workload-updater"
 )
 
 const (
@@ -138,6 +139,8 @@ type VirtControllerApp struct {
 	migrationController *MigrationController
 	migrationInformer   cache.SharedIndexInformer
 
+	workloadUpdateController *workloadupdater.WorkloadUpdateController
+
 	snapshotController        *snapshot.VMSnapshotController
 	restoreController         *snapshot.VMRestoreController
 	vmSnapshotInformer        cache.SharedIndexInformer
@@ -297,6 +300,7 @@ func Execute() {
 	app.initEvacuationController()
 	app.initSnapshotController()
 	app.initRestoreController()
+	app.initWorkloadUpdaterController()
 	go app.Run()
 
 	select {
@@ -405,6 +409,7 @@ func (vca *VirtControllerApp) onStartedLeading() func(ctx context.Context) {
 		go vca.migrationController.Run(vca.migrationControllerThreads, stop)
 		go vca.snapshotController.Run(vca.snapshotControllerThreads, stop)
 		go vca.restoreController.Run(vca.restoreControllerThreads, stop)
+		go vca.workloadUpdateController.Run(stop)
 		cache.WaitForCacheSync(stop, vca.persistentVolumeClaimInformer.HasSynced)
 		close(vca.readyChan)
 		leaderGauge.Set(1)
@@ -473,6 +478,19 @@ func (vca *VirtControllerApp) initDisruptionBudgetController() {
 
 }
 
+func (vca *VirtControllerApp) initWorkloadUpdaterController() {
+	recorder := vca.getNewRecorder(k8sv1.NamespaceAll, "workload-update-controller")
+	vca.workloadUpdateController = workloadupdater.NewWorkloadUpdateController(
+		vca.launcherImage,
+		vca.vmiInformer,
+		vca.kvPodInformer,
+		vca.migrationInformer,
+		vca.kubeVirtInformer,
+		recorder,
+		vca.clientSet,
+		vca.clusterConfig)
+}
+
 func (vca *VirtControllerApp) initEvacuationController() {
 	recorder := vca.getNewRecorder(k8sv1.NamespaceAll, "disruptionbudget-controller")
 	vca.evacuationController = evacuation.NewEvacuationController(

pkg/virt-controller/watch/drain/evacuation/evacuation.go

@@ -320,11 +320,7 @@ func (c *EvacuationController) execute(key string) error {
 		return fmt.Errorf("failed to list VMIs on node: %v", err)
 	}
 
-	migrations, err := migrationutils.ListUnfinishedMigrations(c.migrationInformer)
-
-	if err != nil {
-		return fmt.Errorf("failed to list not finished migrations: %v", err)
-	}
+	migrations := migrationutils.ListUnfinishedMigrations(c.migrationInformer)
 
 	return c.sync(node, vmis, migrations)
 }

pkg/virt-controller/watch/migration.go

@@ -307,7 +307,7 @@ func (c *MigrationController) updateStatus(migration *virtv1.VirtualMachineInsta
 
 		migrationCopy.Status.Phase = virtv1.MigrationFailed
 		c.recorder.Eventf(migration, k8sv1.EventTypeWarning, FailedMigrationReason, "Source node reported migration failed")
-		log.Log.Object(migration).Error("VMI reported migration failed.")
+		log.Log.Object(migration).Errorf("VMI %s/%s reported migration failed.", vmi.Namespace, vmi.Name)
 	} else if migration.DeletionTimestamp != nil && !migration.IsFinal() &&
 		!conditionManager.HasCondition(migration, virtv1.VirtualMachineInstanceMigrationAbortRequested) {
 		condition := virtv1.VirtualMachineInstanceMigrationCondition{
@@ -856,11 +856,7 @@ func (c *MigrationController) outboundMigrationsOnNode(node string, runningMigra
 func (c *MigrationController) findRunningMigrations() ([]*virtv1.VirtualMachineInstanceMigration, error) {
 
 	// Don't start new migrations if we wait for migration object updates because of new target pods
-	notFinishedMigrations, err := migrations.ListUnfinishedMigrations(c.migrationInformer)
-	if err != nil {
-		return nil, err
-	}
-
+	notFinishedMigrations := migrations.ListUnfinishedMigrations(c.migrationInformer)
 	var runningMigrations []*virtv1.VirtualMachineInstanceMigration
 	for _, migration := range notFinishedMigrations {
 		if migration.IsRunning() {