fix: modify methods for updated new libraries

- Bump rancher/wranger to v3
- Add necessary methods for the library `container-storage-interface/spec` v1.10.0
- Deprecated methods of the library `k8s.io/client-go/util/workqueue`

Signed-off-by: James Lu <james.lu@suse.com>

app/daemon.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/gorilla/handlers"
 	"github.com/pkg/errors"
-	"github.com/rancher/wrangler/pkg/signals"
+	"github.com/rancher/wrangler/v3/pkg/signals"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"

controller/base_controller.go

@@ -18,16 +18,17 @@ var (
 type baseController struct {
 	name   string
 	logger *logrus.Entry
-	queue  workqueue.RateLimitingInterface
+	queue  workqueue.TypedRateLimitingInterface[any]
 }
 
 func newBaseController(name string, logger logrus.FieldLogger) *baseController {
+	nameConfig := workqueue.TypedRateLimitingQueueConfig[any]{Name: name}
 	return newBaseControllerWithQueue(name, logger,
-		workqueue.NewNamedRateLimitingQueue(EnhancedDefaultControllerRateLimiter(), name))
+		workqueue.NewTypedRateLimitingQueueWithConfig[any](EnhancedDefaultControllerRateLimiter(), nameConfig))
 }
 
 func newBaseControllerWithQueue(name string, logger logrus.FieldLogger,
-	queue workqueue.RateLimitingInterface) *baseController {
+	queue workqueue.TypedRateLimitingInterface[any]) *baseController {
 	c := &baseController{
 		name:   name,
 		logger: logger.WithField("controller", name),

controller/controller_manager.go

@@ -310,11 +310,11 @@ func isControllerResponsibleFor(controllerID string, ds *datastore.DataStore, na
 
 // EnhancedDefaultControllerRateLimiter is an enhanced version of workqueue.DefaultControllerRateLimiter()
 // See https://github.com/longhorn/longhorn/issues/1058 for details
-func EnhancedDefaultControllerRateLimiter() workqueue.RateLimiter {
-	return workqueue.NewMaxOfRateLimiter(
-		workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond, 1000*time.Second),
+func EnhancedDefaultControllerRateLimiter() workqueue.TypedRateLimiter[any] {
+	return workqueue.NewTypedMaxOfRateLimiter[any](
+		workqueue.NewTypedItemExponentialFailureRateLimiter[any](5*time.Millisecond, 1000*time.Second),
 		// 100 qps, 1000 bucket size
-		&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(100), 1000)},
+		&workqueue.TypedBucketRateLimiter[any]{Limiter: rate.NewLimiter(rate.Limit(100), 1000)},
 	)
 }
 

controller/monitor/snapshot_monitor.go

@@ -70,8 +70,8 @@ type SnapshotMonitor struct {
 
 	checkScheduler *gocron.Scheduler
 
-	snapshotChangeEventQueue workqueue.Interface
-	snapshotCheckTaskQueue   workqueue.RateLimitingInterface
+	snapshotChangeEventQueue workqueue.TypedInterface[any]
+	snapshotCheckTaskQueue   workqueue.TypedRateLimitingInterface[any]
 
 	inProgressSnapshotCheckTasks     map[string]struct{}
 	inProgressSnapshotCheckTasksLock sync.RWMutex
@@ -86,7 +86,7 @@ type SnapshotMonitor struct {
 }
 
 func NewSnapshotMonitor(logger logrus.FieldLogger, ds *datastore.DataStore, nodeName string, eventRecorder record.EventRecorder,
-	snapshotChangeEventQueue workqueue.Interface, syncCallback func(key string)) (*SnapshotMonitor, error) {
+	snapshotChangeEventQueue workqueue.TypedInterface[any], syncCallback func(key string)) (*SnapshotMonitor, error) {
 
 	ctx, quit := context.WithCancel(context.Background())
 
@@ -101,9 +101,9 @@ func NewSnapshotMonitor(logger logrus.FieldLogger, ds *datastore.DataStore, node
 
 		snapshotChangeEventQueue: snapshotChangeEventQueue,
 
-		snapshotCheckTaskQueue: workqueue.NewRateLimitingQueue(workqueue.NewMaxOfRateLimiter(
-			workqueue.NewItemExponentialFailureRateLimiter(1*time.Second, 1000*time.Second),
-			&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
+		snapshotCheckTaskQueue: workqueue.NewTypedRateLimitingQueue[any](workqueue.NewTypedMaxOfRateLimiter[any](
+			workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Second, 1000*time.Second),
+			&workqueue.TypedBucketRateLimiter[any]{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
 		)),
 
 		inProgressSnapshotCheckTasks: map[string]struct{}{},

controller/node_controller.go

@@ -63,7 +63,7 @@ type NodeController struct {
 	diskMonitor monitor.Monitor
 
 	snapshotMonitor              monitor.Monitor
-	snapshotChangeEventQueue     workqueue.Interface
+	snapshotChangeEventQueue     workqueue.TypedInterface[any]
 	snapshotChangeEventQueueLock sync.Mutex
 
 	ds *datastore.DataStore
@@ -102,7 +102,7 @@ func NewNodeController(
 
 		topologyLabelsChecker: util.IsKubernetesVersionAtLeast,
 
-		snapshotChangeEventQueue: workqueue.New(),
+		snapshotChangeEventQueue: workqueue.NewTyped[any](),
 	}
 
 	nc.scheduler = scheduler.NewReplicaScheduler(ds)

controller/system_rollout_controller.go

@@ -169,12 +169,13 @@ func NewSystemRolloutController(
 		Interface: v1core.New(kubeClient.CoreV1().RESTClient()).Events(""),
 	})
 
+	nameConfig := workqueue.TypedRateLimitingQueueConfig[any]{Name: SystemRolloutControllerName}
 	c := &SystemRolloutController{
 		baseController: newBaseControllerWithQueue(SystemRolloutControllerName, logger,
-			workqueue.NewNamedRateLimitingQueue(workqueue.NewMaxOfRateLimiter(
-				workqueue.NewItemExponentialFailureRateLimiter(100*time.Millisecond, 2*time.Second),
-				&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(100), 1000)},
-			), SystemRolloutControllerName),
+			workqueue.NewTypedRateLimitingQueueWithConfig[any](workqueue.NewTypedMaxOfRateLimiter[any](
+				workqueue.NewTypedItemExponentialFailureRateLimiter[any](100*time.Millisecond, 2*time.Second),
+				&workqueue.TypedBucketRateLimiter[any]{Limiter: rate.NewLimiter(rate.Limit(100), 1000)},
+			), nameConfig),
 		),
 		controllerID: controllerID,
 		stopCh:       stopCh,

controller/uninstall_controller.go

@@ -75,12 +75,13 @@ func NewUninstallController(
 	kubeClient clientset.Interface,
 	extensionsClient apiextensionsclientset.Interface,
 ) (*UninstallController, error) {
+	nameConfig := workqueue.TypedRateLimitingQueueConfig[any]{Name: "longhorn-uninstall"}
 	c := &UninstallController{
 		baseController: newBaseControllerWithQueue("longhorn-uninstall", logger,
-			workqueue.NewNamedRateLimitingQueue(workqueue.NewMaxOfRateLimiter(
-				workqueue.NewItemExponentialFailureRateLimiter(100*time.Millisecond, 2*time.Second),
-				&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(100), 1000)},
-			), "longhorn-uninstall"),
+			workqueue.NewTypedRateLimitingQueueWithConfig[any](workqueue.NewTypedMaxOfRateLimiter[any](
+				workqueue.NewTypedItemExponentialFailureRateLimiter[any](100*time.Millisecond, 2*time.Second),
+				&workqueue.TypedBucketRateLimiter[any]{Limiter: rate.NewLimiter(rate.Limit(100), 1000)},
+			), nameConfig),
 		),
 		namespace: namespace,
 		force:     force,

csi/controller_server.go

@@ -46,6 +46,7 @@ const (
 )
 
 type ControllerServer struct {
+	csi.UnimplementedControllerServer
 	apiClient   *longhornclient.RancherClient
 	nodeID      string
 	caps        []*csi.ControllerServiceCapability
@@ -1380,7 +1381,7 @@ func (cs *ControllerServer) waitForSnapshotToBeReady(snapshotName, volumeName st
 
 func (cs *ControllerServer) ControllerModifyVolume(ctx context.Context, req *csi.ControllerModifyVolumeRequest) (*csi.ControllerModifyVolumeResponse, error) {
 	log := cs.log.WithFields(logrus.Fields{"function": "ControllerModifyVolume"})
-	log.Infof("ControllerModifyVolume: called with args %+v", *req)
+	log.Infof("ControllerModifyVolume: called with args %v", req)
 
 	return nil, status.Error(codes.Unimplemented, "")
 }

csi/identity_server.go

@@ -7,6 +7,7 @@ import (
 )
 
 type IdentityServer struct {
+	csi.UnimplementedIdentityServer
 	driverName string
 	version    string
 }

csi/node_server.go

@@ -48,6 +48,7 @@ var supportedFs = map[string]fsParameters{
 }
 
 type NodeServer struct {
+	csi.UnimplementedNodeServer
 	apiClient *longhornclient.RancherClient
 	nodeID    string
 	caps      []*csi.NodeServiceCapability

go.mod

@@ -72,7 +72,7 @@ require (
 	github.com/prometheus/client_golang v1.20.2
 	github.com/rancher/dynamiclistener v0.6.0
 	github.com/rancher/go-rancher v0.1.1-0.20220412083059-ff12399dd57b
-	github.com/rancher/wrangler v1.1.2
+	github.com/rancher/wrangler/v3 v3.0.0
 	github.com/robfig/cron v1.2.0
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.9.0
@@ -114,7 +114,6 @@ require (
 	github.com/opencontainers/runc v1.1.13 // indirect
 	github.com/opencontainers/runtime-spec v1.1.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
-	github.com/rancher/wrangler/v3 v3.0.0 // indirect
 	github.com/shirou/gopsutil/v3 v3.24.5 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect

go.sum

@@ -233,8 +233,6 @@ github.com/rancher/go-rancher v0.1.1-0.20220412083059-ff12399dd57b h1:so40GMVZOZ
 github.com/rancher/go-rancher v0.1.1-0.20220412083059-ff12399dd57b/go.mod h1:7oQvGNiJsGvrUgB+7AH8bmdzuR0uhULfwKb43Ht0hUk=
 github.com/rancher/lasso v0.0.0-20240809125800-8da6f11865d5 h1:qlVhaHTT7wwrI5+AGdkYHpveuoe8Ot4TdQr7LtxmVSk=
 github.com/rancher/lasso v0.0.0-20240809125800-8da6f11865d5/go.mod h1:Efx/+BbH3ivmnTPLu5cA3Gc9wT5oyGS0LBcqEuYTx+A=
-github.com/rancher/wrangler v1.1.2 h1:oXbXo9k7y/H4drUpb4RM1c++vT9O3rpoNEfyusGykiU=
-github.com/rancher/wrangler v1.1.2/go.mod h1:2k9MyhlBdjcutcBGoOJSUAz0HgDAXnMjv81d3n/AaQc=
 github.com/rancher/wrangler/v3 v3.0.0 h1:IHHCA+vrghJDPxjtLk4fmeSCFhNe9fFzLFj3m2B0YpA=
 github.com/rancher/wrangler/v3 v3.0.0/go.mod h1:Dfckuuq7MJk2JWVBDywRlZXMxEyPxHy4XqGrPEzu5Eg=
 github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=

util/client/client.go

@@ -6,8 +6,8 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
-	wranglerClients "github.com/rancher/wrangler/pkg/clients"
-	wranglerSchemes "github.com/rancher/wrangler/pkg/schemes"
+	wranglerClients "github.com/rancher/wrangler/v3/pkg/clients"
+	wranglerSchemes "github.com/rancher/wrangler/v3/pkg/schemes"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/clientcmd"