Merge pull request #44401 from neersighted/swarmkit_revendor

vendor: github.com/moby/swarmkit/v2 v2.0.0-20221102165002-6341884e5fc9

vendor.mod

@@ -55,7 +55,7 @@ require (
 	github.com/moby/locker v1.0.1
 	github.com/moby/patternmatcher v0.5.0
 	github.com/moby/pubsub v1.0.0
-	github.com/moby/swarmkit/v2 v2.0.0-20220721174824-48dd89375d0a
+	github.com/moby/swarmkit/v2 v2.0.0-20221102165002-6341884e5fc9
 	github.com/moby/sys/mount v0.3.3
 	github.com/moby/sys/mountinfo v0.6.2
 	github.com/moby/sys/sequential v0.5.0

vendor.sum

@@ -777,8 +777,8 @@ github.com/moby/patternmatcher v0.5.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YO
 github.com/moby/pubsub v1.0.0 h1:jkp/imWsmJz2f6LyFsk7EkVeN2HxR/HTTOY8kHrsxfA=
 github.com/moby/pubsub v1.0.0/go.mod h1:bXSO+3h5MNXXCaEG+6/NlAIk7MMZbySZlnB+cUQhKKc=
 github.com/moby/spdystream v0.2.0/go.mod h1:f7i0iNDQJ059oMTcWxx8MA/zKFIuD/lY+0GqbN2Wy8c=
-github.com/moby/swarmkit/v2 v2.0.0-20220721174824-48dd89375d0a h1:gLcTxHH4egYVhMVFWRxvWsb79Ok4kfTt1/irZNyovUY=
-github.com/moby/swarmkit/v2 v2.0.0-20220721174824-48dd89375d0a/go.mod h1:/so6Lct4y1x14UprW/loFsOe6xoXVTlvh25V36ULXNQ=
+github.com/moby/swarmkit/v2 v2.0.0-20221102165002-6341884e5fc9 h1:d/XCmjx1zKZdzlBX90kSGDex7V2GE2jdGDr9nXYZg/Q=
+github.com/moby/swarmkit/v2 v2.0.0-20221102165002-6341884e5fc9/go.mod h1:/so6Lct4y1x14UprW/loFsOe6xoXVTlvh25V36ULXNQ=
 github.com/moby/sys/mount v0.3.3 h1:fX1SVkXFJ47XWDoeFW4Sq7PdQJnV2QIDZAqjNqgEjUs=
 github.com/moby/sys/mount v0.3.3/go.mod h1:PBaEorSNTLG5t/+4EgukEQVlAvVEc6ZjTySwKdqp5K0=
 github.com/moby/sys/mountinfo v0.4.0/go.mod h1:rEr8tzG/lsIZHBtN/JjGG+LMYx9eXgW2JI+6q0qou+A=

vendor/github.com/moby/swarmkit/v2/agent/csi/volumes.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/sirupsen/logrus"
 
@@ -16,6 +17,8 @@ import (
 	"github.com/moby/swarmkit/v2/volumequeue"
 )
 
+const CSI_CALL_TIMEOUT = 15 * time.Second
+
 // volumeState keeps track of the state of a volume on this node.
 type volumeState struct {
 	// volume is the actual VolumeAssignment for this volume
@@ -87,14 +90,35 @@ func (r *volumes) tryVolume(ctx context.Context, id string, attempt uint) {
 		return
 	}
 
+	// create a sub-context with a timeout. because we can only process one
+	// volume at a time, if we rely on the server-side or default timeout, we
+	// may be waiting a very long time for a particular volume to fail.
+	//
+	// TODO(dperny): there is almost certainly a more intelligent way to do
+	// this. For example, we could:
+	//
+	//   * Change code such that we can service volumes managed by different
+	//     plugins at the same time.
+	//   * Take longer timeouts when we don't have any other volumes in the
+	//     queue
+	//   * Have interruptible attempts, so that if we're taking longer
+	//     timeouts, we can abort them to service new volumes.
+	//
+	// These are too complicated to be worth the engineering effort at this
+	// time.
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, CSI_CALL_TIMEOUT)
+	// always gotta call the WithTimeout cancel
+	defer cancel()
+
 	if !vs.remove {
-		if err := r.publishVolume(ctx, vs.volume); err != nil {
-			log.G(ctx).WithError(err).Info("publishing volume failed")
+		if err := r.publishVolume(timeoutCtx, vs.volume); err != nil {
+			log.G(timeoutCtx).WithError(err).Info("publishing volume failed")
 			r.pendingVolumes.Enqueue(id, attempt+1)
 		}
 	} else {
-		if err := r.unpublishVolume(ctx, vs.volume); err != nil {
-			log.G(ctx).WithError(err).Info("upublishing volume failed")
+		if err := r.unpublishVolume(timeoutCtx, vs.volume); err != nil {
+			log.G(timeoutCtx).WithError(err).Info("upublishing volume failed")
 			r.pendingVolumes.Enqueue(id, attempt+1)
 		} else {
 			// if unpublishing was successful, then call the callback

vendor/github.com/moby/swarmkit/v2/ca/server.go

@@ -695,7 +695,7 @@ func (s *Server) UpdateRootCA(ctx context.Context, cluster *api.Cluster, reconci
 			log.G(ctx).Warn("no certificate expiration specified, using default")
 		}
 		// Attempt to update our local RootCA with the new parameters
-		updatedRootCA, err := RootCAFromAPI(ctx, rCA, expiry)
+		updatedRootCA, err := RootCAFromAPI(rCA, expiry)
 		if err != nil {
 			return errors.Wrap(err, "invalid Root CA object in cluster")
 		}
@@ -901,7 +901,7 @@ func isFinalState(status api.IssuanceStatus) bool {
 }
 
 // RootCAFromAPI creates a RootCA object from an api.RootCA object
-func RootCAFromAPI(ctx context.Context, apiRootCA *api.RootCA, expiry time.Duration) (RootCA, error) {
+func RootCAFromAPI(apiRootCA *api.RootCA, expiry time.Duration) (RootCA, error) {
 	var intermediates []byte
 	signingCert := apiRootCA.CACert
 	signingKey := apiRootCA.CAKey

vendor/github.com/moby/swarmkit/v2/manager/controlapi/cluster.go

@@ -119,7 +119,7 @@ func (s *Server) UpdateCluster(ctx context.Context, request *api.UpdateClusterRe
 		}
 		// This ensures that we have the current rootCA with which to generate tokens (expiration doesn't matter
 		// for generating the tokens)
-		rootCA, err := ca.RootCAFromAPI(ctx, &cluster.RootCA, ca.DefaultNodeCertExpiration)
+		rootCA, err := ca.RootCAFromAPI(&cluster.RootCA, ca.DefaultNodeCertExpiration)
 		if err != nil {
 			log.G(ctx).WithField(
 				"method", "(*controlapi.Server).UpdateCluster").WithError(err).Error("invalid cluster root CA")

vendor/github.com/moby/swarmkit/v2/manager/controlapi/volume.go

@@ -2,6 +2,7 @@ package controlapi
 
 import (
 	"context"
+	"reflect"
 	"strings"
 
 	"github.com/moby/swarmkit/v2/api"
@@ -94,17 +95,28 @@ func (s *Server) UpdateVolume(ctx context.Context, request *api.UpdateVolumeRequ
 		if request.Spec.Group != volume.Spec.Group {
 			return status.Errorf(codes.InvalidArgument, "Group cannot be updated")
 		}
-		if request.Spec.AccessibilityRequirements != volume.Spec.AccessibilityRequirements {
+		if !reflect.DeepEqual(request.Spec.AccessibilityRequirements, volume.Spec.AccessibilityRequirements) {
 			return status.Errorf(codes.InvalidArgument, "AccessibilityRequirements cannot be updated")
 		}
-		if request.Spec.Driver == nil || request.Spec.Driver.Name != volume.Spec.Driver.Name {
+		if !reflect.DeepEqual(request.Spec.Driver, volume.Spec.Driver) {
 			return status.Errorf(codes.InvalidArgument, "Driver cannot be updated")
 		}
-		if request.Spec.AccessMode.Scope != volume.Spec.AccessMode.Scope || request.Spec.AccessMode.Sharing != volume.Spec.AccessMode.Sharing {
+		if !reflect.DeepEqual(request.Spec.AccessMode, volume.Spec.AccessMode) {
 			return status.Errorf(codes.InvalidArgument, "AccessMode cannot be updated")
 		}
+		if !reflect.DeepEqual(request.Spec.Secrets, volume.Spec.Secrets) {
+			return status.Errorf(codes.InvalidArgument, "Secrets cannot be updated")
+		}
+		if !reflect.DeepEqual(request.Spec.CapacityRange, volume.Spec.CapacityRange) {
+			return status.Errorf(codes.InvalidArgument, "CapacityRange cannot be updated")
+		}
+
+		// to further guard against changing fields we're not allowed to, don't
+		// replace the entire spec. just replace the fields we are allowed to
+		// change
+		volume.Spec.Annotations.Labels = request.Spec.Annotations.Labels
+		volume.Spec.Availability = request.Spec.Availability
 
-		volume.Spec = *request.Spec
 		volume.Meta.Version = *request.VolumeVersion
 		if err := store.UpdateVolume(tx, volume); err != nil {
 			return err

vendor/github.com/moby/swarmkit/v2/manager/csi/manager.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/docker/go-events"
 	"github.com/sirupsen/logrus"
@@ -23,6 +24,10 @@ const (
 	// plugin interface is "docker.csicontroller/1.0". This gets only the CSI
 	// plugins with Controller capability.
 	DockerCSIPluginCap = "csicontroller"
+
+	// CSIRPCTimeout is the client-side timeout duration for RPCs to the CSI
+	// plugin.
+	CSIRPCTimeout = 15 * time.Second
 )
 
 type Manager struct {
@@ -149,11 +154,17 @@ func (vm *Manager) run(pctx context.Context) {
 // processVolumes encapuslates the logic for processing pending Volumes.
 func (vm *Manager) processVolume(ctx context.Context, id string, attempt uint) {
 	// set up log fields for a derrived context to pass to handleVolume.
-	dctx := log.WithFields(ctx, logrus.Fields{
+	logCtx := log.WithFields(ctx, logrus.Fields{
 		"volume.id": id,
 		"attempt":   attempt,
 	})
 
+	// Set a client-side timeout. Without this, one really long server-side
+	// timeout can block processing all volumes until it completes or fails.
+	dctx, cancel := context.WithTimeout(logCtx, CSIRPCTimeout)
+	// always gotta call the WithTimeout cancel
+	defer cancel()
+
 	err := vm.handleVolume(dctx, id)
 	// TODO(dperny): differentiate between retryable and non-retryable
 	// errors.

vendor/modules.txt

@@ -600,7 +600,7 @@ github.com/moby/patternmatcher
 # github.com/moby/pubsub v1.0.0
 ## explicit; go 1.19
 github.com/moby/pubsub
-# github.com/moby/swarmkit/v2 v2.0.0-20220721174824-48dd89375d0a
+# github.com/moby/swarmkit/v2 v2.0.0-20221102165002-6341884e5fc9
 ## explicit; go 1.17
 github.com/moby/swarmkit/v2/agent
 github.com/moby/swarmkit/v2/agent/configs