65915631-9367-4366-bd57-e9b441563d36

Committed 07 May 2026 05:58AM UTC coverage: 71.783% (-0.005%) from 71.788%

Build # 65915631-9367-4366-bd57-e9b441563d36

Build Type

push

prow

Committed by

web-flow

Commit Message

Merge pull request #17354 from xpivarc/domainwatcher-refactor

Domainwatcher refactor

Coverage Stats

48 of 51 new or added lines in 3 files covered. (94.12%)

77673 of 108205 relevant lines covered (71.78%)

583.87 hits per line

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

82.64

/pkg/virt-handler/cache/cache.go

/*
 * This file is part of the KubeVirt project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * Copyright The KubeVirt Authors.
 *
 */

package cache

import (
        "context"
        "fmt"
        "os"
        "path/filepath"
        "sync"
        "time"

        "k8s.io/client-go/tools/record"

        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/apimachinery/pkg/watch"
        "k8s.io/client-go/tools/cache"

        "kubevirt.io/client-go/log"

        "kubevirt.io/kubevirt/pkg/checkpoint"
        "kubevirt.io/kubevirt/pkg/virt-launcher/virtwrap/api"

        notifyserver "kubevirt.io/kubevirt/pkg/virt-handler/notify-server"
)

type IterableCheckpointManager interface {
        ListKeys() []string
        checkpoint.CheckpointManager
}

type iterableCheckpointManager struct {
        base string
        checkpoint.CheckpointManager
}

func (icp *iterableCheckpointManager) ListKeys() []string {
        entries, err := os.ReadDir(icp.base)
        if err != nil {
                return []string{}
        }

        var keys []string
        for _, entry := range entries {
                keys = append(keys, entry.Name())
        }
        return keys

}

func NewIterableCheckpointManager(base string) IterableCheckpointManager {
        return &iterableCheckpointManager{
                base,
                checkpoint.NewSimpleCheckpointManager(base),
        }
}

type ghostRecord struct {
        Name       string    `json:"name"`
        Namespace  string    `json:"namespace"`
        SocketFile string    `json:"socketFile"`
        UID        types.UID `json:"uid"`
}

var GhostRecordGlobalStore GhostRecordStore

type GhostRecordStore struct {
        cache             map[string]ghostRecord
        checkpointManager checkpoint.CheckpointManager
        sync.Mutex
}

func InitializeGhostRecordCache(iterableCPManager IterableCheckpointManager) *GhostRecordStore {

        GhostRecordGlobalStore = GhostRecordStore{
                cache:             make(map[string]ghostRecord),
                checkpointManager: iterableCPManager,
        }

        keys := iterableCPManager.ListKeys()
        for _, key := range keys {
                ghostRecord := ghostRecord{}
                if err := GhostRecordGlobalStore.checkpointManager.Get(key, &ghostRecord); err != nil {
                        log.Log.Reason(err).Errorf("Unable to read ghost record checkpoint, %s", key)
                        continue
                }
                key := ghostRecord.Namespace + "/" + ghostRecord.Name
                GhostRecordGlobalStore.cache[key] = ghostRecord
                log.Log.Infof("Added ghost record for key %s", key)
        }
        return &GhostRecordGlobalStore
}

func (store *GhostRecordStore) LastKnownUID(key string) types.UID {
        store.Lock()
        defer store.Unlock()

        record, ok := store.cache[key]
        if !ok {
                return ""
        }

        return record.UID
}

func (store *GhostRecordStore) list() []ghostRecord {
        store.Lock()
        defer store.Unlock()

        var records []ghostRecord

        for _, record := range store.cache {
                records = append(records, record)
        }

        return records
}

func (store *GhostRecordStore) findBySocket(socketFile string) (ghostRecord, bool) {
        store.Lock()
        defer store.Unlock()

        for _, record := range store.cache {
                if filepath.Clean(record.SocketFile) == socketFile {
                        return record, true
                }
        }

        return ghostRecord{}, false
}

func (store *GhostRecordStore) Exists(namespace string, name string) bool {
        store.Lock()
        defer store.Unlock()

        key := namespace + "/" + name
        _, ok := store.cache[key]

        return ok
}

func (store *GhostRecordStore) Add(namespace string, name string, socketFile string, uid types.UID) (err error) {
        store.Lock()
        defer store.Unlock()
        if name == "" {
                return fmt.Errorf("can not add ghost record when 'name' is not provided")
        } else if namespace == "" {
                return fmt.Errorf("can not add ghost record when 'namespace' is not provided")
        } else if string(uid) == "" {
                return fmt.Errorf("unable to add ghost record with empty UID")
        } else if socketFile == "" {
                return fmt.Errorf("unable to add ghost record without a socketFile")
        }

        key := namespace + "/" + name
        record, ok := store.cache[key]
        if !ok {
                // record doesn't exist, so add new one.
                record := ghostRecord{
                        Name:       name,
                        Namespace:  namespace,
                        SocketFile: socketFile,
                        UID:        uid,
                }
                if err := store.checkpointManager.Store(string(uid), &record); err != nil {
                        return fmt.Errorf("failed to checkpoint %s, %w", uid, err)
                }
                store.cache[key] = record
        }

        // This protects us from stomping on a previous ghost record
        // that was not cleaned up properly. A ghost record that was
        // not deleted indicates that the VMI shutdown process did not
        // properly handle cleanup of local data.
        if ok && record.UID != uid {
                return fmt.Errorf("can not add ghost record when entry already exists with differing UID")
        }

        if ok && filepath.Clean(record.SocketFile) != socketFile {
                return fmt.Errorf("can not add ghost record when entry already exists with differing socket file location")
        }

        return nil
}

func (store *GhostRecordStore) Delete(namespace string, name string) error {
        store.Lock()
        defer store.Unlock()
        key := namespace + "/" + name
        record, ok := store.cache[key]
        if !ok {
                // already deleted
                return nil
        }

        if string(record.UID) == "" {
                return fmt.Errorf("unable to remove ghost record with empty UID")
        }

        if err := store.checkpointManager.Delete(string(record.UID)); err != nil {
                return fmt.Errorf("failed to delete checkpoint %s, %w", record.UID, err)
        }

        delete(store.cache, key)

        return nil
}

func NewSharedInformer(virtShareDir string, watchdogTimeout int, recorder record.EventRecorder, vmiStore cache.Store, resyncPeriod time.Duration) cache.SharedInformer {
        consecutiveFails := new(int)
        runServer := func(ctx context.Context, c chan watch.Event) error {
                return notifyserver.RunServer(virtShareDir, ctx.Done(), c, recorder, vmiStore)
        }
        lw := &cache.ListWatch{
                ListWithContextFunc: func(_ context.Context, _ metav1.ListOptions) (runtime.Object, error) {
                        log.Log.V(3).Info("Synchronizing domains")
                        domains, err := 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
                },
                WatchFuncWithContext: func(ctx context.Context, _ metav1.ListOptions) (watch.Interface, error) {
                        return newDomainWatcher(ctx, runServer, watchdogTimeout, resyncPeriod, recorder, consecutiveFails), nil
                },
        }
        return cache.NewSharedInformer(lw, &api.Domain{}, 0)
}

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 cache
21
22	import (
23	"context"
24	"fmt"
25	"os"
26	"path/filepath"
27	"sync"
28	"time"
29
30	"k8s.io/client-go/tools/record"
31
32	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
33	"k8s.io/apimachinery/pkg/runtime"
34	"k8s.io/apimachinery/pkg/types"
35	"k8s.io/apimachinery/pkg/watch"
36	"k8s.io/client-go/tools/cache"
37
38	"kubevirt.io/client-go/log"
39
40	"kubevirt.io/kubevirt/pkg/checkpoint"
41	"kubevirt.io/kubevirt/pkg/virt-launcher/virtwrap/api"
42
43	notifyserver "kubevirt.io/kubevirt/pkg/virt-handler/notify-server"
44	)
45
46	type IterableCheckpointManager interface {
47	ListKeys() []string
48	checkpoint.CheckpointManager
49	}
50
51	type iterableCheckpointManager struct {
52	base string
53	checkpoint.CheckpointManager
54	}
55
56	func (icp *iterableCheckpointManager) ListKeys() []string {	176✔
57	entries, err := os.ReadDir(icp.base)	176✔
58	if err != nil {	176✔
59	return []string{}	×
60	}	×
61
62	var keys []string	176✔
63	for _, entry := range entries {	180✔
64	keys = append(keys, entry.Name())	4✔
65	}	4✔
66	return keys	176✔
67
68	}
69
70	func NewIterableCheckpointManager(base string) IterableCheckpointManager {	176✔
71	return &iterableCheckpointManager{	176✔
72	base,	176✔
73	checkpoint.NewSimpleCheckpointManager(base),	176✔
74	}	176✔
75	}	176✔
76
77	type ghostRecord struct {
78	Name string `json:"name"`
79	Namespace string `json:"namespace"`
80	SocketFile string `json:"socketFile"`
81	UID types.UID `json:"uid"`
82	}
83
84	var GhostRecordGlobalStore GhostRecordStore
85
86	type GhostRecordStore struct {
87	cache map[string]ghostRecord
88	checkpointManager checkpoint.CheckpointManager
89	sync.Mutex
90	}
91
92	func InitializeGhostRecordCache(iterableCPManager IterableCheckpointManager) *GhostRecordStore {	175✔
93		175✔
94	GhostRecordGlobalStore = GhostRecordStore{	175✔
95	cache: make(map[string]ghostRecord),	175✔
96	checkpointManager: iterableCPManager,	175✔
97	}	175✔
98		175✔
99	keys := iterableCPManager.ListKeys()	175✔
100	for _, key := range keys {	177✔
101	ghostRecord := ghostRecord{}	2✔
102	if err := GhostRecordGlobalStore.checkpointManager.Get(key, &ghostRecord); err != nil {	2✔
103	log.Log.Reason(err).Errorf("Unable to read ghost record checkpoint, %s", key)	×
104	continue	×
105	}
106	key := ghostRecord.Namespace + "/" + ghostRecord.Name	2✔
107	GhostRecordGlobalStore.cache[key] = ghostRecord	2✔
108	log.Log.Infof("Added ghost record for key %s", key)	2✔
109	}
110	return &GhostRecordGlobalStore	175✔
111	}
112
113	func (store *GhostRecordStore) LastKnownUID(key string) types.UID {	2✔
114	store.Lock()	2✔
115	defer store.Unlock()	2✔
116		2✔
117	record, ok := store.cache[key]	2✔
118	if !ok {	3✔
119	return ""	1✔
120	}	1✔
121
122	return record.UID	1✔
123	}
124
125	func (store *GhostRecordStore) list() []ghostRecord {	17✔
126	store.Lock()	17✔
127	defer store.Unlock()	17✔
128		17✔
129	var records []ghostRecord	17✔
130		17✔
131	for _, record := range store.cache {	35✔
132	records = append(records, record)	18✔
133	}	18✔
134
135	return records	17✔
136	}
137
138	func (store *GhostRecordStore) findBySocket(socketFile string) (ghostRecord, bool) {	5✔
139	store.Lock()	5✔
140	defer store.Unlock()	5✔
141		5✔
142	for _, record := range store.cache {	11✔
143	if filepath.Clean(record.SocketFile) == socketFile {	10✔
144	return record, true	4✔
145	}	4✔
146	}
147
148	return ghostRecord{}, false	1✔
149	}
150
151	func (store *GhostRecordStore) Exists(namespace string, name string) bool {	×
152	store.Lock()	×
153	defer store.Unlock()	×
154		×
155	key := namespace + "/" + name	×
156	_, ok := store.cache[key]	×
157		×
158	return ok	×
159	}	×
160
161	func (store *GhostRecordStore) Add(namespace string, name string, socketFile string, uid types.UID) (err error) {	19✔
162	store.Lock()	19✔
163	defer store.Unlock()	19✔
164	if name == "" {	20✔
165	return fmt.Errorf("can not add ghost record when 'name' is not provided")	1✔
166	} else if namespace == "" {	20✔
167	return fmt.Errorf("can not add ghost record when 'namespace' is not provided")	1✔
168	} else if string(uid) == "" {	19✔
169	return fmt.Errorf("unable to add ghost record with empty UID")	1✔
170	} else if socketFile == "" {	18✔
171	return fmt.Errorf("unable to add ghost record without a socketFile")	1✔
172	}	1✔
173
174	key := namespace + "/" + name	15✔
175	record, ok := store.cache[key]	15✔
176	if !ok {	30✔
177	// record doesn't exist, so add new one.	15✔
178	record := ghostRecord{	15✔
179	Name: name,	15✔
180	Namespace: namespace,	15✔
181	SocketFile: socketFile,	15✔
182	UID: uid,	15✔
183	}	15✔
184	if err := store.checkpointManager.Store(string(uid), &record); err != nil {	15✔
185	return fmt.Errorf("failed to checkpoint %s, %w", uid, err)	×
186	}	×
187	store.cache[key] = record	15✔
188	}
189
190	// This protects us from stomping on a previous ghost record
191	// that was not cleaned up properly. A ghost record that was
192	// not deleted indicates that the VMI shutdown process did not
193	// properly handle cleanup of local data.
194	if ok && record.UID != uid {	15✔
195	return fmt.Errorf("can not add ghost record when entry already exists with differing UID")	×
196	}	×
197
198	if ok && filepath.Clean(record.SocketFile) != socketFile {	15✔
199	return fmt.Errorf("can not add ghost record when entry already exists with differing socket file location")	×
200	}	×
201
202	return nil	15✔
203	}
204
205	func (store *GhostRecordStore) Delete(namespace string, name string) error {	19✔
206	store.Lock()	19✔
207	defer store.Unlock()	19✔
208	key := namespace + "/" + name	19✔
209	record, ok := store.cache[key]	19✔
210	if !ok {	29✔
211	// already deleted	10✔
212	return nil	10✔
213	}	10✔
214
215	if string(record.UID) == "" {	9✔
216	return fmt.Errorf("unable to remove ghost record with empty UID")	×
217	}	×
218
219	if err := store.checkpointManager.Delete(string(record.UID)); err != nil {	9✔
220	return fmt.Errorf("failed to delete checkpoint %s, %w", record.UID, err)	×
221	}	×
222
223	delete(store.cache, key)	9✔
224		9✔
225	return nil	9✔
226	}
227
228	func NewSharedInformer(virtShareDir string, watchdogTimeout int, recorder record.EventRecorder, vmiStore cache.Store, resyncPeriod time.Duration) cache.SharedInformer {	8✔
229	consecutiveFails := new(int)	8✔
230	runServer := func(ctx context.Context, c chan watch.Event) error {	12✔
231	return notifyserver.RunServer(virtShareDir, ctx.Done(), c, recorder, vmiStore)	4✔
232	}	4✔
233	lw := &cache.ListWatch{	8✔
234	ListWithContextFunc: func(_ context.Context, _ metav1.ListOptions) (runtime.Object, error) {	12✔
235	log.Log.V(3).Info("Synchronizing domains")	4✔
236	domains, err := listAllKnownDomains()	4✔
237	if err != nil {	4✔
NEW 238	return nil, err	×
NEW 239	}	×
240	list := api.DomainList{	4✔
241	Items: []api.Domain{},	4✔
242	}	4✔
243	for _, domain := range domains {	8✔
244	list.Items = append(list.Items, *domain)	4✔
245	}	4✔
246	return &list, nil	4✔
247	},
248	WatchFuncWithContext: func(ctx context.Context, _ metav1.ListOptions) (watch.Interface, error) {	4✔
249	return newDomainWatcher(ctx, runServer, watchdogTimeout, resyncPeriod, recorder, consecutiveFails), nil	4✔
250	},	4✔
251	}
252	return cache.NewSharedInformer(lw, &api.Domain{}, 0)	8✔
253	}

kubevirt / kubevirt / 65915631-9367-4366-bd57-e9b441563d36

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

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