lightning: Add source dir existence check for s3 (#30674)

br/pkg/backup/client_test.go

@@ -241,7 +241,6 @@ func (r *testBackup) TestSendCreds(c *C) {
 	c.Assert(err, IsNil)
 	opts := &storage.ExternalStorageOptions{
 		SendCredentials: true,
-		SkipCheckPath:   true,
 	}
 	_, err = storage.New(r.ctx, backend, opts)
 	c.Assert(err, IsNil)
@@ -260,7 +259,6 @@ func (r *testBackup) TestSendCreds(c *C) {
 	c.Assert(err, IsNil)
 	opts = &storage.ExternalStorageOptions{
 		SendCredentials: false,
-		SkipCheckPath:   true,
 	}
 	_, err = storage.New(r.ctx, backend, opts)
 	c.Assert(err, IsNil)

br/pkg/lightning/lightning.go

@@ -277,6 +277,19 @@ func (l *Lightning) run(taskCtx context.Context, taskCfg *config.Config, g glue.
 		return errors.Annotate(err, "create storage failed")
 	}
 
+	// return expectedErr means at least meet one file
+	expectedErr := errors.New("Stop Iter")
+	walkErr := s.WalkDir(ctx, &storage.WalkOption{ListCount: 1}, func(string, int64) error {
+		// return an error when meet the first regular file to break the walk loop
+		return expectedErr
+	})
+	if !errors.ErrorEqual(walkErr, expectedErr) {
+		if walkErr == nil {
+			return errors.Errorf("data-source-dir '%s' doesn't exist or contains no files", taskCfg.Mydumper.SourceDir)
+		}
+		return errors.Annotatef(walkErr, "visit data-source-dir '%s' failed", taskCfg.Mydumper.SourceDir)
+	}
+
 	loadTask := log.L().Begin(zap.InfoLevel, "load data source")
 	var mdl *mydump.MDLoader
 	mdl, err = mydump.NewMyDumpLoaderWithStore(ctx, taskCfg, s)

br/pkg/storage/gcs.go

@@ -281,14 +281,6 @@ func newGCSStorage(ctx context.Context, gcs *backuppb.GCS, opts *ExternalStorage
 		// so we need find sst in slash directory
 		gcs.Prefix += "//"
 	}
-	// TODO remove it after BR remove cfg skip-check-path
-	if !opts.SkipCheckPath {
-		// check bucket exists
-		_, err = bucket.Attrs(ctx)
-		if err != nil {
-			return nil, errors.Annotatef(err, "gcs://%s/%s", gcs.Bucket, gcs.Prefix)
-		}
-	}
 	return &gcsStorage{gcs: gcs, bucket: bucket}, nil
 }
 

br/pkg/storage/s3.go

@@ -283,14 +283,6 @@ func newS3Storage(backend *backuppb.S3, opts *ExternalStorageOptions) (*S3Storag
 	}
 
 	c := s3.New(ses)
-	// TODO remove it after BR remove cfg skip-check-path
-	if !opts.SkipCheckPath {
-		err = checkS3Bucket(c, &qs)
-		if err != nil {
-			return nil, errors.Annotatef(berrors.ErrStorageInvalidConfig, "Bucket %s is not accessible: %v", qs.Bucket, err)
-		}
-	}
-
 	if len(qs.Prefix) > 0 && !strings.HasSuffix(qs.Prefix, "/") {
 		qs.Prefix += "/"
 	}

br/pkg/storage/s3_test.go

@@ -288,7 +288,6 @@ func (s *s3Suite) TestS3Storage(c *C) {
 		_, err := New(ctx, s3, &ExternalStorageOptions{
 			SendCredentials:  test.sendCredential,
 			CheckPermissions: test.hackPermission,
-			SkipCheckPath:    true,
 		})
 		if test.errReturn {
 			c.Assert(err, NotNil)
@@ -414,7 +413,7 @@ func (s *s3Suite) TestS3Storage(c *C) {
 func (s *s3Suite) TestS3URI(c *C) {
 	backend, err := ParseBackend("s3://bucket/prefix/", nil)
 	c.Assert(err, IsNil)
-	storage, err := New(context.Background(), backend, &ExternalStorageOptions{SkipCheckPath: true})
+	storage, err := New(context.Background(), backend, &ExternalStorageOptions{})
 	c.Assert(err, IsNil)
 	c.Assert(storage.URI(), Equals, "s3://bucket/prefix/")
 }

br/pkg/storage/storage.go

@@ -121,18 +121,6 @@ type ExternalStorageOptions struct {
 	// NoCredentials means that no cloud credentials are supplied to BR
 	NoCredentials bool
 
-	// SkipCheckPath marks whether to skip checking path's existence.
-	//
-	// This should only be set to true in testing, to avoid interacting with the
-	// real world.
-	// When this field is false (i.e. path checking is enabled), the New()
-	// function will ensure the path referred by the backend exists by
-	// recursively creating the folders. This will also throw an error if such
-	// operation is impossible (e.g. when the bucket storing the path is missing).
-
-	// deprecated: use checkPermissions and specify the checkPermission instead.
-	SkipCheckPath bool
-
 	// HTTPClient to use. The created storage may ignore this field if it is not
 	// directly using HTTP (e.g. the local storage).
 	HTTPClient *http.Client
@@ -148,7 +136,6 @@ type ExternalStorageOptions struct {
 func Create(ctx context.Context, backend *backuppb.StorageBackend, sendCreds bool) (ExternalStorage, error) {
 	return New(ctx, backend, &ExternalStorageOptions{
 		SendCredentials: sendCreds,
-		SkipCheckPath:   false,
 		HTTPClient:      nil,
 	})
 }

br/pkg/task/backup.go

@@ -257,7 +257,6 @@ func RunBackup(c context.Context, g glue.Glue, cmdName string, cfg *BackupConfig
 	opts := storage.ExternalStorageOptions{
 		NoCredentials:   cfg.NoCreds,
 		SendCredentials: cfg.SendCreds,
-		SkipCheckPath:   cfg.SkipCheckPath,
 	}
 	if err = client.SetStorage(ctx, u, &opts); err != nil {
 		return errors.Trace(err)

br/pkg/task/backup_raw.go

@@ -150,7 +150,6 @@ func RunBackupRaw(c context.Context, g glue.Glue, cmdName string, cfg *RawKvConf
 	opts := storage.ExternalStorageOptions{
 		NoCredentials:   cfg.NoCreds,
 		SendCredentials: cfg.SendCreds,
-		SkipCheckPath:   cfg.SkipCheckPath,
 	}
 	if err = client.SetStorage(ctx, u, &opts); err != nil {
 		return errors.Trace(err)

br/pkg/task/common.go

@@ -367,6 +367,10 @@ func (cfg *Config) ParseFromFlags(flags *pflag.FlagSet) error {
 	if cfg.SkipCheckPath, err = flags.GetBool(flagSkipCheckPath); err != nil {
 		return errors.Trace(err)
 	}
+	if cfg.SkipCheckPath {
+		log.L().Info("--skip-check-path is deprecated, need explicitly set it anymore")
+	}
+
 	return cfg.normalizePDURLs()
 }
 
@@ -431,7 +435,6 @@ func storageOpts(cfg *Config) *storage.ExternalStorageOptions {
 	return &storage.ExternalStorageOptions{
 		NoCredentials:   cfg.NoCreds,
 		SendCredentials: cfg.SendCreds,
-		SkipCheckPath:   cfg.SkipCheckPath,
 	}
 }
 

br/pkg/task/restore.go

@@ -221,7 +221,6 @@ func RunRestore(c context.Context, g glue.Glue, cmdName string, cfg *RestoreConf
 	opts := storage.ExternalStorageOptions{
 		NoCredentials:   cfg.NoCreds,
 		SendCredentials: cfg.SendCreds,
-		SkipCheckPath:   cfg.SkipCheckPath,
 	}
 	if err = client.SetStorage(ctx, u, &opts); err != nil {
 		return errors.Trace(err)

br/tests/lightning_s3/run.sh

@@ -61,6 +61,20 @@ _EOF_
 run_sql "DROP DATABASE IF EXISTS $DB;"
 run_sql "DROP TABLE IF EXISTS $DB.$TABLE;"
 
+# test not exist path
+rm -f $TEST_DIR/lightning.log
+SOURCE_DIR="s3://$BUCKET/not-exist-path?endpoint=http%3A//127.0.0.1%3A9900&access_key=$MINIO_ACCESS_KEY&secret_access_key=$MINIO_SECRET_KEY&force_path_style=true"
+! run_lightning -d $SOURCE_DIR --backend local 2> /dev/null
+grep -Eq "data-source-dir .* doesn't exist or contains no files" $TEST_DIR/lightning.log
+
+# test empty dir
+rm -f $TEST_DIR/lightning.log
+emptyPath=empty-bucket/empty-path
+mkdir -p $DBPATH/$emptyPath
+SOURCE_DIR="s3://$emptyPath/not-exist-path?endpoint=http%3A//127.0.0.1%3A9900&access_key=$MINIO_ACCESS_KEY&secret_access_key=$MINIO_SECRET_KEY&force_path_style=true"
+! run_lightning -d $SOURCE_DIR --backend local 2> /dev/null
+grep -Eq "data-source-dir .* doesn't exist or contains no files" $TEST_DIR/lightning.log
+
 SOURCE_DIR="s3://$BUCKET/?endpoint=http%3A//127.0.0.1%3A9900&access_key=$MINIO_ACCESS_KEY&secret_access_key=$MINIO_SECRET_KEY&force_path_style=true"
 run_lightning -d $SOURCE_DIR --backend local 2> /dev/null
 run_sql "SELECT count(*), sum(i) FROM \`$DB\`.$TABLE"