add timeouts around s3 operations

Signed-off-by: Brian Downs <brian.downs@gmail.com>

pkg/cli/cmds/root.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"runtime"
@@ -20,6 +21,8 @@ var (
 	}
 )
 
+var ErrCommandNoArgs = errors.New("this command does not take any arguments")
+
 func init() {
 	// hack - force "file,dns" lookup order if go dns is used
 	if os.Getenv("RES_OPTIONS") == "" {

pkg/cli/etcdsnapshot/etcd_snapshot.go

@@ -18,21 +18,21 @@ import (
 
 // commandSetup setups up common things needed
 // for each etcd command.
-func commandSetup(app *cli.Context) error {
+func commandSetup(app *cli.Context, cfg *cmds.Server) (string, error) {
 	gspt.SetProcTitle(os.Args[0])
 
 	nodeName := app.String("node-name")
 	if nodeName == "" {
 		h, err := os.Hostname()
 		if err != nil {
-			return err
+			return "", err
 		}
 		nodeName = h
 	}
 
 	os.Setenv("NODE_NAME", nodeName)
 
-	return nil
+	return server.ResolveDataDir(cfg.DataDir)
 }
 
 func Run(app *cli.Context) error {
@@ -43,17 +43,13 @@ func Run(app *cli.Context) error {
 }
 
 func run(app *cli.Context, cfg *cmds.Server) error {
-	if err := commandSetup(app); err != nil {
+	dataDir, err := commandSetup(app, cfg)
+	if err != nil {
 		return err
 	}
 
 	if len(app.Args()) > 0 {
-		return errors.New("the etcd-snapshot command does not take any arguments")
-	}
-
-	dataDir, err := server.ResolveDataDir(cfg.DataDir)
-	if err != nil {
-		return err
+		return cmds.ErrCommandNoArgs
 	}
 
 	var serverConfig server.Config
@@ -112,7 +108,8 @@ func Delete(app *cli.Context) error {
 }
 
 func delete(app *cli.Context, cfg *cmds.Server) error {
-	if err := commandSetup(app); err != nil {
+	dataDir, err := commandSetup(app, cfg)
+	if err != nil {
 		return err
 	}
 
@@ -121,11 +118,6 @@ func delete(app *cli.Context, cfg *cmds.Server) error {
 		return errors.New("no snapshots given for removal")
 	}
 
-	dataDir, err := server.ResolveDataDir(cfg.DataDir)
-	if err != nil {
-		return err
-	}
-
 	var serverConfig server.Config
 	serverConfig.DisableAgent = true
 	serverConfig.ControlConfig.DataDir = dataDir

pkg/etcd/etcd.go

@@ -991,25 +991,36 @@ func (e *ETCD) DeleteSnapshots(ctx context.Context, snapshots []string) error {
 
 			objectsCh := make(chan minio.ObjectInfo)
 
-			go func() {
-				for obj := range e.s3.client.ListObjects(ctx, e.config.EtcdS3BucketName, minio.ListObjectsOptions{}) {
-					if obj.Err != nil {
-						logrus.Error(obj.Err)
-						continue
-					}
+			toCtx, cancel := context.WithTimeout(ctx, time.Second*30)
+			defer cancel()
 
-					// iterate through the given snapshots and only
-					// add them to the channel for remove if they're
-					// actually found from the bucket listing.
-					for _, snapshot := range snapshots {
-						if snapshot == obj.Key {
-							objectsCh <- obj
+			go func(ctx context.Context) {
+				defer close(objectsCh)
+
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					default:
+						for obj := range e.s3.client.ListObjects(ctx, e.config.EtcdS3BucketName, minio.ListObjectsOptions{}) {
+							if obj.Err != nil {
+								logrus.Error(obj.Err)
+								continue
+							}
+
+							// iterate through the given snapshots and only
+							// add them to the channel for remove if they're
+							// actually found from the bucket listing.
+							for _, snapshot := range snapshots {
+								if snapshot == obj.Key {
+									objectsCh <- obj
+								}
+							}
 						}
 					}
 				}
 
-				close(objectsCh)
-			}()
+			}(toCtx)
 
 			for roErr := range e.s3.client.RemoveObjects(ctx, e.config.EtcdS3BucketName, objectsCh, minio.RemoveObjectsOptions{}) {
 				logrus.Errorf("Error detected during deletion: %v", roErr)

pkg/etcd/s3.go

@@ -14,6 +14,7 @@ import (
 	"path/filepath"
 	"sort"
 	"strings"
+	"time"
 
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
@@ -22,6 +23,8 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+const defaultS3OpTimeout = time.Second * 30
+
 // s3 maintains state for S3 functionality.
 type s3 struct {
 	config *config.Control
@@ -61,7 +64,11 @@ func newS3(ctx context.Context, config *config.Control) (*s3, error) {
 	}
 
 	logrus.Infof("Checking if S3 bucket %s exists", config.EtcdS3BucketName)
-	exists, err := c.BucketExists(ctx, config.EtcdS3BucketName)
+
+	toCtx, cancel := context.WithTimeout(ctx, defaultS3OpTimeout)
+	defer cancel()
+
+	exists, err := c.BucketExists(toCtx, config.EtcdS3BucketName)
 	if err != nil {
 		return nil, err
 	}
@@ -87,11 +94,13 @@ func (s *s3) upload(ctx context.Context, snapshot string) error {
 		snapshotFileName = basename
 	}
 
+	toCtx, cancel := context.WithTimeout(ctx, defaultS3OpTimeout)
+	defer cancel()
 	opts := minio.PutObjectOptions{
 		ContentType: "application/zip",
 		NumThreads:  2,
 	}
-	if _, err := s.client.FPutObject(ctx, s.config.EtcdS3BucketName, snapshotFileName, snapshot, opts); err != nil {
+	if _, err := s.client.FPutObject(toCtx, s.config.EtcdS3BucketName, snapshotFileName, snapshot, opts); err != nil {
 		logrus.Errorf("Error received in attempt to upload snapshot to S3: %s", err)
 	}
 
@@ -109,7 +118,10 @@ func (s *s3) download(ctx context.Context) error {
 	}
 
 	logrus.Debugf("retrieving snapshot: %s", remotePath)
-	r, err := s.client.GetObject(ctx, s.config.EtcdS3BucketName, remotePath, minio.GetObjectOptions{})
+	toCtx, cancel := context.WithTimeout(ctx, defaultS3OpTimeout)
+	defer cancel()
+
+	r, err := s.client.GetObject(toCtx, s.config.EtcdS3BucketName, remotePath, minio.GetObjectOptions{})
 	if err != nil {
 		return nil
 	}
@@ -160,11 +172,14 @@ func (s *s3) snapshotPrefix() string {
 func (s *s3) snapshotRetention(ctx context.Context) error {
 	var snapshotFiles []minio.ObjectInfo
 
+	toCtx, cancel := context.WithTimeout(ctx, defaultS3OpTimeout)
+	defer cancel()
+
 	loo := minio.ListObjectsOptions{
 		Recursive: true,
 		Prefix:    s.snapshotPrefix(),
 	}
-	for info := range s.client.ListObjects(ctx, s.config.EtcdS3BucketName, loo) {
+	for info := range s.client.ListObjects(toCtx, s.config.EtcdS3BucketName, loo) {
 		if info.Err != nil {
 			return info.Err
 		}