Adds ONTAP HTTP rate limiting

Author: reederc42

cli/k8s_client/yaml_factory.go

@@ -783,7 +783,7 @@ spec:
         - "--csi-address=$(ADDRESS)"
         - "--retry-interval-start=8s"
         - "--retry-interval-max=30s"
-        - "--worker-threads=5"
+        - "--worker-threads=100"
         {K8S_API_CLIENT_SIDECAR_THROTTLE}
         {CSI_SIDECAR_PROVISIONER_RESOURCES}
         env:
@@ -803,7 +803,7 @@ spec:
         - "--v={SIDECAR_LOG_LEVEL}"
         - "--timeout=60s"
         - "--retry-interval-start=10s"
-        - "--worker-threads=10"
+        - "--worker-threads=100"
         - "--csi-address=$(ADDRESS)"
         {K8S_API_CLIENT_SIDECAR_THROTTLE}
         {CSI_SIDECAR_ATTACHER_RESOURCES}
@@ -823,7 +823,7 @@ spec:
         args:
         - "--v={SIDECAR_LOG_LEVEL}"
         - "--timeout=300s"
-        - "--workers=10"
+        - "--workers=100"
         - "--csi-address=$(ADDRESS)"
         {K8S_API_CLIENT_SIDECAR_THROTTLE}
         {CSI_SIDECAR_RESIZER_RESOURCES}
@@ -843,7 +843,7 @@ spec:
         args:
         - "--v={SIDECAR_LOG_LEVEL}"
         - "--timeout=300s"
-        - "--worker-threads=10"
+        - "--worker-threads=100"
         - "--csi-address=$(ADDRESS)"
         {FEATURE_GATES_CSI_SNAPSHOTTER}
         {K8S_API_CLIENT_SIDECAR_THROTTLE}

config/config.go

@@ -247,10 +247,10 @@ const (
 	UnknownVolumeType VolumeType = ""
 
 	/* Driver-related constants */
-	DefaultSolidFireVAG      = OrchestratorName
-	UnknownDriver            = "UnknownDriver"
-	StorageAPITimeoutSeconds = 90
-	SANResizeDelta           = 50000000 // 50mb
+	DefaultSolidFireVAG = OrchestratorName
+	UnknownDriver       = "UnknownDriver"
+	StorageAPITimeout   = 90 * time.Second
+	SANResizeDelta      = 50000000 // 50mb
 
 	// Storage driver names specified in the config file, etc.
 	OntapNASStorageDriverName          = "ontap-nas"

go.mod

@@ -57,7 +57,7 @@ require (
 	golang.org/x/crypto v0.40.0 // github.com/golang/crypto
 	golang.org/x/net v0.42.0 // github.com/golang/net
 	golang.org/x/oauth2 v0.30.0 // github.com/golang/oauth2
-	golang.org/x/sync v0.16.0 // indirect
+	golang.org/x/sync v0.16.0
 	golang.org/x/sys v0.34.0 // github.com/golang/sys
 	golang.org/x/text v0.27.0 // github.com/golang/text
 	golang.org/x/time v0.11.0 // github.com/golang/time

mocks/mock_storage_drivers/mock_ontap/mock_api.go

@@ -8,9 +8,16 @@ import (
 	"encoding/gob"
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
 	"reflect"
 	"strconv"
 	"strings"
+	"sync"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"golang.org/x/sync/semaphore"
 
 	trident "github.com/netapp/trident/config"
 	. "github.com/netapp/trident/logging"
@@ -399,3 +406,94 @@ func RemoveSCSIDeviceByPublishInfo(ctx context.Context, publishInfo *tridentmode
 		}
 	}
 }
+
+var (
+	semaphoresLock    = sync.Mutex{}
+	semaphores        = make(map[string]*rcSem)
+	ONTAPRequestLimit = 20
+)
+
+const (
+	initialInterval = 1 * time.Second
+	multiplier      = 1.414 // approx sqrt(2)
+	maxInterval     = 6 * time.Second
+	randomFactor    = 0.1
+)
+
+// rcSem is a reference-counted semaphore.
+type rcSem struct {
+	sem  *semaphore.Weighted
+	refs int
+}
+
+// NewSemaphore returns a named semaphore, creating it if it does not already exist.
+// The semaphore is initialized with the specified maxConcurrent value only the first time it is created.
+// Each call to NewSemaphore should be matched with a call to FreeSemaphore to allow proper cleanup.
+func NewSemaphore(name string, maxConcurrent int) *semaphore.Weighted {
+	semaphoresLock.Lock()
+	defer semaphoresLock.Unlock()
+
+	s, exists := semaphores[name]
+	if !exists {
+		s = &rcSem{
+			sem:  semaphore.NewWeighted(int64(maxConcurrent)),
+			refs: 0,
+		}
+		semaphores[name] = s
+	}
+	s.refs++
+	return s.sem
+}
+
+func FreeSemaphore(name string) {
+	semaphoresLock.Lock()
+	defer semaphoresLock.Unlock()
+
+	counter, exists := semaphores[name]
+	if exists {
+		counter.refs--
+		if counter.refs <= 0 {
+			delete(semaphores, name)
+		}
+	}
+}
+
+type LimitedRetryTransport struct {
+	base http.RoundTripper
+	sem  *semaphore.Weighted
+	b    *backoff.ExponentialBackOff
+}
+
+// NewLimitedRetryTransport wraps a base transport to limit the number of concurrent requests and add retries
+// on io.EOF errors.
+func NewLimitedRetryTransport(sem *semaphore.Weighted, base http.RoundTripper) *LimitedRetryTransport {
+	b := backoff.NewExponentialBackOff()
+	b.InitialInterval = initialInterval
+	b.Multiplier = multiplier
+	b.MaxInterval = maxInterval
+	b.RandomizationFactor = randomFactor
+	return &LimitedRetryTransport{
+		base: base,
+		sem:  sem,
+		b:    b,
+	}
+}
+
+func (lrt *LimitedRetryTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	var resp *http.Response
+	f := func() error {
+		if err := lrt.sem.Acquire(req.Context(), 1); err != nil {
+			return backoff.Permanent(err)
+		} else {
+			defer lrt.sem.Release(1)
+		}
+		r, err := lrt.base.RoundTrip(req)
+		resp = r
+		if err != nil && !errors.Is(err, io.EOF) {
+			return backoff.Permanent(err)
+		}
+		return err
+	}
+	lrt.b.Reset()
+	return resp, backoff.Retry(f, lrt.b)
+}