Skip to content

Commit 647c7eb

Browse files
skogtaskogta
committed
Add CnsVolumeAttachment CRD
1 parent 986aa2d commit 647c7eb

File tree

14 files changed

+968
-22
lines changed

14 files changed

+968
-22
lines changed

‎hack/release.sh

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ set -o nounset
2323
set -o pipefail
2424
set -x
2525

26-
DO_WINDOWS_BUILD=${DO_WINDOWS_BUILD_ENV:-true}
26+
DO_WINDOWS_BUILD=${DO_WINDOWS_BUILD_ENV:-false}
2727

2828
# BASE_REPO is the root path of the image repository
2929
readonly BASE_IMAGE_REPO=us-central1-docker.pkg.dev/k8s-staging-images/csi-vsphere

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

Lines changed: 347 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,347 @@
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 instance UUID to the list of
47+
// attached VMs for the given volume.
48+
AddVmToAttachedList(ctx context.Context, volumeName, VmInstanceUUID 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, VmInstanceUUID string) (error, bool)
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 InstanceUUID 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, VmInstanceUUID string) error {
110+
log := logger.GetLogger(ctx)
111+
112+
log.Infof("Adding VM %s to cnsVolumeAttachment %s",
113+
VmInstanceUUID, 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+
VmInstanceUUID,
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 VmInstanceUUID exists in existing AttachedVMs list.
166+
log.Debugf("Verifying if VM %s exists in current list of attached Vms. Current list: %+v",
167+
VmInstanceUUID, instance.Spec.AttachedVms)
168+
currentAttachedVmsList := instance.Spec.AttachedVms
169+
for _, currentAttachedVM := range currentAttachedVmsList {
170+
if currentAttachedVM == VmInstanceUUID {
171+
log.Debugf("Found VM %s in list. Returning.", VmInstanceUUID)
172+
return nil
173+
}
174+
}
175+
newAttachVmsList := append(currentAttachedVmsList, VmInstanceUUID)
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, VmInstanceUUID string) (error, bool) {
195+
log := logger.GetLogger(ctx)
196+
log.Infof("Removing VmInstanceUUID %s from cnsVolumeAttachment %s",
197+
VmInstanceUUID, 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+
false
203+
}
204+
instanceLock.Lock()
205+
log.Infof("Acquired lock for cnsVolumeAttachment instance %s", volumeName)
206+
defer func() {
207+
instanceLock.Unlock()
208+
log.Infof("Released lock for instance %s", volumeName)
209+
}()
210+
211+
instance := &v1alpha1.CnsVolumeAttachment{}
212+
instanceNamespace, instanceName, err := cache.SplitMetaNamespaceKey(volumeName)
213+
if err != nil {
214+
log.Errorf("failed to split key %s with error: %+v", volumeName, err)
215+
return err, false
216+
}
217+
instanceKey := types.NamespacedName{
218+
Namespace: instanceNamespace,
219+
Name: instanceName,
220+
}
221+
err = f.client.Get(ctx, instanceKey, instance)
222+
if err != nil {
223+
if errors.IsNotFound(err) {
224+
log.Infof("cnsVolumeAttachment instance %s does not exist on API server", volumeName)
225+
return nil, true
226+
}
227+
log.Errorf("failed to get cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
228+
return err, false
229+
}
230+
231+
log.Infof("Verifying if VM UUID %s exists in list of already attached VMs. Current list: %+v",
232+
volumeName, instance.Spec.AttachedVms)
233+
for index, existingAttachedVM := range instance.Spec.AttachedVms {
234+
if VmInstanceUUID != existingAttachedVM {
235+
continue
236+
}
237+
log.Infof("Removing VmUUID %s from Attached VMs list", VmInstanceUUID)
238+
instance.Spec.AttachedVms = append(
239+
instance.Spec.AttachedVms[:index],
240+
instance.Spec.AttachedVms[index+1:]...)
241+
if len(instance.Spec.AttachedVms) == 0 {
242+
log.Infof("Deleting cnsVolumeAttachment instance %s from API server", volumeName)
243+
// Remove finalizer from CnsVolumeAttachment instance
244+
err = removeFinalizer(ctx, f.client, instance)
245+
if err != nil {
246+
log.Errorf("failed to remove finalizer from cnsVolumeAttachment instance %s with error: %+v",
247+
volumeName, err)
248+
return err, false
249+
}
250+
err = f.client.Delete(ctx, instance)
251+
if err != nil {
252+
// In case of namespace deletion, we will have deletion timestamp added on the
253+
// CnsVolumeAttachment instance. So, as soon as we delete finalizer, instance might
254+
// get deleted immediately. In such cases we will get NotFound error here, return success
255+
// if instance is already deleted.
256+
if errors.IsNotFound(err) {
257+
log.Infof("cnsVolumeAttachment instance %s seems to be already deleted.", volumeName)
258+
f.volumeLock.Delete(volumeName)
259+
return nil, true
260+
}
261+
log.Errorf("failed to delete cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
262+
return err, false
263+
}
264+
log.Infof("Successfully deleted cnsVolumeAttachment instance %s", volumeName)
265+
f.volumeLock.Delete(volumeName)
266+
return nil, true
267+
}
268+
log.Debugf("Updating cnsVolumeAttachment instance %s with spec: %+v", volumeName, instance)
269+
err = f.client.Update(ctx, instance)
270+
if err != nil {
271+
log.Errorf("failed to update cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
272+
}
273+
return err, false
274+
}
275+
log.Infof("Could not find VM %s in list. Returning.", VmInstanceUUID)
276+
return nil, false
277+
}
278+
279+
// CnsVolumeAttachmentExistsForPvc returns true if CnsVolumeAttachment instance
280+
// for the given PVC exists.
281+
// Callers need to specify volumeName as a combination of
282+
// "<SV-namespace>/<SV-PVC-name>". This combination is used to uniquely
283+
// identify CnsVolumeAttachment instances.
284+
// If the given VM was the last client for this file volume, the instance is
285+
// deleted from the API server.
286+
// Returns an error if the operation cannot be persisted on the API server.
287+
func (f *cnsVolumeAttachment) CnsVolumeAttachmentExistsForPvc(ctx context.Context, volumeName string) (bool, error) {
288+
log := logger.GetLogger(ctx)
289+
290+
log.Infof("Fetching cnsfilevolumeclient instance for volume %s", volumeName)
291+
actual, _ := f.volumeLock.LoadOrStore(volumeName, &sync.Mutex{})
292+
instanceLock, ok := actual.(*sync.Mutex)
293+
if !ok {
294+
return true, fmt.Errorf("failed to cast lock for cnsVolumeAttachment instance: %s", volumeName)
295+
}
296+
instanceLock.Lock()
297+
log.Infof("Acquired lock for cnsVolumeAttachment instance %s", volumeName)
298+
defer func() {
299+
instanceLock.Unlock()
300+
log.Infof("Released lock for instance %s", volumeName)
301+
}()
302+
303+
instance := &v1alpha1.CnsVolumeAttachment{}
304+
instanceNamespace, instanceName, err := cache.SplitMetaNamespaceKey(volumeName)
305+
if err != nil {
306+
log.Errorf("failed to split key %s with error: %+v", volumeName, err)
307+
return true, err
308+
}
309+
instanceKey := types.NamespacedName{
310+
Namespace: instanceNamespace,
311+
Name: instanceName,
312+
}
313+
err = f.client.Get(ctx, instanceKey, instance)
314+
if err != nil {
315+
if errors.IsNotFound(err) {
316+
// CnsVolumeAttachment instance not found.
317+
// This means PVC is not being used by any VM.
318+
return false, nil
319+
}
320+
log.Errorf("failed to get cnsVolumeAttachment instance %s with error: %+v", volumeName, err)
321+
return true, err
322+
}
323+
return true, nil
324+
}
325+
326+
// removeFinalizer will remove the CNS Finalizer = cns.vmware.com,
327+
// from a given CnsVolumeAttachment instance.
328+
func removeFinalizer(ctx context.Context, client client.Client,
329+
instance *v1alpha1.CnsVolumeAttachment) error {
330+
log := logger.GetLogger(ctx)
331+
332+
if !controllerutil.ContainsFinalizer(instance, cnsoperatortypes.CNSFinalizer) {
333+
// Finalizer not present on instance. Nothing to do.
334+
return nil
335+
}
336+
337+
finalizersOnInstance := instance.Finalizers
338+
for i, finalizer := range instance.Finalizers {
339+
if finalizer == cnsoperatortypes.CNSFinalizer {
340+
log.Infof("Removing %q finalizer from CnsNodeVmBatchAttachment instance with name: %q on namespace: %q",
341+
cnsoperatortypes.CNSFinalizer, instance.Name, instance.Namespace)
342+
finalizersOnInstance = append(instance.Finalizers[:i], instance.Finalizers[i+1:]...)
343+
break
344+
}
345+
}
346+
return k8s.PatchFinalizers(ctx, client, instance, finalizersOnInstance)
347+
}

0 commit comments

Comments
 (0)