Skip to content

Commit 968218a

Browse files
skogtaskogta
committed
Add CnsVolumeAttachment CRD
1 parent f03e1fe commit 968218a

File tree

9 files changed

+763
-0
lines changed

9 files changed

+763
-0
lines changed

‎pkg/internalapis/cnsoperator/cnsvolumeattachment/cnsvolumeattachment.go

Lines changed: 343 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,343 @@
1+
/*
2+
Copyright 2025 The Kubernetes Authors.
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+
17+
package cnsvolumeattachment
18+
19+
import (
20+
"context"
21+
"fmt"
22+
"sync"
23+
24+
"k8s.io/apimachinery/pkg/api/errors"
25+
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
26+
"k8s.io/apimachinery/pkg/types"
27+
"k8s.io/client-go/tools/cache"
28+
"sigs.k8s.io/controller-runtime/pkg/client"
29+
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
30+
k8s "sigs.k8s.io/vsphere-csi-driver/v3/pkg/kubernetes"
31+
32+
"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/logger"
33+
"sigs.k8s.io/vsphere-csi-driver/v3/pkg/internalapis"
34+
"sigs.k8s.io/vsphere-csi-driver/v3/pkg/internalapis/cnsoperator/cnsvolumeattachment/v1alpha1"
35+
cnsoperatortypes "sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/types"
36+
)
37+
38+
var (
39+
cnsVolumeAttachmentInstanceLock sync.Mutex
40+
cnsVolumeAttachmentInstance *cnsVolumeAttachment
41+
)
42+
43+
// CnsVolumeAttachment exposes an interface to support adding
44+
// and removing information about attached VMs to a PVC.
45+
type CnsVolumeAttachment interface {
46+
// Add adds the input VM UUID to the list of
47+
// attached VMs for the given volume.
48+
AddVmToAttachedList(ctx context.Context, volumeName, VmUUID string) error
49+
// RemoveVmFromAttachedList removes the input VM UUID from
50+
// the list of attached VMs for the given volume.
51+
RemoveVmFromAttachedList(ctx context.Context, volumeName, VmUUID string) error
52+
// CnsVolumeAttachmentExistsForPvc returns true if CnsVolumeAttachment for a PVC is found.
53+
CnsVolumeAttachmentExistsForPvc(ctx context.Context, volumeName string) (bool, error)
54+
}
55+
56+
// cnsVolumeAttachment maintains a client to the API
57+
// server for operations on CnsVolumeAttachment instance.
58+
// It also contains a per instance lock to handle
59+
// concurrent operations.
60+
type cnsVolumeAttachment struct {
61+
client client.Client
62+
// Per volume lock for concurrent access to CnsVolumeAttachment instances.
63+
// Keys are strings representing PVC names.
64+
// Values are individual sync.Mutex locks that need to be held
65+
// to make updates to the CnsVolumeAttachment instance on the API server.
66+
volumeLock *sync.Map
67+
}
68+
69+
// GetCnsVolumeAttachmentInstance returns a singleton of type CnsVolumeAttachment.
70+
// Initializes the singleton if not already initialized.
71+
func GetCnsVolumeAttachmentInstance(ctx context.Context) (CnsVolumeAttachment, error) {
72+
log := logger.GetLogger(ctx)
73+
74+
cnsVolumeAttachmentInstanceLock.Lock()
75+
log.Infof("Acquired lock for cnsVolumeAttachmentInstanceLock")
76+
defer func() {
77+
cnsVolumeAttachmentInstanceLock.Unlock()
78+
log.Infof("Released lock for cnsVolumeAttachmentInstanceLock")
79+
}()
80+
81+
if cnsVolumeAttachmentInstance == nil {
82+
config, err := k8s.GetKubeConfig(ctx)
83+
if err != nil {
84+
log.Errorf("failed to get kubeconfig. Err: %v", err)
85+
return nil, err
86+
}
87+
k8sclient, err := k8s.NewClientForGroup(ctx, config, internalapis.GroupName)
88+
if err != nil {
89+
log.Errorf("failed to create k8s client. Err: %v", err)
90+
return nil, err
91+
}
92+
cnsVolumeAttachmentInstance = &cnsVolumeAttachment{
93+
client: k8sclient,
94+
volumeLock: &sync.Map{},
95+
}
96+
}
97+
98+
return cnsVolumeAttachmentInstance, nil
99+
}
100+
101+
// Add adds the input VM UUID to the list of
102+
// attached VMs for the given volume.
103+
// Callers need to specify cnsVolumeAttachment as a combination of
104+
// "<SV-namespace>/<SV-PVC-name>". This combination is used to uniquely
105+
// identify CnsVolumeAttachment instances.
106+
// The instance is created if it doesn't exist.
107+
// Returns an error if the operation cannot be persisted on the API server.
108+
func (f *cnsVolumeAttachment) AddVmToAttachedList(ctx context.Context,
109+
volumeName, VmUUID string) error {
110+
log := logger.GetLogger(ctx)
111+
112+
log.Infof("Adding VM %s to cnsVolumeAttachment %s",
113+
VmUUID, volumeName)
114+
actual, _ := f.volumeLock.LoadOrStore(volumeName, &sync.Mutex{})
115+
instanceLock, ok := actual.(*sync.Mutex)
116+
if !ok {
117+
return fmt.Errorf("failed to cast lock for cnsVolumeAttachment instance: %s", volumeName)
118+
}
119+
instanceLock.Lock()
120+
log.Infof("Acquired lock for cnsVolumeAttachment instance %s", volumeName)
121+
defer func() {
122+
instanceLock.Unlock()
123+
log.Infof("Released lock for instance %s", volumeName)
124+
}()
125+
126+
instance := &v1alpha1.CnsVolumeAttachment{}
127+
instanceNamespace, instanceName, err := cache.SplitMetaNamespaceKey(volumeName)
128+
if err != nil {
129+
log.Errorf("failed to split key %s with error: %+v", volumeName, err)
130+
return err
131+
}
132+
instanceKey := types.NamespacedName{
133+
Namespace: instanceNamespace,
134+
Name: instanceName,
135+
}
136+
err = f.client.Get(ctx, instanceKey, instance)
137+
if err != nil {
138+
if errors.IsNotFound(err) {
139+
// Create the instance as it does not exist on the API server.
140+
instance = &v1alpha1.CnsVolumeAttachment{
141+
ObjectMeta: v1.ObjectMeta{
142+
Name: instanceName,
143+
Namespace: instanceNamespace,
144+
// Add finalizer so that CnsVolumeAttachment instance doesn't get deleted abruptly
145+
Finalizers: []string{cnsoperatortypes.CNSFinalizer},
146+
},
147+
Spec: v1alpha1.CnsVolumeAttachmentSpec{
148+
AttachedVms: []string{
149+
VmUUID,
150+
},
151+
},
152+
}
153+
log.Debugf("Creating cnsVolumeAttachment instance %s with spec: %+v", volumeName, instance)
154+
err = f.client.Create(ctx, instance)
155+
if err != nil {
156+
log.Errorf("failed to create cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
157+
return err
158+
}
159+
return nil
160+
}
161+
log.Errorf("failed to get cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
162+
return err
163+
}
164+
165+
// Verify if input VmUUID exists in existing AttachedVMs list.
166+
log.Debugf("Verifying if VM %s exists in current list of attached Vms. Current list: %+v",
167+
VmUUID, instance.Spec.AttachedVms)
168+
currentAttachedVmsList := instance.Spec.AttachedVms
169+
for _, currentAttachedVM := range currentAttachedVmsList {
170+
if currentAttachedVM == VmUUID {
171+
log.Debugf("Found VM %s in list. Returning.", VmUUID)
172+
return nil
173+
}
174+
}
175+
newAttachVmsList := append(currentAttachedVmsList, VmUUID)
176+
instance.Spec.AttachedVms = newAttachVmsList
177+
log.Debugf("Updating cnsVolumeAttachment instance %s with spec: %+v", volumeName, instance)
178+
err = f.client.Update(ctx, instance)
179+
if err != nil {
180+
log.Errorf("failed to update cnsVolumeAttachment instance %s/%s with error: %+v", volumeName, err)
181+
}
182+
return err
183+
}
184+
185+
// RemoveVmFromAttachedList removes the input VM UUID from
186+
// the list of attached VMs for the given volume.
187+
// Callers need to specify volumeName as a combination of
188+
// "<SV-namespace>/<SV-PVC-name>". This combination is used to uniquely
189+
// identify CnsVolumeAttachment instances.
190+
// If the given VM was the last client for this file volume, the instance is
191+
// deleted from the API server.
192+
// Returns an error if the operation cannot be persisted on the API server.
193+
func (f *cnsVolumeAttachment) RemoveVmFromAttachedList(ctx context.Context,
194+
volumeName, VmUUID string) error {
195+
log := logger.GetLogger(ctx)
196+
log.Infof("Removing VmUUID %s from cnsVolumeAttachment %s",
197+
VmUUID, volumeName)
198+
actual, _ := f.volumeLock.LoadOrStore(volumeName, &sync.Mutex{})
199+
instanceLock, ok := actual.(*sync.Mutex)
200+
if !ok {
201+
return fmt.Errorf("failed to cast lock for cnsVolumeAttachment instance: %s", volumeName)
202+
}
203+
instanceLock.Lock()
204+
log.Infof("Acquired lock for cnsVolumeAttachment instance %s", volumeName)
205+
defer func() {
206+
instanceLock.Unlock()
207+
log.Infof("Released lock for instance %s", volumeName)
208+
}()
209+
210+
instance := &v1alpha1.CnsVolumeAttachment{}
211+
instanceNamespace, instanceName, err := cache.SplitMetaNamespaceKey(volumeName)
212+
if err != nil {
213+
log.Errorf("failed to split key %s with error: %+v", volumeName, err)
214+
return err
215+
}
216+
instanceKey := types.NamespacedName{
217+
Namespace: instanceNamespace,
218+
Name: instanceName,
219+
}
220+
err = f.client.Get(ctx, instanceKey, instance)
221+
if err != nil {
222+
if errors.IsNotFound(err) {
223+
log.Infof("cnsVolumeAttachment instance %s does not exist on API server", volumeName)
224+
return nil
225+
}
226+
log.Errorf("failed to get cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
227+
return err
228+
}
229+
230+
log.Debugf("Verifying if VM UUID %s exists in list of already attached VMs. Current list: %+v",
231+
volumeName, instance.Spec.AttachedVms)
232+
for index, existingAttachedVM := range instance.Spec.AttachedVms {
233+
if VmUUID == existingAttachedVM {
234+
log.Infof("Removing VmUUID %s from Attached VMs list", VmUUID)
235+
instance.Spec.AttachedVms = append(
236+
instance.Spec.AttachedVms[:index],
237+
instance.Spec.AttachedVms[index+1:]...)
238+
if len(instance.Spec.AttachedVms) == 0 {
239+
log.Infof("Deleting cnsVolumeAttachment instance %s from API server", volumeName)
240+
// Remove finalizer from CnsVolumeAttachment instance
241+
err = removeFinalizer(ctx, f.client, instance)
242+
if err != nil {
243+
log.Errorf("failed to remove finalizer from cnsVolumeAttachment instance %s with error: %+v",
244+
volumeName, err)
245+
}
246+
err = f.client.Delete(ctx, instance)
247+
if err != nil {
248+
// In case of namespace deletion, we will have deletion timestamp added on the
249+
// CnsVolumeAttachment instance. So, as soon as we delete finalizer, instance might
250+
// get deleted immediately. In such cases we will get NotFound error here, return success
251+
// if instance is already deleted.
252+
if errors.IsNotFound(err) {
253+
log.Infof("cnsVolumeAttachment instance %s seems to be already deleted.", volumeName)
254+
f.volumeLock.Delete(volumeName)
255+
return nil
256+
}
257+
log.Errorf("failed to delete cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
258+
return err
259+
}
260+
f.volumeLock.Delete(volumeName)
261+
return nil
262+
}
263+
log.Debugf("Updating cnsVolumeAttachment instance %s with spec: %+v", volumeName, instance)
264+
err = f.client.Update(ctx, instance)
265+
if err != nil {
266+
log.Errorf("failed to update cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
267+
}
268+
return err
269+
}
270+
}
271+
log.Infof("Could not find VM %s in list. Returning.", VmUUID)
272+
return nil
273+
}
274+
275+
// CnsVolumeAttachmentExistsForPvc returns true if CnsVolumeAttachment instance
276+
// for the given PVC exists.
277+
// Callers need to specify volumeName as a combination of
278+
// "<SV-namespace>/<SV-PVC-name>". This combination is used to uniquely
279+
// identify CnsVolumeAttachment instances.
280+
// If the given VM was the last client for this file volume, the instance is
281+
// deleted from the API server.
282+
// Returns an error if the operation cannot be persisted on the API server.
283+
func (f *cnsVolumeAttachment) CnsVolumeAttachmentExistsForPvc(ctx context.Context, volumeName string) (bool, error) {
284+
log := logger.GetLogger(ctx)
285+
286+
log.Infof("Fetching cnsfilevolumeclient instance for volume %s", volumeName)
287+
actual, _ := f.volumeLock.LoadOrStore(volumeName, &sync.Mutex{})
288+
instanceLock, ok := actual.(*sync.Mutex)
289+
if !ok {
290+
return true, fmt.Errorf("failed to cast lock for cnsVolumeAttachment instance: %s", volumeName)
291+
}
292+
instanceLock.Lock()
293+
log.Infof("Acquired lock for cnsVolumeAttachment instance %s", volumeName)
294+
defer func() {
295+
instanceLock.Unlock()
296+
log.Infof("Released lock for instance %s", volumeName)
297+
}()
298+
299+
instance := &v1alpha1.CnsVolumeAttachment{}
300+
instanceNamespace, instanceName, err := cache.SplitMetaNamespaceKey(volumeName)
301+
if err != nil {
302+
log.Errorf("failed to split key %s with error: %+v", volumeName, err)
303+
return true, err
304+
}
305+
instanceKey := types.NamespacedName{
306+
Namespace: instanceNamespace,
307+
Name: instanceName,
308+
}
309+
err = f.client.Get(ctx, instanceKey, instance)
310+
if err != nil {
311+
if errors.IsNotFound(err) {
312+
// CnsVolumeAttachment instance not found.
313+
// This means PVC is not being used by any VM.
314+
return false, nil
315+
}
316+
log.Errorf("failed to get cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
317+
return true, err
318+
}
319+
return true, nil
320+
}
321+
322+
// removeFinalizer will remove the CNS Finalizer = cns.vmware.com,
323+
// from a given CnsVolumeAttachment instance.
324+
func removeFinalizer(ctx context.Context, client client.Client,
325+
instance *v1alpha1.CnsVolumeAttachment) error {
326+
log := logger.GetLogger(ctx)
327+
328+
if !controllerutil.ContainsFinalizer(instance, cnsoperatortypes.CNSFinalizer) {
329+
// Finalizer not present on instance. Nothing to do.
330+
return nil
331+
}
332+
333+
finalizersOnInstance := instance.Finalizers
334+
for i, finalizer := range instance.Finalizers {
335+
if finalizer == cnsoperatortypes.CNSFinalizer {
336+
log.Infof("Removing %q finalizer from CnsNodeVmBatchAttachment instance with name: %q on namespace: %q",
337+
cnsoperatortypes.CNSFinalizer, instance.Name, instance.Namespace)
338+
finalizersOnInstance = append(instance.Finalizers[:i], instance.Finalizers[i+1:]...)
339+
break
340+
}
341+
}
342+
return k8s.PatchFinalizers(ctx, client, instance, finalizersOnInstance)
343+
}

0 commit comments

Comments
 (0)