diff --git a/pkg/cli/cmds/root.go b/pkg/cli/cmds/root.go index ebd536685c..13d523b430 100644 --- a/pkg/cli/cmds/root.go +++ b/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") == "" { diff --git a/pkg/cli/etcdsnapshot/etcd_snapshot.go b/pkg/cli/etcdsnapshot/etcd_snapshot.go index eb1081fa88..051d80625c 100644 --- a/pkg/cli/etcdsnapshot/etcd_snapshot.go +++ b/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 diff --git a/pkg/etcd/etcd.go b/pkg/etcd/etcd.go index 8b271ec357..f657821345 100644 --- a/pkg/etcd/etcd.go +++ b/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) diff --git a/pkg/etcd/s3.go b/pkg/etcd/s3.go index 7b328dc5ab..378dbe0d03 100644 --- a/pkg/etcd/s3.go +++ b/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 }