engine(externalresource): configure storage only on master

deployments/engine/docker-compose/3m3e_with_s3.yaml

@@ -75,7 +75,7 @@ services:
     ports:
       - "11241:10241"
     volumes:
-      - ./config/executor_with_s3.toml:/config.toml
+      - ./config/executor.toml:/config.toml
       - /tmp/tiflow_engine_test:/log
     command:
       - "/tiflow"
@@ -99,7 +99,7 @@ services:
     ports:
       - "11242:10241"
     volumes:
-      - ./config/executor_with_s3.toml:/config.toml
+      - ./config/executor.toml:/config.toml
       - /tmp/tiflow_engine_test:/log
     command:
       - "/tiflow"
@@ -123,7 +123,7 @@ services:
     ports:
       - "11243:10241"
     volumes:
-      - ./config/executor_with_s3.toml:/config.toml
+      - ./config/executor.toml:/config.toml
       - /tmp/tiflow_engine_test:/log
     command:
       - "/tiflow"

deployments/engine/docker-compose/3m3e_with_tls.yaml

@@ -75,7 +75,7 @@ services:
     ports:
       - "11241:10241"
     volumes:
-      - ./config/executor_with_s3.toml:/config.toml
+      - ./config/executor.toml:/config.toml
       - /tmp/tiflow_engine_test:/log
     command:
       - "/tiflow"
@@ -101,7 +101,7 @@ services:
     ports:
       - "11242:10241"
     volumes:
-      - ./config/executor_with_s3.toml:/config.toml
+      - ./config/executor.toml:/config.toml
       - /tmp/tiflow_engine_test:/log
     command:
       - "/tiflow"
@@ -127,7 +127,7 @@ services:
     ports:
       - "11243:10241"
     volumes:
-      - ./config/executor_with_s3.toml:/config.toml
+      - ./config/executor.toml:/config.toml
       - /tmp/tiflow_engine_test:/log
     command:
       - "/tiflow"

deployments/engine/helm/tiflow/values.yaml

@@ -45,12 +45,6 @@ executor:
     keepalive-interval = "500ms"
     session-ttl = 20
 
-    [storage.s3]
-    bucket = "engine-it"
-    endpoint = "http://chaos-minio:9000/"
-    access-key = "engine"
-    secret-access-key = "engineSecret"
-
 metastore:
   frameworkStorage: 5Gi
   businessStorage: 5Gi

engine/executor/config.go

@@ -15,16 +15,13 @@ package executor
 
 import (
 	"bytes"
-	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"path/filepath"
 	"strings"
 	"time"
 
 	"github.com/BurntSushi/toml"
 	"github.com/pingcap/log"
-	resModel "github.com/pingcap/tiflow/engine/pkg/externalresource/model"
 	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/label"
 	"github.com/pingcap/tiflow/pkg/logutil"
@@ -39,10 +36,8 @@ var (
 	defaultRPCTimeout        = "3s"
 	defaultMetricInterval    = 15 * time.Second
 
-	defaultCapability            int64 = 100 // TODO: make this configurable
-	defaultLocalStorageDirPrefix       = "/tmp/dfe-storage/"
-
-	defaultExecutorAddr = "127.0.0.1:10340"
+	defaultCapability   int64 = 100 // TODO: make this configurable
+	defaultExecutorAddr       = "127.0.0.1:10340"
 )
 
 // Config is the configuration.
@@ -64,8 +59,6 @@ type Config struct {
 	KeepAliveIntervalStr string `toml:"keepalive-interval" json:"keepalive-interval"`
 	RPCTimeoutStr        string `toml:"rpc-timeout" json:"rpc-timeout"`
 
-	Storage resModel.Config `toml:"storage" json:"storage"`
-
 	KeepAliveTTL      time.Duration `toml:"-" json:"-"`
 	KeepAliveInterval time.Duration `toml:"-" json:"-"`
 	RPCTimeout        time.Duration `toml:"-" json:"-"`
@@ -111,13 +104,6 @@ func (c *Config) configFromFile(path string) error {
 	return nil
 }
 
-func getDefaultLocalStorageDir(executorName string) string {
-	// Use hex encoding in case there are special characters in the
-	// executor name.
-	encodedExecutorName := hex.EncodeToString([]byte(executorName))
-	return filepath.Join(defaultLocalStorageDirPrefix, encodedExecutorName)
-}
-
 // Adjust adjusts the executor configuration
 func (c *Config) Adjust() (err error) {
 	if c.AdvertiseAddr == "" {
@@ -128,10 +114,6 @@ func (c *Config) Adjust() (err error) {
 		c.Name = fmt.Sprintf("executor-%s", c.AdvertiseAddr)
 	}
 
-	if c.Storage.Local.BaseDir == "" {
-		c.Storage.Local.BaseDir = getDefaultLocalStorageDir(c.Name)
-	}
-
 	c.KeepAliveInterval, err = time.ParseDuration(c.KeepAliveIntervalStr)
 	if err != nil {
 		return
@@ -169,6 +151,5 @@ func GetDefaultExecutorConfig() *Config {
 		KeepAliveTTLStr:      defaultKeepAliveTTL,
 		KeepAliveIntervalStr: defaultKeepAliveInterval,
 		RPCTimeoutStr:        defaultRPCTimeout,
-		Storage:              resModel.DefaultConfig,
 	}
 }

engine/executor/config_test.go

@@ -14,7 +14,6 @@
 package executor
 
 import (
-	"encoding/hex"
 	"os"
 	"testing"
 
@@ -36,10 +35,8 @@ join = "127.0.0.1:10240"
 	err = cfg.Adjust()
 	require.NoError(t, err)
 
-	expectedPath := "/tmp/dfe-storage/" + hex.EncodeToString([]byte("executor-1"))
 	require.Equal(t, "executor-1", cfg.Name)
 	require.Equal(t, "0.0.0.0:10241", cfg.AdvertiseAddr)
-	require.Equal(t, expectedPath, cfg.Storage.Local.BaseDir)
 }
 
 func TestConfigDefaultLocalStoragePathNoName(t *testing.T) {
@@ -56,9 +53,7 @@ join = "127.0.0.1:10240"
 	err = cfg.Adjust()
 	require.NoError(t, err)
 
-	expectedPath := "/tmp/dfe-storage/" + hex.EncodeToString([]byte("executor-0.0.0.0:10241"))
 	require.Equal(t, "0.0.0.0:10241", cfg.AdvertiseAddr)
-	require.Equal(t, expectedPath, cfg.Storage.Local.BaseDir)
 }
 
 func TestConfigStorage(t *testing.T) {
@@ -68,18 +63,13 @@ func TestConfigStorage(t *testing.T) {
 name = "executor-1"
 addr = "0.0.0.0:10241"
 join = "127.0.0.1:10240"
-
-[storage]
-local.base-dir = "/tmp/my-base-dir"
 `
 	fileName := mustWriteToTempFile(t, testToml)
 	cfg := GetDefaultExecutorConfig()
 	err := cfg.configFromFile(fileName)
 	require.NoError(t, err)
 	err = cfg.Adjust()
 	require.NoError(t, err)
-
-	require.Equal(t, "/tmp/my-base-dir", cfg.Storage.Local.BaseDir)
 }
 
 func mustWriteToTempFile(t *testing.T, content string) (filePath string) {

engine/executor/dm/worker.go

@@ -193,11 +193,17 @@ func (w *dmWorker) CloseImpl(ctx context.Context) {
 // setupStorage opens and configs external storage
 func (w *dmWorker) setupStorage(ctx context.Context) error {
 	rid := dm.NewDMResourceID(w.cfg.Name, w.cfg.SourceID)
-	h, err := w.OpenStorage(ctx, rid)
+	opts := []broker.OpenStorageOption{}
+	if w.workerType == frameModel.WorkerDMDump {
+		// always use an empty storage for dumpling task
+		opts = append(opts, broker.WithCleanBeforeOpen())
+	}
+
+	h, err := w.OpenStorage(ctx, rid, opts...)
 	for status.Code(err) == codes.Unavailable {
 		w.Logger().Info("simple retry", zap.Error(err))
 		time.Sleep(time.Second)
-		h, err = w.OpenStorage(ctx, rid)
+		h, err = w.OpenStorage(ctx, rid, opts...)
 	}
 	if err != nil {
 		return errors.Trace(err)

engine/executor/server.go

@@ -444,11 +444,7 @@ func (s *Server) Run(ctx context.Context) error {
 		return err
 	}
 
-	if err := broker.PreCheckConfig(s.cfg.Storage); err != nil {
-		return err
-	}
-
-	s.resourceBroker, err = broker.NewBroker(&s.cfg.Storage, s.selfID, s.masterClient)
+	s.resourceBroker, err = broker.NewBroker(ctx, s.selfID, s.masterClient)
 	if err != nil {
 		return err
 	}

engine/framework/worker.go

@@ -122,7 +122,9 @@ type BaseWorker interface {
 	SendMessage(ctx context.Context, topic p2p.Topic, message interface{}, nonblocking bool) error
 
 	// OpenStorage creates a resource and return the resource handle
-	OpenStorage(ctx context.Context, resourcePath resModel.ResourceID) (broker.Handle, error)
+	OpenStorage(
+		ctx context.Context, resourcePath resModel.ResourceID, opts ...broker.OpenStorageOption,
+	) (broker.Handle, error)
 
 	// Exit should be called when worker (in user logic) wants to exit.
 	// exitReason: ExitReasonFinished/ExitReasonCanceled/ExitReasonFailed
@@ -505,10 +507,12 @@ func (w *DefaultBaseWorker) SendMessage(
 }
 
 // OpenStorage implements BaseWorker.OpenStorage
-func (w *DefaultBaseWorker) OpenStorage(ctx context.Context, resourcePath resModel.ResourceID) (broker.Handle, error) {
+func (w *DefaultBaseWorker) OpenStorage(
+	ctx context.Context, resourcePath resModel.ResourceID, opts ...broker.OpenStorageOption,
+) (broker.Handle, error) {
 	ctx, cancel := w.errCenter.WithCancelOnFirstError(ctx)
 	defer cancel()
-	return w.resourceBroker.OpenStorage(ctx, w.projectInfo, w.id, w.masterID, resourcePath)
+	return w.resourceBroker.OpenStorage(ctx, w.projectInfo, w.id, w.masterID, resourcePath, opts...)
 }
 
 // Exit implements BaseWorker.Exit