diff --git a/datastore/datastore.go b/datastore/datastore.go index e933165592..f090d11a60 100644 --- a/datastore/datastore.go +++ b/datastore/datastore.go @@ -45,51 +45,55 @@ type DataStore struct { cacheSyncs []cache.InformerSynced - lhClient lhclientset.Interface - volumeLister lhlisters.VolumeLister - VolumeInformer cache.SharedInformer - engineLister lhlisters.EngineLister - EngineInformer cache.SharedInformer - replicaLister lhlisters.ReplicaLister - ReplicaInformer cache.SharedInformer - engineImageLister lhlisters.EngineImageLister - EngineImageInformer cache.SharedInformer - nodeLister lhlisters.NodeLister - NodeInformer cache.SharedInformer - settingLister lhlisters.SettingLister - SettingInformer cache.SharedInformer - instanceManagerLister lhlisters.InstanceManagerLister - InstanceManagerInformer cache.SharedInformer - shareManagerLister lhlisters.ShareManagerLister - ShareManagerInformer cache.SharedInformer - backingImageLister lhlisters.BackingImageLister - BackingImageInformer cache.SharedInformer - backingImageManagerLister lhlisters.BackingImageManagerLister - BackingImageManagerInformer cache.SharedInformer - backingImageDataSourceLister lhlisters.BackingImageDataSourceLister - BackingImageDataSourceInformer cache.SharedInformer - backupBackingImageLister lhlisters.BackupBackingImageLister - BackupBackingImageInformer cache.SharedInformer - backupTargetLister lhlisters.BackupTargetLister - BackupTargetInformer cache.SharedInformer - backupVolumeLister lhlisters.BackupVolumeLister - BackupVolumeInformer cache.SharedInformer - backupLister lhlisters.BackupLister - BackupInformer cache.SharedInformer - recurringJobLister lhlisters.RecurringJobLister - RecurringJobInformer cache.SharedInformer - orphanLister lhlisters.OrphanLister - OrphanInformer cache.SharedInformer - snapshotLister lhlisters.SnapshotLister - SnapshotInformer cache.SharedInformer - supportBundleLister lhlisters.SupportBundleLister - SupportBundleInformer cache.SharedInformer - systemBackupLister lhlisters.SystemBackupLister - SystemBackupInformer cache.SharedInformer - systemRestoreLister lhlisters.SystemRestoreLister - SystemRestoreInformer cache.SharedInformer - lhVolumeAttachmentLister lhlisters.VolumeAttachmentLister - LHVolumeAttachmentInformer cache.SharedInformer + lhClient lhclientset.Interface + volumeLister lhlisters.VolumeLister + VolumeInformer cache.SharedInformer + engineLister lhlisters.EngineLister + EngineInformer cache.SharedInformer + replicaLister lhlisters.ReplicaLister + ReplicaInformer cache.SharedInformer + engineImageLister lhlisters.EngineImageLister + EngineImageInformer cache.SharedInformer + nodeLister lhlisters.NodeLister + NodeInformer cache.SharedInformer + settingLister lhlisters.SettingLister + SettingInformer cache.SharedInformer + instanceManagerLister lhlisters.InstanceManagerLister + InstanceManagerInformer cache.SharedInformer + shareManagerLister lhlisters.ShareManagerLister + ShareManagerInformer cache.SharedInformer + backingImageLister lhlisters.BackingImageLister + BackingImageInformer cache.SharedInformer + backingImageManagerLister lhlisters.BackingImageManagerLister + BackingImageManagerInformer cache.SharedInformer + backingImageDataSourceLister lhlisters.BackingImageDataSourceLister + BackingImageDataSourceInformer cache.SharedInformer + backupBackingImageLister lhlisters.BackupBackingImageLister + BackupBackingImageInformer cache.SharedInformer + backupTargetLister lhlisters.BackupTargetLister + BackupTargetInformer cache.SharedInformer + backupVolumeLister lhlisters.BackupVolumeLister + BackupVolumeInformer cache.SharedInformer + backupLister lhlisters.BackupLister + BackupInformer cache.SharedInformer + recurringJobLister lhlisters.RecurringJobLister + RecurringJobInformer cache.SharedInformer + orphanLister lhlisters.OrphanLister + OrphanInformer cache.SharedInformer + snapshotLister lhlisters.SnapshotLister + SnapshotInformer cache.SharedInformer + supportBundleLister lhlisters.SupportBundleLister + SupportBundleInformer cache.SharedInformer + systemBackupLister lhlisters.SystemBackupLister + SystemBackupInformer cache.SharedInformer + systemRestoreLister lhlisters.SystemRestoreLister + SystemRestoreInformer cache.SharedInformer + lhVolumeAttachmentLister lhlisters.VolumeAttachmentLister + LHVolumeAttachmentInformer cache.SharedInformer + dataEngineUpgradeManagerLister lhlisters.DataEngineUpgradeManagerLister + DataEngineUpgradeManagerInformer cache.SharedInformer + nodeDataEngineUpgradeLister lhlisters.NodeDataEngineUpgradeLister + NodeDataEngineUpgradeInformer cache.SharedInformer kubeClient clientset.Interface podLister corelisters.PodLister @@ -179,6 +183,10 @@ func NewDataStore(namespace string, lhClient lhclientset.Interface, kubeClient c cacheSyncs = append(cacheSyncs, systemRestoreInformer.Informer().HasSynced) lhVolumeAttachmentInformer := informerFactories.LhInformerFactory.Longhorn().V1beta2().VolumeAttachments() cacheSyncs = append(cacheSyncs, lhVolumeAttachmentInformer.Informer().HasSynced) + dataEngineUpgradeManagerInformer := informerFactories.LhInformerFactory.Longhorn().V1beta2().DataEngineUpgradeManagers() + cacheSyncs = append(cacheSyncs, dataEngineUpgradeManagerInformer.Informer().HasSynced) + nodeDataEngineUpgradeInformer := informerFactories.LhInformerFactory.Longhorn().V1beta2().NodeDataEngineUpgrades() + cacheSyncs = append(cacheSyncs, nodeDataEngineUpgradeInformer.Informer().HasSynced) // Kube Informers podInformer := informerFactories.KubeInformerFactory.Core().V1().Pods() @@ -223,51 +231,55 @@ func NewDataStore(namespace string, lhClient lhclientset.Interface, kubeClient c cacheSyncs: cacheSyncs, - lhClient: lhClient, - volumeLister: volumeInformer.Lister(), - VolumeInformer: volumeInformer.Informer(), - engineLister: engineInformer.Lister(), - EngineInformer: engineInformer.Informer(), - replicaLister: replicaInformer.Lister(), - ReplicaInformer: replicaInformer.Informer(), - engineImageLister: engineImageInformer.Lister(), - EngineImageInformer: engineImageInformer.Informer(), - nodeLister: nodeInformer.Lister(), - NodeInformer: nodeInformer.Informer(), - settingLister: settingInformer.Lister(), - SettingInformer: settingInformer.Informer(), - instanceManagerLister: instanceManagerInformer.Lister(), - InstanceManagerInformer: instanceManagerInformer.Informer(), - shareManagerLister: shareManagerInformer.Lister(), - ShareManagerInformer: shareManagerInformer.Informer(), - backingImageLister: backingImageInformer.Lister(), - BackingImageInformer: backingImageInformer.Informer(), - backingImageManagerLister: backingImageManagerInformer.Lister(), - BackingImageManagerInformer: backingImageManagerInformer.Informer(), - backingImageDataSourceLister: backingImageDataSourceInformer.Lister(), - BackingImageDataSourceInformer: backingImageDataSourceInformer.Informer(), - backupBackingImageLister: backupBackingImageInformer.Lister(), - BackupBackingImageInformer: backupBackingImageInformer.Informer(), - backupTargetLister: backupTargetInformer.Lister(), - BackupTargetInformer: backupTargetInformer.Informer(), - backupVolumeLister: backupVolumeInformer.Lister(), - BackupVolumeInformer: backupVolumeInformer.Informer(), - backupLister: backupInformer.Lister(), - BackupInformer: backupInformer.Informer(), - recurringJobLister: recurringJobInformer.Lister(), - RecurringJobInformer: recurringJobInformer.Informer(), - orphanLister: orphanInformer.Lister(), - OrphanInformer: orphanInformer.Informer(), - snapshotLister: snapshotInformer.Lister(), - SnapshotInformer: snapshotInformer.Informer(), - supportBundleLister: supportBundleInformer.Lister(), - SupportBundleInformer: supportBundleInformer.Informer(), - systemBackupLister: systemBackupInformer.Lister(), - SystemBackupInformer: systemBackupInformer.Informer(), - systemRestoreLister: systemRestoreInformer.Lister(), - SystemRestoreInformer: systemRestoreInformer.Informer(), - lhVolumeAttachmentLister: lhVolumeAttachmentInformer.Lister(), - LHVolumeAttachmentInformer: lhVolumeAttachmentInformer.Informer(), + lhClient: lhClient, + volumeLister: volumeInformer.Lister(), + VolumeInformer: volumeInformer.Informer(), + engineLister: engineInformer.Lister(), + EngineInformer: engineInformer.Informer(), + replicaLister: replicaInformer.Lister(), + ReplicaInformer: replicaInformer.Informer(), + engineImageLister: engineImageInformer.Lister(), + EngineImageInformer: engineImageInformer.Informer(), + nodeLister: nodeInformer.Lister(), + NodeInformer: nodeInformer.Informer(), + settingLister: settingInformer.Lister(), + SettingInformer: settingInformer.Informer(), + instanceManagerLister: instanceManagerInformer.Lister(), + InstanceManagerInformer: instanceManagerInformer.Informer(), + shareManagerLister: shareManagerInformer.Lister(), + ShareManagerInformer: shareManagerInformer.Informer(), + backingImageLister: backingImageInformer.Lister(), + BackingImageInformer: backingImageInformer.Informer(), + backingImageManagerLister: backingImageManagerInformer.Lister(), + BackingImageManagerInformer: backingImageManagerInformer.Informer(), + backingImageDataSourceLister: backingImageDataSourceInformer.Lister(), + BackingImageDataSourceInformer: backingImageDataSourceInformer.Informer(), + backupBackingImageLister: backupBackingImageInformer.Lister(), + BackupBackingImageInformer: backupBackingImageInformer.Informer(), + backupTargetLister: backupTargetInformer.Lister(), + BackupTargetInformer: backupTargetInformer.Informer(), + backupVolumeLister: backupVolumeInformer.Lister(), + BackupVolumeInformer: backupVolumeInformer.Informer(), + backupLister: backupInformer.Lister(), + BackupInformer: backupInformer.Informer(), + recurringJobLister: recurringJobInformer.Lister(), + RecurringJobInformer: recurringJobInformer.Informer(), + orphanLister: orphanInformer.Lister(), + OrphanInformer: orphanInformer.Informer(), + snapshotLister: snapshotInformer.Lister(), + SnapshotInformer: snapshotInformer.Informer(), + supportBundleLister: supportBundleInformer.Lister(), + SupportBundleInformer: supportBundleInformer.Informer(), + systemBackupLister: systemBackupInformer.Lister(), + SystemBackupInformer: systemBackupInformer.Informer(), + systemRestoreLister: systemRestoreInformer.Lister(), + SystemRestoreInformer: systemRestoreInformer.Informer(), + lhVolumeAttachmentLister: lhVolumeAttachmentInformer.Lister(), + LHVolumeAttachmentInformer: lhVolumeAttachmentInformer.Informer(), + dataEngineUpgradeManagerLister: dataEngineUpgradeManagerInformer.Lister(), + DataEngineUpgradeManagerInformer: dataEngineUpgradeManagerInformer.Informer(), + nodeDataEngineUpgradeLister: nodeDataEngineUpgradeInformer.Lister(), + NodeDataEngineUpgradeInformer: nodeDataEngineUpgradeInformer.Informer(), kubeClient: kubeClient, podLister: podInformer.Lister(), diff --git a/datastore/longhorn.go b/datastore/longhorn.go index b8e5d98bcc..7b6de9115a 100644 --- a/datastore/longhorn.go +++ b/datastore/longhorn.go @@ -1798,7 +1798,7 @@ func (s *DataStore) ListVolumePDBProtectedHealthyReplicasRO(volumeName string) ( func (s *DataStore) getRunningReplicaInstanceManagerRO(r *longhorn.Replica) (im *longhorn.InstanceManager, err error) { if r.Status.InstanceManagerName == "" { - im, err = s.GetInstanceManagerByInstanceRO(r) + im, err = s.GetInstanceManagerByInstanceRO(r, false) if err != nil && !types.ErrorIsNotFound(err) { return nil, err } @@ -3739,15 +3739,15 @@ func (s *DataStore) ListInstanceManagersBySelectorRO(node, imImage string, imTyp // GetInstanceManagerByInstance returns an InstanceManager for a given object, // or an error if more than one InstanceManager is found. -func (s *DataStore) GetInstanceManagerByInstance(obj interface{}) (*longhorn.InstanceManager, error) { - im, err := s.GetInstanceManagerByInstanceRO(obj) +func (s *DataStore) GetInstanceManagerByInstance(obj interface{}, isInstanceOnRemoteNode bool) (*longhorn.InstanceManager, error) { + im, err := s.GetInstanceManagerByInstanceRO(obj, isInstanceOnRemoteNode) if err != nil { return nil, err } return im.DeepCopy(), nil } -func (s *DataStore) GetInstanceManagerByInstanceRO(obj interface{}) (*longhorn.InstanceManager, error) { +func (s *DataStore) GetInstanceManagerByInstanceRO(obj interface{}, isInstanceOnRemoteNode bool) (*longhorn.InstanceManager, error) { var ( name string // name of the object nodeID string @@ -3759,6 +3759,9 @@ func (s *DataStore) GetInstanceManagerByInstanceRO(obj interface{}) (*longhorn.I name = obj.Name dataEngine = obj.Spec.DataEngine nodeID = obj.Spec.NodeID + if isInstanceOnRemoteNode { + nodeID = obj.Spec.TargetNodeIDForLiveUpgrade + } case *longhorn.Replica: name = obj.Name dataEngine = obj.Spec.DataEngine @@ -3985,14 +3988,14 @@ func (s *DataStore) GetEngineImageCLIAPIVersion(imageName string) (int, error) { // GetDataEngineImageCLIAPIVersion get engine or instance manager image for the given name and returns the CLIAPIVersion func (s *DataStore) GetDataEngineImageCLIAPIVersion(imageName string, dataEngine longhorn.DataEngineType) (int, error) { - if imageName == "" { - return -1, fmt.Errorf("cannot check the CLI API Version based on empty image name") - } - if types.IsDataEngineV2(dataEngine) { return 0, nil } + if imageName == "" { + return -1, fmt.Errorf("cannot check the CLI API Version based on empty image name") + } + ei, err := s.GetEngineImageRO(types.GetEngineImageChecksumName(imageName)) if err != nil { return -1, errors.Wrapf(err, "failed to get engine image object based on image name %v", imageName) @@ -5635,6 +5638,251 @@ func (s *DataStore) ListBackupBackingImagesRO() ([]*longhorn.BackupBackingImage, return s.backupBackingImageLister.BackupBackingImages(s.namespace).List(labels.Everything()) } +// CreateDataEngineUpgradeManager creates a Longhorn DataEngineUpgradeManager resource and verifies creation +func (s *DataStore) CreateDataEngineUpgradeManager(upgradeManager *longhorn.DataEngineUpgradeManager) (*longhorn.DataEngineUpgradeManager, error) { + ret, err := s.lhClient.LonghornV1beta2().DataEngineUpgradeManagers(s.namespace).Create(context.TODO(), upgradeManager, metav1.CreateOptions{}) + if err != nil { + return nil, err + } + if SkipListerCheck { + return ret, nil + } + + obj, err := verifyCreation(ret.Name, "dataEngineUpgradeManager", func(name string) (k8sruntime.Object, error) { + return s.GetDataEngineUpgradeManagerRO(name) + }) + if err != nil { + return nil, err + } + ret, ok := obj.(*longhorn.DataEngineUpgradeManager) + if !ok { + return nil, fmt.Errorf("BUG: datastore: verifyCreation returned wrong type for dataEngineUpgradeManager") + } + + return ret.DeepCopy(), nil +} + +// GetDataEngineUpgradeManagerRO returns the DataEngineUpgradeManager with the given dataEngineUpgradeManager name in the cluster +func (s *DataStore) GetDataEngineUpgradeManagerRO(upgradeManagerName string) (*longhorn.DataEngineUpgradeManager, error) { + return s.dataEngineUpgradeManagerLister.DataEngineUpgradeManagers(s.namespace).Get(upgradeManagerName) +} + +// GetDataEngineUpgradeManager returns a copy of DataEngineUpgradeManager with the given dataEngineUpgradeManager name in the cluster +func (s *DataStore) GetDataEngineUpgradeManager(name string) (*longhorn.DataEngineUpgradeManager, error) { + resultRO, err := s.GetDataEngineUpgradeManagerRO(name) + if err != nil { + return nil, err + } + // Cannot use cached object from lister + return resultRO.DeepCopy(), nil +} + +// UpdateDataEngineUpgradeManager updates the given Longhorn dataEngineUpgradeManager in the cluster DataEngineUpgradeManager CR and verifies update +func (s *DataStore) UpdateDataEngineUpgradeManager(upgradeManager *longhorn.DataEngineUpgradeManager) (*longhorn.DataEngineUpgradeManager, error) { + obj, err := s.lhClient.LonghornV1beta2().DataEngineUpgradeManagers(s.namespace).Update(context.TODO(), upgradeManager, metav1.UpdateOptions{}) + if err != nil { + return nil, err + } + verifyUpdate(upgradeManager.Name, obj, func(name string) (k8sruntime.Object, error) { + return s.GetDataEngineUpgradeManagerRO(name) + }) + return obj, nil +} + +// UpdateDataEngineUpgradeManagerStatus updates the given Longhorn dataEngineUpgradeManager status in the cluster DataEngineUpgradeManagers CR status and verifies update +func (s *DataStore) UpdateDataEngineUpgradeManagerStatus(upgradeManager *longhorn.DataEngineUpgradeManager) (*longhorn.DataEngineUpgradeManager, error) { + obj, err := s.lhClient.LonghornV1beta2().DataEngineUpgradeManagers(s.namespace).UpdateStatus(context.TODO(), upgradeManager, metav1.UpdateOptions{}) + if err != nil { + return nil, err + } + verifyUpdate(upgradeManager.Name, obj, func(name string) (k8sruntime.Object, error) { + return s.GetDataEngineUpgradeManagerRO(name) + }) + return obj, nil +} + +// RemoveFinalizerForDataEngineUpgradeManager will result in deletion if DeletionTimestamp was set +func (s *DataStore) RemoveFinalizerForDataEngineUpgradeManager(upgradeManager *longhorn.DataEngineUpgradeManager) error { + if !util.FinalizerExists(longhornFinalizerKey, upgradeManager) { + // finalizer already removed + return nil + } + if err := util.RemoveFinalizer(longhornFinalizerKey, upgradeManager); err != nil { + return err + } + _, err := s.lhClient.LonghornV1beta2().DataEngineUpgradeManagers(s.namespace).Update(context.TODO(), upgradeManager, metav1.UpdateOptions{}) + if err != nil { + // workaround `StorageError: invalid object, Code: 4` due to empty object + if upgradeManager.DeletionTimestamp != nil { + return nil + } + return errors.Wrapf(err, "unable to remove finalizer for dataEngineUpgradeManager %s", upgradeManager.Name) + } + return nil +} + +func (s *DataStore) listDataEngineUpgradeManagers(selector labels.Selector) (map[string]*longhorn.DataEngineUpgradeManager, error) { + list, err := s.dataEngineUpgradeManagerLister.DataEngineUpgradeManagers(s.namespace).List(selector) + if err != nil { + return nil, err + } + + itemMap := map[string]*longhorn.DataEngineUpgradeManager{} + for _, itemRO := range list { + // Cannot use cached object from lister + itemMap[itemRO.Name] = itemRO.DeepCopy() + } + return itemMap, nil +} + +// ListDataEngineUpgradeManagers returns an object contains all DataEngineUpgradeManagers for the given namespace +func (s *DataStore) ListDataEngineUpgradeManagers() (map[string]*longhorn.DataEngineUpgradeManager, error) { + return s.listDataEngineUpgradeManagers(labels.Everything()) +} + +// ListDataEngineUpgradeManagersRO returns a list of all UpgradeManagers for the given namespace +func (s *DataStore) ListDataEngineUpgradeManagersRO() ([]*longhorn.DataEngineUpgradeManager, error) { + return s.dataEngineUpgradeManagerLister.DataEngineUpgradeManagers(s.namespace).List(labels.Everything()) +} + +// DeleteDataEngineUpgradeManager won't result in immediately deletion since finalizer was set by default +func (s *DataStore) DeleteDataEngineUpgradeManager(upgradeManagerName string) error { + return s.lhClient.LonghornV1beta2().DataEngineUpgradeManagers(s.namespace).Delete(context.TODO(), upgradeManagerName, metav1.DeleteOptions{}) +} + +// GetOwnerReferencesForDataEngineUpgradeManager returns OwnerReference for the given DataEngineUpgradeManager name and UID +func GetOwnerReferencesForDataEngineUpgradeManager(upgradeManager *longhorn.DataEngineUpgradeManager) []metav1.OwnerReference { + return []metav1.OwnerReference{ + { + APIVersion: longhorn.SchemeGroupVersion.String(), + Kind: types.LonghornKindDataEngineUpgradeManager, + Name: upgradeManager.Name, + UID: upgradeManager.UID, + }, + } +} + +// CreateNodeDataEngineUpgrade creates a Longhorn NodeDataEngineUpgrade resource and verifies creation +func (s *DataStore) CreateNodeDataEngineUpgrade(nodeUpgrade *longhorn.NodeDataEngineUpgrade) (*longhorn.NodeDataEngineUpgrade, error) { + ret, err := s.lhClient.LonghornV1beta2().NodeDataEngineUpgrades(s.namespace).Create(context.TODO(), nodeUpgrade, metav1.CreateOptions{}) + if err != nil { + return nil, err + } + if SkipListerCheck { + return ret, nil + } + + obj, err := verifyCreation(ret.Name, "nodeDataEngineUpgrade", func(name string) (k8sruntime.Object, error) { + return s.GetNodeDataEngineUpgradeRO(name) + }) + if err != nil { + return nil, err + } + ret, ok := obj.(*longhorn.NodeDataEngineUpgrade) + if !ok { + return nil, fmt.Errorf("BUG: datastore: verifyCreation returned wrong type for nodeDataEngineUpgrade") + } + + return ret.DeepCopy(), nil +} + +// GetNodeDataEngineUpgradeRO returns the NodeDataEngineUpgrade with the given nodeDataEngineUpgrade name in the cluster +func (s *DataStore) GetNodeDataEngineUpgradeRO(upgradeName string) (*longhorn.NodeDataEngineUpgrade, error) { + return s.nodeDataEngineUpgradeLister.NodeDataEngineUpgrades(s.namespace).Get(upgradeName) +} + +// GetNodeDataEngineUpgrade returns a copy of NodeDataEngineUpgrade with the given nodeDataEngineUpgrade name in the cluster +func (s *DataStore) GetNodeDataEngineUpgrade(name string) (*longhorn.NodeDataEngineUpgrade, error) { + resultRO, err := s.GetNodeDataEngineUpgradeRO(name) + if err != nil { + return nil, err + } + // Cannot use cached object from lister + return resultRO.DeepCopy(), nil +} + +// UpdateNodeDataEngineUpgrade updates the given Longhorn nodeDataEngineUpgrade in the cluster NodeDataEngineUpgrade CR and verifies update +func (s *DataStore) UpdateNodeDataEngineUpgrade(upgrade *longhorn.NodeDataEngineUpgrade) (*longhorn.NodeDataEngineUpgrade, error) { + obj, err := s.lhClient.LonghornV1beta2().NodeDataEngineUpgrades(s.namespace).Update(context.TODO(), upgrade, metav1.UpdateOptions{}) + if err != nil { + return nil, err + } + verifyUpdate(upgrade.Name, obj, func(name string) (k8sruntime.Object, error) { + return s.GetNodeDataEngineUpgradeRO(name) + }) + return obj, nil +} + +// UpdateNodeDataEngineUpgradeStatus updates the given Longhorn nodeDataEngineUpgrade status in the cluster NodeDataEngineUpgrades CR status and verifies update +func (s *DataStore) UpdateNodeDataEngineUpgradeStatus(upgrade *longhorn.NodeDataEngineUpgrade) (*longhorn.NodeDataEngineUpgrade, error) { + obj, err := s.lhClient.LonghornV1beta2().NodeDataEngineUpgrades(s.namespace).UpdateStatus(context.TODO(), upgrade, metav1.UpdateOptions{}) + if err != nil { + return nil, err + } + verifyUpdate(upgrade.Name, obj, func(name string) (k8sruntime.Object, error) { + return s.GetNodeDataEngineUpgradeRO(name) + }) + return obj, nil +} + +// RemoveFinalizerForNodeDataEngineUpgrade will result in deletion if DeletionTimestamp was set +func (s *DataStore) RemoveFinalizerForNodeDataEngineUpgrade(upgrade *longhorn.NodeDataEngineUpgrade) error { + if !util.FinalizerExists(longhornFinalizerKey, upgrade) { + // finalizer already removed + return nil + } + if err := util.RemoveFinalizer(longhornFinalizerKey, upgrade); err != nil { + return err + } + _, err := s.lhClient.LonghornV1beta2().NodeDataEngineUpgrades(s.namespace).Update(context.TODO(), upgrade, metav1.UpdateOptions{}) + if err != nil { + // workaround `StorageError: invalid object, Code: 4` due to empty object + if upgrade.DeletionTimestamp != nil { + return nil + } + return errors.Wrapf(err, "unable to remove finalizer for nodeDataEngineUpgrade %s", upgrade.Name) + } + return nil +} + +func (s *DataStore) listNodeDataEngineUpgrades(selector labels.Selector) (map[string]*longhorn.NodeDataEngineUpgrade, error) { + list, err := s.nodeDataEngineUpgradeLister.NodeDataEngineUpgrades(s.namespace).List(selector) + if err != nil { + return nil, err + } + + itemMap := map[string]*longhorn.NodeDataEngineUpgrade{} + for _, itemRO := range list { + // Cannot use cached object from lister + itemMap[itemRO.Name] = itemRO.DeepCopy() + } + return itemMap, nil +} + +// ListNodeDataEngineUpgrades returns an object contains all NodeDataEngineUpgrades for the given namespace +func (s *DataStore) ListNodeDataEngineUpgrades() (map[string]*longhorn.NodeDataEngineUpgrade, error) { + return s.listNodeDataEngineUpgrades(labels.Everything()) +} + +// ListNodeDataEngineUpgradesByNodeRO returns a list of all NodeDataEngineUpgrades for the given node +func (s *DataStore) ListNodeDataEngineUpgradesByNodeRO(name string) ([]*longhorn.NodeDataEngineUpgrade, error) { + nodeSelector, err := getLonghornNodeSelector(name) + if err != nil { + return nil, err + } + return s.nodeDataEngineUpgradeLister.NodeDataEngineUpgrades(s.namespace).List(nodeSelector) +} + +// ListNodeDataEngineUpgradesRO returns a list of all NodeDataEngineUpgrades for the given namespace +func (s *DataStore) ListNodeDataEngineUpgradesRO() ([]*longhorn.NodeDataEngineUpgrade, error) { + return s.nodeDataEngineUpgradeLister.NodeDataEngineUpgrades(s.namespace).List(labels.Everything()) +} + +// DeleteNodeDataEngineUpgrade won't result in immediately deletion since finalizer was set by default +func (s *DataStore) DeleteNodeDataEngineUpgrade(upgradeName string) error { + return s.lhClient.LonghornV1beta2().NodeDataEngineUpgrades(s.namespace).Delete(context.TODO(), upgradeName, metav1.DeleteOptions{}) +} + // GetRunningInstanceManagerByNodeRO returns the running instance manager for the given node and data engine func (s *DataStore) GetRunningInstanceManagerByNodeRO(node string, dataEngine longhorn.DataEngineType) (*longhorn.InstanceManager, error) { // Trying to get the default instance manager first. @@ -5755,3 +6003,13 @@ func (s *DataStore) IsStorageNetworkForRWXVolume() (bool, error) { return types.IsStorageNetworkForRWXVolume(storageNetworkSetting, storageNetworkForRWXVolumeEnabled), nil } + +func (s *DataStore) IsNodeDataEngineUpgradeRequested(name string) (bool, error) { + node, err := s.GetNodeRO(name) + if err != nil { + return false, err + } + + dataEngineUpgradeRequestedCondition := types.GetCondition(node.Status.Conditions, longhorn.NodeConditionTypeDataEngineUpgradeRequested) + return dataEngineUpgradeRequestedCondition.Status == longhorn.ConditionStatusTrue, nil +}