• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

kubevirt / kubevirt / b9cf39a6-2db5-4abb-a9e0-298ab865ee3f

26 Jun 2025 02:54PM UTC coverage: 70.492% (-0.2%) from 70.726%
b9cf39a6-2db5-4abb-a9e0-298ab865ee3f

push

prow

web-flow
Merge pull request #15013 from kubevirt-bot/autoupdate

Run go run ./robots/cmd/uploader -workspace /home/prow/go/src/github.com/kubevirt/project-infra/../kubevirt/WORKSPACE -dry-run=false

66894 of 94896 relevant lines covered (70.49%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.96
/pkg/virt-handler/migration-source.go
1
/*
2
 * This file is part of the KubeVirt project
3
 *
4
 * Licensed under the Apache License, Version 2.0 (the "License");
5
 * you may not use this file except in compliance with the License.
6
 * You may obtain a copy of the License at
7
 *
8
 *     http://www.apache.org/licenses/LICENSE-2.0
9
 *
10
 * Unless required by applicable law or agreed to in writing, software
11
 * distributed under the License is distributed on an "AS IS" BASIS,
12
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13
 * See the License for the specific language governing permissions and
14
 * limitations under the License.
15
 *
16
 * Copyright The KubeVirt Authors.
17
 *
18
 */
19

20
package virthandler
21

22
import (
23
        "context"
24
        "encoding/json"
25
        "errors"
26
        "fmt"
27
        "path/filepath"
28
        "time"
29

30
        "libvirt.org/go/libvirtxml"
31

32
        k8sv1 "k8s.io/api/core/v1"
33
        "k8s.io/apimachinery/pkg/api/equality"
34
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
35
        "k8s.io/apimachinery/pkg/util/wait"
36
        "k8s.io/client-go/tools/cache"
37
        "k8s.io/client-go/tools/record"
38
        "k8s.io/client-go/util/workqueue"
39

40
        v1 "kubevirt.io/api/core/v1"
41
        "kubevirt.io/client-go/kubecli"
42
        "kubevirt.io/client-go/log"
43

44
        "kubevirt.io/kubevirt/pkg/controller"
45
        hostdisk "kubevirt.io/kubevirt/pkg/host-disk"
46
        "kubevirt.io/kubevirt/pkg/pointer"
47
        "kubevirt.io/kubevirt/pkg/util/migrations"
48
        virtconfig "kubevirt.io/kubevirt/pkg/virt-config"
49
        cmdclient "kubevirt.io/kubevirt/pkg/virt-handler/cmd-client"
50
        "kubevirt.io/kubevirt/pkg/virt-handler/isolation"
51

52
        launcher_clients "kubevirt.io/kubevirt/pkg/virt-handler/launcher-clients"
53
        migrationproxy "kubevirt.io/kubevirt/pkg/virt-handler/migration-proxy"
54
        "kubevirt.io/kubevirt/pkg/virt-launcher/virtwrap/api"
55
)
56

57
var errWaitingForTargetPorts = errors.New("waiting for target to publish migration ports")
58

59
type passtRepairSourceHandler interface {
60
        HandleMigrationSource(*v1.VirtualMachineInstance, func(*v1.VirtualMachineInstance) (string, error)) error
61
}
62

63
type MigrationSourceController struct {
64
        *BaseController
65
        capabilities                *libvirtxml.Caps
66
        clientset                   kubecli.KubevirtClient
67
        queue                       workqueue.TypedRateLimitingInterface[string]
68
        launcherClients             launcher_clients.LauncherClientsManager
69
        migrationProxy              migrationproxy.ProxyManager
70
        podIsolationDetector        isolation.PodIsolationDetector
71
        recorder                    record.EventRecorder
72
        virtLauncherFSRunDirPattern string
73
        vmiExpectations             *controller.UIDTrackingControllerExpectations
74
        passtRepairHandler          passtRepairSourceHandler
75
}
76

77
func NewMigrationSourceController(
78
        recorder record.EventRecorder,
79
        clientset kubecli.KubevirtClient,
80
        host string,
81
        launcherClients launcher_clients.LauncherClientsManager,
82
        vmiInformer cache.SharedIndexInformer,
83
        domainInformer cache.SharedInformer,
84
        clusterConfig *virtconfig.ClusterConfig,
85
        podIsolationDetector isolation.PodIsolationDetector,
86
        migrationProxy migrationproxy.ProxyManager,
87
        virtLauncherFSRunDirPattern string,
88
        passtRepairHandler passtRepairSourceHandler,
89
) (*MigrationSourceController, error) {
1✔
90

1✔
91
        baseCtrl, err := NewBaseController(
1✔
92
                host,
1✔
93
                vmiInformer,
1✔
94
                domainInformer,
1✔
95
                clusterConfig,
1✔
96
                podIsolationDetector,
1✔
97
        )
1✔
98
        if err != nil {
1✔
99
                return nil, err
×
100
        }
×
101

102
        queue := workqueue.NewTypedRateLimitingQueueWithConfig[string](
1✔
103
                workqueue.DefaultTypedControllerRateLimiter[string](),
1✔
104
                workqueue.TypedRateLimitingQueueConfig[string]{Name: "virt-handler-source"},
1✔
105
        )
1✔
106

1✔
107
        c := &MigrationSourceController{
1✔
108
                BaseController:              baseCtrl,
1✔
109
                clientset:                   clientset,
1✔
110
                queue:                       queue,
1✔
111
                launcherClients:             launcherClients,
1✔
112
                migrationProxy:              migrationProxy,
1✔
113
                podIsolationDetector:        podIsolationDetector,
1✔
114
                recorder:                    recorder,
1✔
115
                virtLauncherFSRunDirPattern: virtLauncherFSRunDirPattern,
1✔
116
                vmiExpectations:             controller.NewUIDTrackingControllerExpectations(controller.NewControllerExpectations()),
1✔
117
                passtRepairHandler:          passtRepairHandler,
1✔
118
        }
1✔
119

1✔
120
        _, err = vmiInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
1✔
121
                AddFunc:    c.addFunc,
1✔
122
                DeleteFunc: c.deleteFunc,
1✔
123
                UpdateFunc: c.updateFunc,
1✔
124
        })
1✔
125
        if err != nil {
1✔
126
                return nil, err
×
127
        }
×
128

129
        _, err = domainInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
1✔
130
                AddFunc:    c.addDeleteDomainFunc,
1✔
131
                DeleteFunc: c.addDeleteDomainFunc,
1✔
132
                UpdateFunc: c.updateDomainFunc,
1✔
133
        })
1✔
134
        if err != nil {
1✔
135
                return nil, err
×
136
        }
×
137

138
        return c, nil
1✔
139
}
140

141
func (c *MigrationSourceController) hasTargetDetectedReadyDomain(vmi *v1.VirtualMachineInstance) (bool, int64) {
1✔
142
        // give the target node 60 seconds to discover the libvirt domain via the domain informer
1✔
143
        // before allowing the VMI to be processed. This closes the gap between the
1✔
144
        // VMI's status getting updated to reflect the new source node, and the domain
1✔
145
        // informer firing the event to alert the source node of the new domain.
1✔
146
        migrationTargetDelayTimeout := 60
1✔
147

1✔
148
        if vmi.Status.MigrationState == nil ||
1✔
149
                vmi.Status.MigrationState.EndTimestamp == nil {
2✔
150
                return false, int64(migrationTargetDelayTimeout)
1✔
151
        }
1✔
152
        if vmi.Status.MigrationState != nil &&
×
153
                vmi.Status.MigrationState.TargetState != nil &&
×
154
                vmi.Status.MigrationState.TargetState.DomainDetected &&
×
155
                vmi.Status.MigrationState.TargetState.DomainReadyTimestamp != nil {
×
156

×
157
                return true, 0
×
158
        }
×
159

160
        nowUnix := time.Now().UTC().Unix()
×
161
        migrationEndUnix := vmi.Status.MigrationState.EndTimestamp.Time.UTC().Unix()
×
162

×
163
        diff := nowUnix - migrationEndUnix
×
164

×
165
        if diff > int64(migrationTargetDelayTimeout) {
×
166
                return false, 0
×
167
        }
×
168

169
        timeLeft := int64(migrationTargetDelayTimeout) - diff
×
170

×
171
        enqueueTime := timeLeft
×
172
        if enqueueTime < 5 {
×
173
                enqueueTime = 5
×
174
        }
×
175

176
        // re-enqueue the key to ensure it gets processed again within the right time.
177
        c.queue.AddAfter(controller.VirtualMachineInstanceKey(vmi), time.Duration(enqueueTime)*time.Second)
×
178

×
179
        return false, timeLeft
×
180
}
181

182
func domainMigrated(domain *api.Domain) bool {
1✔
183
        return domain != nil && domain.Status.Status == api.Shutoff && domain.Status.Reason == api.ReasonMigrated
1✔
184
}
1✔
185

186
func (c *MigrationSourceController) setMigrationProgressStatus(vmi *v1.VirtualMachineInstance, domain *api.Domain) {
1✔
187
        if domain == nil ||
1✔
188
                domain.Spec.Metadata.KubeVirt.Migration == nil ||
1✔
189
                vmi.Status.MigrationState == nil ||
1✔
190
                !c.isMigrationSource(vmi) {
2✔
191
                return
1✔
192
        }
1✔
193

194
        migrationMetadata := domain.Spec.Metadata.KubeVirt.Migration
1✔
195
        if migrationMetadata.UID != vmi.Status.MigrationState.MigrationUID {
2✔
196
                return
1✔
197
        }
1✔
198
        vmi.Status.MigrationState.StartTimestamp = migrationMetadata.StartTimestamp
1✔
199

1✔
200
        vmi.Status.MigrationState.Failed = migrationMetadata.Failed
1✔
201

1✔
202
        if migrationMetadata.Failed {
2✔
203
                vmi.Status.MigrationState.EndTimestamp = migrationMetadata.EndTimestamp
1✔
204
                vmi.Status.MigrationState.FailureReason = migrationMetadata.FailureReason
1✔
205
                c.recorder.Event(vmi, k8sv1.EventTypeWarning, v1.Migrated.String(), fmt.Sprintf("VirtualMachineInstance migration uid %s failed. reason:%s", string(migrationMetadata.UID), migrationMetadata.FailureReason))
1✔
206
        }
1✔
207

208
        vmi.Status.MigrationState.AbortStatus = v1.MigrationAbortStatus(migrationMetadata.AbortStatus)
1✔
209
        if migrationMetadata.AbortStatus == string(v1.MigrationAbortSucceeded) {
1✔
210
                vmi.Status.MigrationState.EndTimestamp = migrationMetadata.EndTimestamp
×
211
        }
×
212

213
        vmi.Status.MigrationState.Mode = migrationMetadata.Mode
1✔
214
}
215

216
func (c *MigrationSourceController) updateStatus(vmi *v1.VirtualMachineInstance, domain *api.Domain) error {
1✔
217
        c.setMigrationProgressStatus(vmi, domain)
1✔
218

1✔
219
        // handle migrations differently than normal status updates.
1✔
220
        //
1✔
221
        // When a successful migration is detected, we must transfer ownership of the VMI
1✔
222
        // from the source node (this node) to the target node (node the domain was migrated to).
1✔
223
        //
1✔
224
        // Transfer ownership by...
1✔
225
        // 1. Marking vmi.Status.MigrationState as completed
1✔
226
        // 2. Update the vmi.Status.NodeName to reflect the target node's name
1✔
227
        // 3. Update the VMI's NodeNameLabel annotation to reflect the target node's name
1✔
228
        // 4. Clear the LauncherContainerImageVersion which virt-controller will detect
1✔
229
        //    and accurately based on the version used on the target pod
1✔
230
        //
1✔
231
        // After a migration, the VMI's phase is no longer owned by this node. Only the
1✔
232
        // MigrationState status field is eligible to be mutated.
1✔
233
        migrationHost := ""
1✔
234
        if vmi.Status.MigrationState != nil {
2✔
235
                migrationHost = vmi.Status.MigrationState.TargetNode
1✔
236
        }
1✔
237

238
        targetNodeDetectedDomain, timeLeft := c.hasTargetDetectedReadyDomain(vmi)
1✔
239
        // If we can't detect where the migration went to, then we have no
1✔
240
        // way of transferring ownership. The only option here is to move the
1✔
241
        // vmi to failed.  The cluster vmi controller will then tear down the
1✔
242
        // resulting pods.
1✔
243
        if migrationHost == "" {
1✔
244
                // migrated to unknown host.
×
245
                vmi.Status.Phase = v1.Failed
×
246
                vmi.Status.MigrationState.Completed = true
×
247
                vmi.Status.MigrationState.Failed = true
×
248

×
249
                log.Log.Object(vmi).Warning("the vmi migrated to an unknown host")
×
250
                c.recorder.Event(vmi, k8sv1.EventTypeWarning, v1.Migrated.String(), fmt.Sprintf("The VirtualMachineInstance migrated to unknown host."))
×
251
        } else if !targetNodeDetectedDomain {
2✔
252
                if timeLeft <= 0 {
1✔
253
                        vmi.Status.Phase = v1.Failed
×
254
                        vmi.Status.MigrationState.Completed = true
×
255
                        vmi.Status.MigrationState.Failed = true
×
256

×
257
                        log.Log.Object(vmi).Warning("the domain was never observed on the taget after the migration completed within the timeout period")
×
258
                        c.recorder.Event(vmi, k8sv1.EventTypeWarning, v1.Migrated.String(), fmt.Sprintf("The VirtualMachineInstance's domain was never observed on the target after the migration completed within the timeout period."))
×
259
                }
×
260
        }
261

262
        if targetNodeDetectedDomain && vmi.IsDecentralizedMigration() && vmi.Status.MigrationState != nil && vmi.Status.MigrationState.Completed {
1✔
263
                log.Log.Object(vmi).V(2).Infof("decentralized migration completed successfully, marking VMI as succeeded")
×
264
                // this is a decentralized migration, and the migration completed successfully, we need to mark the VMI as succeeded
×
265
                vmi.Status.Phase = v1.Succeeded
×
266
        }
×
267

268
        return nil
1✔
269
}
270

271
func (c *MigrationSourceController) Run(threadiness int, stopCh chan struct{}) {
×
272
        defer c.queue.ShutDown()
×
273
        log.Log.Info("Starting virt-handler source controller.")
×
274

×
275
        cache.WaitForCacheSync(stopCh, c.hasSynced)
×
276

×
277
        // queue keys for previous Domains on the host that no longer exist
×
278
        // in the cache. This ensures we perform local cleanup of deleted VMs.
×
279
        for _, domain := range c.domainStore.List() {
×
280
                d := domain.(*api.Domain)
×
281
                vmiRef := v1.NewVMIReferenceWithUUID(
×
282
                        d.ObjectMeta.Namespace,
×
283
                        d.ObjectMeta.Name,
×
284
                        d.Spec.Metadata.KubeVirt.UID)
×
285

×
286
                key := controller.VirtualMachineInstanceKey(vmiRef)
×
287

×
288
                _, exists, _ := c.vmiStore.GetByKey(key)
×
289
                if !exists {
×
290
                        c.queue.Add(key)
×
291
                }
×
292
        }
293

294
        // Start the actual work
295
        for i := 0; i < threadiness; i++ {
×
296
                go wait.Until(c.runWorker, time.Second, stopCh)
×
297
        }
×
298

299
        <-stopCh
×
300
        log.Log.Info("Stopping virt-handler source controller.")
×
301
}
302

303
func (c *MigrationSourceController) runWorker() {
×
304
        for c.Execute() {
×
305
        }
×
306
}
307

308
func (c *MigrationSourceController) Execute() bool {
1✔
309
        key, quit := c.queue.Get()
1✔
310
        if quit {
1✔
311
                return false
×
312
        }
×
313
        defer c.queue.Done(key)
1✔
314
        if err := c.execute(key); err != nil {
1✔
315
                log.Log.Reason(err).Infof("re-enqueuing VirtualMachineInstance %v", key)
×
316
                c.queue.AddRateLimited(key)
×
317
        } else {
1✔
318
                log.Log.V(4).Infof("processed VirtualMachineInstance %v", key)
1✔
319
                c.queue.Forget(key)
1✔
320
        }
1✔
321
        return true
1✔
322
}
323

324
func (c *MigrationSourceController) sync(vmi *v1.VirtualMachineInstance, domain *api.Domain) error {
1✔
325
        if domain != nil {
2✔
326
                log.Log.Object(vmi).Infof("VMI is in phase: %v | Domain status: %v, reason: %v", vmi.Status.Phase, domain.Status.Status, domain.Status.Reason)
1✔
327
        } else {
1✔
328
                log.Log.Object(vmi).Infof("VMI is in phase: %v", vmi.Status.Phase)
×
329
        }
×
330

331
        oldStatus := vmi.Status.DeepCopy()
1✔
332

1✔
333
        syncErr := c.processVMI(vmi, domain)
1✔
334

1✔
335
        if syncErr != nil {
1✔
336
                c.recorder.Event(vmi, k8sv1.EventTypeWarning, v1.SyncFailed.String(), syncErr.Error())
×
337
                // `syncErr` will be propagated anyway, and it will be logged in `re-enqueueing`
×
338
                // so there is no need to log it twice in hot path without increased verbosity.
×
339
                log.Log.Object(vmi).Reason(syncErr).Error("Synchronizing the VirtualMachineInstance failed.")
×
340
        }
×
341

342
        updateErr := c.updateStatus(vmi, domain)
1✔
343

1✔
344
        if updateErr != nil {
1✔
345
                log.Log.Object(vmi).Reason(updateErr).Error("Updating the migration status failed.")
×
346
        }
×
347

348
        // update the VMI if necessary
349
        if !equality.Semantic.DeepEqual(*oldStatus, vmi.Status) {
2✔
350
                key := controller.VirtualMachineInstanceKey(vmi)
1✔
351
                c.vmiExpectations.SetExpectations(key, 1, 0)
1✔
352
                _, err := c.clientset.VirtualMachineInstance(vmi.ObjectMeta.Namespace).Update(context.Background(), vmi, metav1.UpdateOptions{})
1✔
353
                if err != nil {
1✔
354
                        c.vmiExpectations.SetExpectations(key, 0, 0)
×
355
                        return err
×
356
                }
×
357
        }
358

359
        if syncErr != nil {
1✔
360
                return syncErr
×
361
        }
×
362

363
        if updateErr != nil {
1✔
364
                return updateErr
×
365
        }
×
366

367
        log.Log.Object(vmi).V(4).Info("Source synchronization loop succeeded.")
1✔
368
        return nil
1✔
369

370
}
371

372
func (c *MigrationSourceController) execute(key string) error {
1✔
373
        vmi, vmiExists, err := c.getVMIFromCache(key)
1✔
374
        if err != nil {
1✔
375
                return err
×
376
        }
×
377

378
        if !vmiExists || vmi.IsFinal() || vmi.DeletionTimestamp != nil {
1✔
379
                log.Log.V(4).Infof("vmi for key %v is terminating, final or does not exists", key)
×
380
                return nil
×
381
        }
×
382

383
        if !c.vmiExpectations.SatisfiedExpectations(key) {
1✔
384
                log.Log.V(4).Object(vmi).Info("waiting for expectations to be satisfied")
×
385
                return nil
×
386
        }
×
387

388
        domain, domainExists, _, err := c.getDomainFromCache(key)
1✔
389
        if err != nil {
1✔
390
                return err
×
391
        }
×
392

393
        if domainExists && domain.Spec.Metadata.KubeVirt.UID != vmi.UID {
1✔
394
                log.Log.V(4).Object(vmi).Infof("Detected stale vmi %s that still needs cleanup before new vmi with identical name/namespace can be processed", vmi.UID)
×
395
                return nil
×
396
        }
×
397

398
        if vmi.Status.MigrationState == nil {
1✔
399
                log.Log.V(4).Object(vmi).Info("no migration is in progress")
×
400
                return nil
×
401
        }
×
402

403
        // post migration clean up
404
        if isMigrationDone(vmi.Status.MigrationState) {
1✔
405
                c.migrationProxy.StopSourceListener(string(vmi.UID))
×
406
                return nil
×
407
        }
×
408

409
        if !c.isMigrationSource(vmi) {
1✔
410
                log.Log.Object(vmi).V(4).Info("not a migration source")
×
411
                return nil
×
412
        }
×
413

414
        return c.sync(vmi.DeepCopy(), domain)
1✔
415
}
416

417
func (c *MigrationSourceController) isMigrationSource(vmi *v1.VirtualMachineInstance) bool {
1✔
418
        return vmi.Status.MigrationState != nil &&
1✔
419
                vmi.Status.MigrationState.SourceNode == c.host &&
1✔
420
                (!vmi.IsDecentralizedMigration() || vmi.IsMigrationSource()) &&
1✔
421
                vmi.Status.MigrationState.TargetNodeAddress != "" &&
1✔
422
                !vmi.Status.MigrationState.Completed
1✔
423
}
1✔
424

425
func (c *MigrationSourceController) handleSourceMigrationProxy(vmi *v1.VirtualMachineInstance) error {
1✔
426

1✔
427
        res, err := c.podIsolationDetector.Detect(vmi)
1✔
428
        if err != nil {
1✔
429
                return err
×
430
        }
×
431
        // the migration-proxy is no longer shared via host mount, so we
432
        // pass in the virt-launcher's baseDir to reach the unix sockets.
433
        baseDir := fmt.Sprintf(filepath.Join(c.virtLauncherFSRunDirPattern, "kubevirt"), res.Pid())
1✔
434
        if vmi.Status.MigrationState.TargetDirectMigrationNodePorts == nil {
1✔
435
                return errWaitingForTargetPorts
×
436
        }
×
437

438
        err = c.migrationProxy.StartSourceListener(
1✔
439
                string(vmi.UID),
1✔
440
                vmi.Status.MigrationState.TargetNodeAddress,
1✔
441
                vmi.Status.MigrationState.TargetDirectMigrationNodePorts,
1✔
442
                baseDir,
1✔
443
        )
1✔
444
        if err != nil {
1✔
445
                return err
×
446
        }
×
447

448
        return nil
1✔
449
}
450

451
func (c *MigrationSourceController) migrateVMI(vmi *v1.VirtualMachineInstance, domain *api.Domain) error {
1✔
452
        isUnresponsive, isInitialized, err := c.launcherClients.IsLauncherClientUnresponsive(vmi)
1✔
453
        if err != nil {
1✔
454
                return err
×
455
        }
×
456
        if !isInitialized {
1✔
457
                log.Log.Object(vmi).V(4).Info("launcher client is not initialized")
×
458
                c.queue.AddAfter(controller.VirtualMachineInstanceKey(vmi), time.Second*1)
×
459
                return nil
×
460
        } else if isUnresponsive {
1✔
461
                return errors.New(fmt.Sprintf("Can not update a VirtualMachineInstance with unresponsive command server."))
×
462
        }
×
463

464
        client, err := c.launcherClients.GetLauncherClient(vmi)
1✔
465
        if err != nil {
1✔
466
                return fmt.Errorf(unableCreateVirtLauncherConnectionFmt, err)
×
467
        }
×
468

469
        if vmi.Status.MigrationState.AbortRequested {
2✔
470
                err = c.handleMigrationAbort(vmi, client)
1✔
471
                return err
1✔
472
        }
1✔
473

474
        if isMigrationInProgress(vmi, domain) {
1✔
475
                // we already started this migration, no need to rerun this
×
476
                log.Log.Object(vmi).V(4).Infof("migration %s has already been started", vmi.Status.MigrationState.MigrationUID)
×
477
                return nil
×
478
        }
×
479

480
        err = c.handleSourceMigrationProxy(vmi)
1✔
481
        if errors.Is(err, errWaitingForTargetPorts) {
1✔
482
                log.Log.Object(vmi).V(4).Info("waiting for target node to publish migration ports")
×
483
                c.queue.AddAfter(controller.VirtualMachineInstanceKey(vmi), 1*time.Second)
×
484
                return nil
×
485
        } else if err != nil {
1✔
486
                return fmt.Errorf("failed to handle migration proxy: %v", err)
×
487
        }
×
488

489
        migrationConfiguration := vmi.Status.MigrationState.MigrationConfiguration
1✔
490
        if migrationConfiguration == nil {
2✔
491
                migrationConfiguration = c.clusterConfig.GetMigrationConfiguration()
1✔
492
        }
1✔
493

494
        options := &cmdclient.MigrationOptions{
1✔
495
                Bandwidth:               *migrationConfiguration.BandwidthPerMigration,
1✔
496
                ProgressTimeout:         *migrationConfiguration.ProgressTimeout,
1✔
497
                CompletionTimeoutPerGiB: *migrationConfiguration.CompletionTimeoutPerGiB,
1✔
498
                UnsafeMigration:         *migrationConfiguration.UnsafeMigrationOverride,
1✔
499
                AllowAutoConverge:       *migrationConfiguration.AllowAutoConverge,
1✔
500
                AllowPostCopy:           *migrationConfiguration.AllowPostCopy,
1✔
501
                AllowWorkloadDisruption: *migrationConfiguration.AllowWorkloadDisruption,
1✔
502
        }
1✔
503

1✔
504
        configureParallelMigrationThreads(options, vmi)
1✔
505

1✔
506
        marshalledOptions, err := json.Marshal(options)
1✔
507
        if err != nil {
1✔
508
                log.Log.Object(vmi).Warning("failed to marshall matched migration options")
×
509
        } else {
1✔
510
                log.Log.Object(vmi).Infof("migration options matched for vmi %s: %s", vmi.Name, string(marshalledOptions))
1✔
511
        }
1✔
512

513
        vmiCopy := vmi.DeepCopy()
1✔
514
        err = hostdisk.ReplacePVCByHostDisk(vmiCopy)
1✔
515
        if err != nil {
1✔
516
                return err
×
517
        }
×
518

519
        if c.clusterConfig.PasstIPStackMigrationEnabled() {
1✔
520
                if err := c.passtRepairHandler.HandleMigrationSource(vmi, c.passtSocketDirOnHostMigrationSource); err != nil {
×
521
                        log.Log.Object(vmi).Warningf("failed to call passt-repair for migration source, %v", err)
×
522
                }
×
523
        }
524

525
        err = client.MigrateVirtualMachine(vmiCopy, options)
1✔
526
        if err != nil {
1✔
527
                return err
×
528
        }
×
529
        c.recorder.Event(vmi, k8sv1.EventTypeNormal, v1.Migrating.String(), VMIMigrating)
1✔
530
        return nil
1✔
531
}
532

533
func (c *MigrationSourceController) passtSocketDirOnHostMigrationSource(vmi *v1.VirtualMachineInstance) (string, error) {
×
534
        path, err := c.podIsolationDetector.Detect(vmi)
×
535
        if err != nil {
×
536
                return "", err
×
537
        }
×
538
        return passtSocketDirOnHost(path)
×
539
}
540

541
func isMigrationDone(state *v1.VirtualMachineInstanceMigrationState) bool {
1✔
542
        return state == nil || (state.EndTimestamp != nil && (state.Completed || state.Failed))
1✔
543
}
1✔
544

545
func (c *MigrationSourceController) processVMI(vmi *v1.VirtualMachineInstance, domain *api.Domain) error {
1✔
546
        domainAlive := domain != nil &&
1✔
547
                domain.Status.Status != api.Shutoff &&
1✔
548
                domain.Status.Status != api.Crashed &&
1✔
549
                domain.Status.Status != ""
1✔
550

1✔
551
        if !domainAlive {
1✔
552
                log.Log.V(4).Object(vmi).Info("domain is not alive")
×
553
                return nil
×
554
        }
×
555

556
        return c.migrateVMI(vmi, domain)
1✔
557
}
558

559
func (c *MigrationSourceController) addFunc(obj interface{}) {
×
560
        key, err := controller.KeyFunc(obj)
×
561
        if err == nil {
×
562
                c.vmiExpectations.SetExpectations(key, 0, 0)
×
563
                c.queue.Add(key)
×
564
        }
×
565
}
566

567
func (c *MigrationSourceController) deleteFunc(obj interface{}) {
×
568
        key, err := controller.KeyFunc(obj)
×
569
        if err == nil {
×
570
                c.queue.Add(key)
×
571
        }
×
572
}
573

574
func (c *MigrationSourceController) updateFunc(_, new interface{}) {
×
575
        key, err := controller.KeyFunc(new)
×
576
        if err == nil {
×
577
                c.vmiExpectations.SetExpectations(key, 0, 0)
×
578
                c.queue.Add(key)
×
579
        }
×
580
}
581

582
func (c *MigrationSourceController) addDeleteDomainFunc(obj interface{}) {
×
583
        key, err := controller.KeyFunc(obj)
×
584
        if err == nil {
×
585
                c.queue.Add(key)
×
586
        }
×
587
}
588

589
func (c *MigrationSourceController) updateDomainFunc(_, new interface{}) {
×
590
        key, err := controller.KeyFunc(new)
×
591
        if err == nil {
×
592
                c.queue.Add(key)
×
593
        }
×
594
}
595

596
func (c *MigrationSourceController) handleMigrationAbort(vmi *v1.VirtualMachineInstance, client cmdclient.LauncherClient) error {
1✔
597
        if vmi.Status.MigrationState.AbortStatus == v1.MigrationAbortInProgress || vmi.Status.MigrationState.AbortStatus == v1.MigrationAbortSucceeded {
2✔
598
                return nil
1✔
599
        }
1✔
600

601
        if err := client.CancelVirtualMachineMigration(vmi); err != nil {
2✔
602
                if err.Error() == migrations.CancelMigrationFailedVmiNotMigratingErr {
1✔
603
                        // If migration did not even start there is no need to cancel it
×
604
                        log.Log.Object(vmi).Infof("skipping migration cancellation since vmi is not migrating")
×
605
                }
×
606
                return err
1✔
607
        }
608

609
        c.recorder.Event(vmi, k8sv1.EventTypeNormal, v1.Migrating.String(), VMIAbortingMigration)
1✔
610
        return nil
1✔
611
}
612

613
func configureParallelMigrationThreads(options *cmdclient.MigrationOptions, vm *v1.VirtualMachineInstance) {
1✔
614
        // When the CPU is limited, there's a risk of the migration threads choking the CPU resources on the compute container.
1✔
615
        // For this reason, we will avoid configuring migration threads in such scenarios.
1✔
616
        if cpuLimit, cpuLimitExists := vm.Spec.Domain.Resources.Limits[k8sv1.ResourceCPU]; cpuLimitExists && !cpuLimit.IsZero() {
2✔
617
                return
1✔
618
        }
1✔
619

620
        options.ParallelMigrationThreads = pointer.P(parallelMultifdMigrationThreads)
1✔
621
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc