Clean up orphaned snapshot attachment tickets while upgrading

Longhorn 6652

Signed-off-by: Eric Weber <[email protected]>

upgrade/util/util.go

@@ -562,6 +562,26 @@ func ListAndUpdateRecurringJobsInProvidedCache(namespace string, lhClient *lhcli
 	return recurringJobs, nil
 }
 
+// ListAndUpdateVolumeAttachmentsInProvidedCache list all volumeAttachments and save them into the provided cached `resourceMap`. This method is not thread-safe.
+func ListAndUpdateVolumeAttachmentsInProvidedCache(namespace string, lhClient *lhclientset.Clientset, resourceMaps map[string]interface{}) (map[string]*longhorn.VolumeAttachment, error) {
+	if v, ok := resourceMaps[types.LonghornKindVolumeAttachment]; ok {
+		return v.(map[string]*longhorn.VolumeAttachment), nil
+	}
+
+	volumeAttachments := map[string]*longhorn.VolumeAttachment{}
+	volumeAttachmentList, err := lhClient.LonghornV1beta2().VolumeAttachments(namespace).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+	for i, volumeAttachment := range volumeAttachmentList.Items {
+		volumeAttachments[volumeAttachment.Name] = &volumeAttachmentList.Items[i]
+	}
+
+	resourceMaps[types.LonghornKindVolumeAttachment] = volumeAttachments
+
+	return volumeAttachments, nil
+}
+
 // CreateAndUpdateRecurringJobInProvidedCache creates a recurringJob and saves it into the provided cached `resourceMap`. This method is not thread-safe.
 func CreateAndUpdateRecurringJobInProvidedCache(namespace string, lhClient *lhclientset.Clientset, resourceMaps map[string]interface{}, job *longhorn.RecurringJob) (*longhorn.RecurringJob, error) {
 	obj, err := lhClient.LonghornV1beta2().RecurringJobs(namespace).Create(context.TODO(), job, metav1.CreateOptions{})
@@ -911,23 +931,21 @@ func updateOrphans(namespace string, lhClient *lhclientset.Clientset, orphans ma
 	return nil
 }
 
-func updateVolumeAttachments(namespace string, lhClient *lhclientset.Clientset, volumeAttachment map[string]*longhorn.VolumeAttachment) error {
+func updateVolumeAttachments(namespace string, lhClient *lhclientset.Clientset, volumeAttachments map[string]*longhorn.VolumeAttachment) error {
 	existingVolumeAttachmentList, err := lhClient.LonghornV1beta2().VolumeAttachments(namespace).List(context.TODO(), metav1.ListOptions{})
 	if err != nil {
 		return err
 	}
-
-	existingVolumeAttachmentMap := map[string]bool{}
-	for _, volume := range existingVolumeAttachmentList.Items {
-		existingVolumeAttachmentMap[volume.Name] = true
-	}
-
-	for _, va := range volumeAttachment {
-		if _, ok := existingVolumeAttachmentMap[va.Name]; ok {
+	for _, existingVolumeAttachment := range existingVolumeAttachmentList.Items {
+		volumeAttachment, ok := volumeAttachments[existingVolumeAttachment.Name]
+		if !ok {
 			continue
 		}
-		if _, err = lhClient.LonghornV1beta2().VolumeAttachments(namespace).Create(context.TODO(), va, metav1.CreateOptions{}); err != nil {
-			return err
+		if !reflect.DeepEqual(existingVolumeAttachment.Spec, volumeAttachment.Spec) ||
+			!reflect.DeepEqual(existingVolumeAttachment.ObjectMeta, volumeAttachment.ObjectMeta) {
+			if _, err = lhClient.LonghornV1beta2().VolumeAttachments(namespace).Update(context.TODO(), volumeAttachment, metav1.UpdateOptions{}); err != nil {
+				return err
+			}
 		}
 	}
 

upgrade/v15xto160/upgrade.go

@@ -16,9 +16,11 @@ const (
 )
 
 func UpgradeResources(namespace string, lhClient *lhclientset.Clientset, kubeClient *clientset.Clientset, resourceMaps map[string]interface{}) error {
-	// We will probably need to upgrade other resources as well. See upgradeVolumes or previous Longhorn versions for
-	// examples.
-	return upgradeVolumes(namespace, lhClient, resourceMaps)
+	if err := upgradeVolumes(namespace, lhClient, resourceMaps); err != nil {
+		return err
+	}
+
+	return upgradeVolumeAttachments(namespace, lhClient, resourceMaps)
 }
 
 func upgradeVolumes(namespace string, lhClient *lhclientset.Clientset, resourceMaps map[string]interface{}) (err error) {
@@ -43,6 +45,48 @@ func upgradeVolumes(namespace string, lhClient *lhclientset.Clientset, resourceM
 	return nil
 }
 
+func upgradeVolumeAttachments(namespace string, lhClient *lhclientset.Clientset, resourceMaps map[string]interface{}) (err error) {
+	defer func() {
+		err = errors.Wrapf(err, upgradeLogPrefix+"upgrade VolumeAttachment failed")
+	}()
+
+	volumeAttachmentMap, err := upgradeutil.ListAndUpdateVolumeAttachmentsInProvidedCache(namespace, lhClient, resourceMaps)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return nil
+		}
+		return errors.Wrapf(err, "failed to list all existing Longhorn VolumeAttachments during the Longhorn VolumeAttachment upgrade")
+	}
+
+	snapshotMap, err := upgradeutil.ListAndUpdateSnapshotsInProvidedCache(namespace, lhClient, resourceMaps)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return nil
+		}
+		return errors.Wrapf(err, "failed to list all existing Snapshots during the Longhorn VolumeAttachment a upgrade")
+	}
+
+	ticketIDsForExistingSnapshotsMap := map[string]interface{}{}
+	for snapshotName := range snapshotMap {
+		ticketID := longhorn.GetAttachmentTicketID(longhorn.AttacherTypeSnapshotController, snapshotName)
+		ticketIDsForExistingSnapshotsMap[ticketID] = nil
+	}
+
+	// Previous Longhorn versions may have created attachmentTickets for snapshots that no longer exist. Clean these up.
+	for _, volumeAttachment := range volumeAttachmentMap {
+		for ticketID, ticket := range volumeAttachment.Spec.AttachmentTickets {
+			if ticket.Type != longhorn.AttacherTypeSnapshotController {
+				continue
+			}
+			if _, ok := ticketIDsForExistingSnapshotsMap[ticketID]; !ok {
+				delete(volumeAttachment.Spec.AttachmentTickets, ticketID)
+			}
+		}
+	}
+
+	return nil
+}
+
 func UpgradeResourcesStatus(namespace string, lhClient *lhclientset.Clientset, kubeClient *clientset.Clientset, resourceMaps map[string]interface{}) error {
 	// Currently there are no statuses to upgrade. See UpgradeResources -> upgradeVolumes or previous Longhorn versions
 	// for examples.