openapi(dm): support task timezone

dm/config/task_converters.go

@@ -161,6 +161,9 @@ func OpenAPITaskToSubTaskConfigs(task *openapi.Task, toDBCfg *dbconfig.DBConfig,
 		// set task name and mode
 		subTaskCfg.Name = task.Name
 		subTaskCfg.Mode = string(task.TaskMode)
+		if task.Timezone != nil {
+			subTaskCfg.Timezone = *task.Timezone
+		}
 		// set task meta
 		subTaskCfg.MetaSchema = *task.MetaSchema
 		// add binlog meta
@@ -201,7 +204,7 @@ func OpenAPITaskToSubTaskConfigs(task *openapi.Task, toDBCfg *dbconfig.DBConfig,
 		subTaskCfg.CaseSensitive = sourceCfgMap[sourceCfg.SourceName].CaseSensitive
 		// set source db config
 		subTaskCfg.SourceID = sourceCfg.SourceName
-		subTaskCfg.From = sourceCfgMap[sourceCfg.SourceName].From
+		subTaskCfg.From = *sourceCfgMap[sourceCfg.SourceName].From.Clone()
 		// set target db config
 		subTaskCfg.To = *toDBCfg.Clone()
 		// TODO ExprFilter
@@ -683,6 +686,10 @@ func SubTaskConfigsToOpenAPITask(subTaskConfigList []*SubTaskConfig) *openapi.Ta
 			Password: oneSubtaskConfig.To.Password,
 		},
 	}
+	if oneSubtaskConfig.Timezone != "" {
+		timezone := oneSubtaskConfig.Timezone
+		task.Timezone = &timezone
+	}
 	if oneSubtaskConfig.ShardMode != "" {
 		taskShardMode := openapi.TaskShardMode(oneSubtaskConfig.ShardMode)
 		task.ShardMode = &taskShardMode

dm/config/task_converters_test.go

@@ -60,12 +60,14 @@ func testNoShardTaskToSubTaskConfigs(c *check.C) {
 	c.Assert(err, check.IsNil)
 	source1Name := task.SourceConfig.SourceConf[0].SourceName
 	sourceCfg1.SourceID = task.SourceConfig.SourceConf[0].SourceName
+	sourceCfg1.From.Session = map[string]string{"time_zone": "UTC"}
 	sourceCfgMap := map[string]*SourceConfig{source1Name: sourceCfg1}
 	toDBCfg := &dbconfig.DBConfig{
 		Host:     task.TargetConfig.Host,
 		Port:     task.TargetConfig.Port,
 		User:     task.TargetConfig.User,
 		Password: task.TargetConfig.Password,
+		Session:  map[string]string{"time_zone": "UTC"},
 		Security: &security.Security{
 			SSLCABytes:    []byte(task.TargetConfig.Security.SslCaContent),
 			SSLCertBytes:  []byte(task.TargetConfig.Security.SslCertContent),
@@ -82,6 +84,7 @@ func testNoShardTaskToSubTaskConfigs(c *check.C) {
 	subTaskConfig := subTaskConfigList[0]
 	// check task name and mode
 	c.Assert(subTaskConfig.Name, check.Equals, task.Name)
+	c.Assert(subTaskConfig.Timezone, check.Equals, *task.Timezone)
 	// check task meta
 	c.Assert(subTaskConfig.MetaSchema, check.Equals, *task.MetaSchema)
 	c.Assert(subTaskConfig.Meta, check.IsNil)
@@ -93,8 +96,12 @@ func testNoShardTaskToSubTaskConfigs(c *check.C) {
 	c.Assert(subTaskConfig.CaseSensitive, check.Equals, sourceCfg1.CaseSensitive)
 	// check from
 	c.Assert(subTaskConfig.From.Host, check.Equals, sourceCfg1.From.Host)
+	c.Assert(subTaskConfig.From.Session["time_zone"], check.Equals, *task.Timezone)
+	c.Assert(sourceCfg1.From.Session["time_zone"], check.Equals, "UTC")
 	// check to
 	c.Assert(subTaskConfig.To.Host, check.Equals, toDBCfg.Host)
+	c.Assert(subTaskConfig.To.Session["time_zone"], check.Equals, *task.Timezone)
+	c.Assert(toDBCfg.Session["time_zone"], check.Equals, "UTC")
 	// check dumpling loader syncer config
 	c.Assert(subTaskConfig.MydumperConfig.Threads, check.Equals, *task.SourceConfig.FullMigrateConf.ExportThreads)
 	c.Assert(subTaskConfig.LoaderConfig.Dir, check.Equals, fmt.Sprintf(
@@ -464,11 +471,34 @@ func TestConvertBetweenOpenAPITaskAndTaskConfig(t *testing.T) {
 	taskCfg, err := OpenAPITaskToTaskConfig(&task, sourceCfgMap)
 	require.NoError(t, err)
 	require.NotNil(t, taskCfg)
+	require.Equal(t, *task.Timezone, taskCfg.Timezone)
 	task1, err := TaskConfigToOpenAPITask(taskCfg, sourceCfgMap)
 	require.NoError(t, err)
 	require.NotNil(t, task1)
 	require.EqualValues(t, task1, &task)
 
+	configuredTimezone := task.Timezone
+	emptyTimezone := ""
+	for _, tc := range []struct {
+		name     string
+		timezone *string
+	}{
+		{name: "omitted", timezone: nil},
+		{name: "empty", timezone: &emptyTimezone},
+	} {
+		t.Run(tc.name+" timezone uses downstream default", func(t *testing.T) {
+			task.Timezone = tc.timezone
+			taskCfgWithoutTimezone, convertErr := OpenAPITaskToTaskConfig(&task, sourceCfgMap)
+			require.NoError(t, convertErr)
+			require.Empty(t, taskCfgWithoutTimezone.Timezone)
+
+			taskWithoutTimezone, convertErr := TaskConfigToOpenAPITask(taskCfgWithoutTimezone, sourceCfgMap)
+			require.NoError(t, convertErr)
+			require.Nil(t, taskWithoutTimezone.Timezone)
+		})
+	}
+	task.Timezone = configuredTimezone
+
 	// test update some fields in task
 	{
 		batch := 1000

dm/master/openapi_controller_test.go

@@ -298,6 +298,9 @@ func (s *OpenAPIControllerSuite) TestTaskController() {
 		res, err := server.createTask(ctx, createTaskReq)
 		s.Nil(err)
 		s.EqualValues(*s.testTask, res.Task)
+		subTaskCfgs := server.scheduler.GetSubTaskCfgsByTask(s.testTask.Name)
+		s.Len(subTaskCfgs, 1)
+		s.Equal(*s.testTask.Timezone, subTaskCfgs[s.testSource.SourceName].Timezone)
 	}
 
 	// update
@@ -470,6 +473,22 @@ func (s *OpenAPIControllerSuite) TestTaskController() {
 	}
 }
 
+func (s *OpenAPIControllerSuite) TestCreateTaskRejectsInvalidTimezone() {
+	ctx, cancel := context.WithCancel(context.Background())
+	server := setupTestServer(ctx, s.T())
+	defer func() {
+		cancel()
+		server.Close()
+	}()
+
+	task := *s.testTask
+	invalidTimezone := "invalid/timezone"
+	task.Timezone = &invalidTimezone
+	res, err := server.createTask(ctx, openapi.CreateTaskRequest{Task: task})
+	s.Nil(res)
+	s.True(terror.ErrConfigInvalidTimezone.Equal(err))
+}
+
 func (s *OpenAPIControllerSuite) TestTaskStatusSourceErrorFallback() {
 	ctx, cancel := context.WithCancel(context.Background())
 	server := setupTestServer(ctx, s.T())

dm/master/openapi_view_test.go

@@ -41,9 +41,11 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	"github.com/pingcap/tiflow/dm/syncer"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	"github.com/tikv/pd/pkg/utils/tempurl"
+	"google.golang.org/grpc"
 )
 
 // some data for test.
@@ -549,6 +551,7 @@ func (s *OpenAPIViewSuite) TestTaskTemplatesAPI() {
 	var createTaskResp openapi.Task
 	s.NoError(result.UnmarshalBodyToObject(&createTaskResp))
 	s.Equal(createTaskResp.Name, task.Name)
+	s.EqualValues(task.Timezone, createTaskResp.Timezone)
 
 	// create again will fail
 	result = testutil.NewRequest().Post(url).WithJsonBody(task).GoWithHTTPHandler(s.T(), s1.openapiHandles)
@@ -564,6 +567,7 @@ func (s *OpenAPIViewSuite) TestTaskTemplatesAPI() {
 	s.NoError(result.UnmarshalBodyToObject(&resultTaskList))
 	s.Equal(1, resultTaskList.Total)
 	s.Equal(task.Name, resultTaskList.Data[0].Name)
+	s.EqualValues(task.Timezone, resultTaskList.Data[0].Timezone)
 
 	// get detail
 	oneURL := fmt.Sprintf("%s/%s", url, task.Name)
@@ -572,6 +576,7 @@ func (s *OpenAPIViewSuite) TestTaskTemplatesAPI() {
 	var respTask openapi.Task
 	s.NoError(result.UnmarshalBodyToObject(&respTask))
 	s.Equal(task.Name, respTask.Name)
+	s.EqualValues(task.Timezone, respTask.Timezone)
 
 	// get not exist
 	notExistURL := fmt.Sprintf("%s/%s", url, "notexist")
@@ -1000,9 +1005,11 @@ func (s *OpenAPIViewSuite) TestTaskAPI() {
 	var createTaskResp openapi.OperateTaskResponse
 	s.NoError(result.UnmarshalBodyToObject(&createTaskResp))
 	s.Equal(createTaskResp.Task.Name, task.Name)
+	s.EqualValues(task.Timezone, createTaskResp.Task.Timezone)
 	subTaskM := s1.scheduler.GetSubTaskCfgsByTask(task.Name)
 	s.Len(subTaskM, 1)
 	s.Equal(task.Name, subTaskM[source1Name].Name)
+	s.Equal(*task.Timezone, subTaskM[source1Name].Timezone)
 
 	// get task
 	task1URL := fmt.Sprintf("%s/%s", taskURL, task.Name)
@@ -1011,6 +1018,44 @@ func (s *OpenAPIViewSuite) TestTaskAPI() {
 	s.Equal(http.StatusOK, result.Code())
 	s.NoError(result.UnmarshalBodyToObject(&task1FromHTTP))
 	s.Equal(task1FromHTTP.Name, task.Name)
+	s.EqualValues(task.Timezone, task1FromHTTP.Timezone)
+
+	// changing timezone through PUT is rejected by the bottom update guard.
+	updatedTimezone := "UTC"
+	timezoneUpdateTask := task
+	timezoneUpdateTask.Timezone = &updatedTimezone
+	oldSubTaskCfg := subTaskM[source1Name]
+	syncUnit := syncer.NewSyncer(oldSubTaskCfg, nil, nil)
+	mockUpdateWorkerClient := pbmock.NewMockWorkerClient(ctrl)
+	mockUpdateWorkerClient.EXPECT().CheckSubtasksCanUpdate(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ context.Context, req *pb.CheckSubtasksCanUpdateRequest, _ ...grpc.CallOption) (*pb.CheckSubtasksCanUpdateResponse, error) {
+			newSubTaskCfg := config.NewSubTaskConfig()
+			decodeErr := newSubTaskCfg.Decode(req.SubtaskCfgTomlString, false)
+			s.NoError(decodeErr)
+			if decodeErr != nil {
+				return &pb.CheckSubtasksCanUpdateResponse{Msg: decodeErr.Error()}, nil
+			}
+			s.Equal(updatedTimezone, newSubTaskCfg.Timezone)
+
+			guardErr := syncUnit.CheckCanUpdateCfg(newSubTaskCfg)
+			s.Error(guardErr)
+			resp := &pb.CheckSubtasksCanUpdateResponse{Success: guardErr == nil}
+			if guardErr != nil {
+				resp.Msg = guardErr.Error()
+			}
+			return resp, nil
+		},
+	)
+	s1.scheduler.SetWorkerClientForTest(workerName1, newMockRPCClient(mockUpdateWorkerClient))
+	timezoneUpdateReq := openapi.UpdateTaskRequest{Task: timezoneUpdateTask}
+	result = testutil.NewRequest().Put(task1URL).WithJsonBody(timezoneUpdateReq).GoWithHTTPHandler(s.T(), s1.openapiHandles)
+	s.Equal(http.StatusBadRequest, result.Code())
+	var timezoneUpdateErr openapi.ErrorWithMessage
+	s.NoError(result.UnmarshalBodyToObject(&timezoneUpdateErr))
+	s.Equal(int(terror.ErrSchedulerSubTaskCfgUpdate.Code()), timezoneUpdateErr.ErrorCode)
+	s.Contains(timezoneUpdateErr.ErrorMsg, "fields that should not be changed")
+	storedSubTaskCfgs := s1.scheduler.GetSubTaskCfgsByTask(task.Name)
+	s.Equal(*task.Timezone, storedSubTaskCfgs[source1Name].Timezone)
 
 	// update a task
 	s.NoError(failpoint.Enable("github.com/pingcap/tiflow/dm/master/scheduler/operateCheckSubtasksCanUpdate", `return("success")`))
@@ -1032,6 +1077,7 @@ func (s *OpenAPIViewSuite) TestTaskAPI() {
 	s.NoError(result.UnmarshalBodyToObject(&resultTaskList))
 	s.Equal(1, resultTaskList.Total)
 	s.Equal(task.Name, resultTaskList.Data[0].Name)
+	s.EqualValues(task.Timezone, resultTaskList.Data[0].Timezone)
 
 	s.testImportTaskTemplate(&task, s1)
 

dm/openapi/fixtures/task.go

@@ -27,6 +27,7 @@ var (
 		"meta_schema": "dm_meta",
 		"name": "test",
 		"on_duplicate": "replace",
+		"timezone": "Asia/Shanghai",
 		"source_config": {
 		  "full_migrate_conf": {
 			"data_dir": "./exported_data",