714a6919-5213-4f7a-a6a4-0ffce1f1c8ea

Committed 25 Feb 2025 01:54AM UTC coverage: 71.574% (+0.002%) from 71.572%

Build # 714a6919-5213-4f7a-a6a4-0ffce1f1c8ea

Build Type

push

prow

Committed by

web-flow

Commit Message

Merge pull request #13816 from xpivarc/ghostRecordsStruct

Refactor Ghost records

Coverage Stats

63 of 66 new or added lines in 3 files covered. (95.45%)

2 existing lines in 1 file now uncovered.

61846 of 86409 relevant lines covered (71.57%)

0.8 hits per line

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

88.8

/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 2017 Red Hat, Inc.
 *
 */

package cache

import (
        "fmt"
        "os"
        "sync"
        "time"

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

        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/cache"

        "kubevirt.io/client-go/log"

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

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{}
        }

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

kubevirt / kubevirt / 714a6919-5213-4f7a-a6a4-0ffce1f1c8ea

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