get storage config from 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"

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/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
 	}