copy tiflow config to tigate (pingcap#432)

api/middleware/middleware.go

@@ -19,12 +19,12 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/flowbehappy/tigate/pkg/node"
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/api"
 	"github.com/pingcap/tiflow/cdc/model"
-	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/httputil"
 	"go.uber.org/zap"
 )

api/v2/changefeed.go

@@ -19,6 +19,8 @@ import (
 	"net/url"
 	"time"
 
+	"github.com/flowbehappy/tigate/pkg/config"
+	"github.com/flowbehappy/tigate/pkg/filter"
 	"github.com/flowbehappy/tigate/version"
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
@@ -27,7 +29,6 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/owner"
 	"github.com/pingcap/tiflow/pkg/errors"
-	"github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/txnutil/gc"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/tikv/client-go/v2/oracle"
@@ -128,7 +129,7 @@ func (h *OpenAPIV2) createChangefeed(c *gin.Context) {
 	}
 
 	pdClient := h.server.GetPdClient()
-	info := &model.ChangeFeedInfo{
+	info := &config.ChangeFeedInfo{
 		UpstreamID:     pdClient.GetClusterID(ctx),
 		Namespace:      cfg.Namespace,
 		ID:             cfg.ID,
@@ -265,7 +266,7 @@ func (h *OpenAPIV2) getChangeFeed(c *gin.Context) {
 }
 
 func toAPIModel(
-	info *model.ChangeFeedInfo,
+	info *config.ChangeFeedInfo,
 	resolvedTs uint64,
 	checkpointTs uint64,
 	taskStatus []model.CaptureTaskStatus,
@@ -555,7 +556,7 @@ func (h *OpenAPIV2) updateChangefeed(c *gin.Context) {
 	}
 
 	// verify changefeed filter
-	_, err = filter.NewFilter(oldCfInfo.Config, "")
+	_, err = filter.NewFilter(oldCfInfo.Config.Filter, "", oldCfInfo.Config.CaseSensitive)
 	if err != nil {
 		_ = c.Error(errors.ErrChangefeedUpdateRefused.
 			GenWithStackByArgs(errors.Cause(err).Error()))

api/v2/model.go

@@ -17,10 +17,10 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tiflow/cdc/model"
 	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
-	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
 	"github.com/pingcap/tiflow/pkg/security"

cmd/cli/cli_changefeed_create.go

@@ -23,13 +23,13 @@ import (
 	v2 "github.com/flowbehappy/tigate/api/v2"
 	"github.com/flowbehappy/tigate/cmd/factory"
 	apiv2client "github.com/flowbehappy/tigate/pkg/api/v2"
+	"github.com/flowbehappy/tigate/pkg/config"
+	"github.com/flowbehappy/tigate/pkg/filter"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	cmdcontext "github.com/pingcap/tiflow/pkg/cmd/context"
 	"github.com/pingcap/tiflow/pkg/cmd/util"
-	"github.com/pingcap/tiflow/pkg/config"
-	"github.com/pingcap/tiflow/pkg/filter"
 	putil "github.com/pingcap/tiflow/pkg/util"
 	"github.com/spf13/cobra"
 	"github.com/tikv/client-go/v2/oracle"

coordinator/changefeed/changefeed.go

@@ -18,6 +18,7 @@ import (
 	"net/url"
 
 	"github.com/flowbehappy/tigate/heartbeatpb"
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/flowbehappy/tigate/pkg/messaging"
 	"github.com/flowbehappy/tigate/pkg/node"
 	"github.com/pingcap/log"
@@ -30,7 +31,7 @@ import (
 // Changefeed is a memory present for changefeed info and status
 type Changefeed struct {
 	ID       model.ChangeFeedID
-	Info     *model.ChangeFeedInfo
+	Info     *config.ChangeFeedInfo
 	IsMQSink bool
 
 	nodeID      node.ID
@@ -44,7 +45,7 @@ type Changefeed struct {
 
 // NewChangefeed creates a new changefeed instance
 func NewChangefeed(cfID model.ChangeFeedID,
-	info *model.ChangeFeedInfo,
+	info *config.ChangeFeedInfo,
 	checkpointTs uint64) *Changefeed {
 	uri, err := url.Parse(info.SinkURI)
 	if err != nil {

coordinator/changefeed/changefeed_db.go

@@ -17,6 +17,7 @@ import (
 	"math"
 	"sync"
 
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/flowbehappy/tigate/pkg/node"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -321,7 +322,7 @@ func (db *ChangefeedDB) CalculateGCSafepoint() uint64 {
 }
 
 // ReplaceStoppedChangefeed updates the stopped changefeed
-func (db *ChangefeedDB) ReplaceStoppedChangefeed(cf *model.ChangeFeedInfo) {
+func (db *ChangefeedDB) ReplaceStoppedChangefeed(cf *config.ChangeFeedInfo) {
 	db.lock.Lock()
 	defer db.lock.Unlock()
 

coordinator/changefeed/changefeed_db_backend.go

@@ -16,6 +16,7 @@ package changefeed
 import (
 	"context"
 
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/pingcap/tiflow/cdc/model"
 )
 
@@ -24,9 +25,9 @@ type Backend interface {
 	// GetAllChangefeeds returns all changefeeds from the backend db, include stopped and failed changefeeds
 	GetAllChangefeeds(ctx context.Context) (map[model.ChangeFeedID]*ChangefeedMetaWrapper, error)
 	// CreateChangefeed saves changefeed info and status to db
-	CreateChangefeed(ctx context.Context, info *model.ChangeFeedInfo) error
+	CreateChangefeed(ctx context.Context, info *config.ChangeFeedInfo) error
 	// UpdateChangefeed updates changefeed info  to db
-	UpdateChangefeed(ctx context.Context, info *model.ChangeFeedInfo) error
+	UpdateChangefeed(ctx context.Context, info *config.ChangeFeedInfo) error
 	// PauseChangefeed persists the pause status to db for a changefeed
 	PauseChangefeed(ctx context.Context, id model.ChangeFeedID) error
 	// DeleteChangefeed removes all related info of a changefeed from db
@@ -39,6 +40,6 @@ type Backend interface {
 
 // ChangefeedMetaWrapper is a wrapper for the changefeed load from the DB
 type ChangefeedMetaWrapper struct {
-	Info   *model.ChangeFeedInfo
+	Info   *config.ChangeFeedInfo
 	Status *model.ChangeFeedStatus
 }

coordinator/changefeed/etcd_backend.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/errors"
@@ -69,7 +70,7 @@ func (b *EtcdBackend) GetAllChangefeeds(ctx context.Context) (map[model.ChangeFe
 			}
 			meta.Status = status
 		} else {
-			detail := &model.ChangeFeedInfo{}
+			detail := &config.ChangeFeedInfo{}
 			err = detail.Unmarshal(kv.Value)
 			if err != nil {
 				log.Warn("failed to unmarshal change feed Info, ignore",
@@ -113,7 +114,7 @@ func (b *EtcdBackend) GetAllChangefeeds(ctx context.Context) (map[model.ChangeFe
 }
 
 func (b *EtcdBackend) CreateChangefeed(ctx context.Context,
-	info *model.ChangeFeedInfo) error {
+	info *config.ChangeFeedInfo) error {
 	changefeedID := model.DefaultChangeFeedID(info.ID)
 	infoKey := etcd.GetEtcdKeyChangeFeedInfo(b.etcdClient.GetClusterID(), changefeedID)
 	infoValue, err := info.Marshal()
@@ -146,7 +147,7 @@ func (b *EtcdBackend) CreateChangefeed(ctx context.Context,
 	return nil
 }
 
-func (b *EtcdBackend) UpdateChangefeed(ctx context.Context, info *model.ChangeFeedInfo) error {
+func (b *EtcdBackend) UpdateChangefeed(ctx context.Context, info *config.ChangeFeedInfo) error {
 	infoKey := etcd.GetEtcdKeyChangeFeedInfo(b.etcdClient.GetClusterID(), model.DefaultChangeFeedID(info.ID))
 	newStr, err := info.Marshal()
 	if err != nil {

coordinator/controller.go

@@ -23,6 +23,7 @@ import (
 	"github.com/flowbehappy/tigate/heartbeatpb"
 	"github.com/flowbehappy/tigate/pkg/bootstrap"
 	appcontext "github.com/flowbehappy/tigate/pkg/common/context"
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/flowbehappy/tigate/pkg/messaging"
 	"github.com/flowbehappy/tigate/pkg/metrics"
 	"github.com/flowbehappy/tigate/pkg/node"
@@ -148,7 +149,7 @@ func (c *Controller) HandleEvent(event *Event) bool {
 	return false
 }
 
-func (c *Controller) CreateChangefeed(ctx context.Context, info *model.ChangeFeedInfo) error {
+func (c *Controller) CreateChangefeed(ctx context.Context, info *config.ChangeFeedInfo) error {
 	if !c.bootstrapped {
 		return errors.New("not initialized, wait a moment")
 	}
@@ -387,7 +388,7 @@ func (c *Controller) ResumeChangefeed(ctx context.Context, id model.ChangeFeedID
 	return nil
 }
 
-func (c *Controller) UpdateChangefeed(ctx context.Context, change *model.ChangeFeedInfo) error {
+func (c *Controller) UpdateChangefeed(ctx context.Context, change *config.ChangeFeedInfo) error {
 	id := model.ChangeFeedID{
 		Namespace: change.Namespace,
 		ID:        change.ID,
@@ -403,9 +404,9 @@ func (c *Controller) UpdateChangefeed(ctx context.Context, change *model.ChangeF
 	return nil
 }
 
-func (c *Controller) ListChangefeeds(ctx context.Context) ([]*model.ChangeFeedInfo, []*model.ChangeFeedStatus, error) {
+func (c *Controller) ListChangefeeds(ctx context.Context) ([]*config.ChangeFeedInfo, []*model.ChangeFeedStatus, error) {
 	cfs := c.changefeedDB.GetAllChangefeeds()
-	infos := make([]*model.ChangeFeedInfo, 0, len(cfs))
+	infos := make([]*config.ChangeFeedInfo, 0, len(cfs))
 	statuses := make([]*model.ChangeFeedStatus, 0, len(cfs))
 	for _, cf := range cfs {
 		infos = append(infos, cf.Info)
@@ -414,7 +415,7 @@ func (c *Controller) ListChangefeeds(ctx context.Context) ([]*model.ChangeFeedIn
 	return infos, statuses, nil
 }
 
-func (c *Controller) GetChangefeed(ctx context.Context, id model.ChangeFeedID) (*model.ChangeFeedInfo, *model.ChangeFeedStatus, error) {
+func (c *Controller) GetChangefeed(ctx context.Context, id model.ChangeFeedID) (*config.ChangeFeedInfo, *model.ChangeFeedStatus, error) {
 	cf := c.changefeedDB.GetByID(id)
 	if cf == nil {
 		return nil, nil, cerror.ErrChangeFeedNotExists.GenWithStackByArgs(id.ID)

coordinator/coordinator.go

@@ -21,6 +21,7 @@ import (
 	"github.com/flowbehappy/tigate/coordinator/changefeed"
 	"github.com/flowbehappy/tigate/heartbeatpb"
 	appcontext "github.com/flowbehappy/tigate/pkg/common/context"
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/flowbehappy/tigate/pkg/messaging"
 	"github.com/flowbehappy/tigate/pkg/metrics"
 	"github.com/flowbehappy/tigate/pkg/node"
@@ -142,7 +143,7 @@ func (c *coordinator) Run(ctx context.Context) error {
 	}
 }
 
-func (c *coordinator) CreateChangefeed(ctx context.Context, info *model.ChangeFeedInfo) error {
+func (c *coordinator) CreateChangefeed(ctx context.Context, info *config.ChangeFeedInfo) error {
 	return c.controller.CreateChangefeed(ctx, info)
 }
 
@@ -158,15 +159,15 @@ func (c *coordinator) ResumeChangefeed(ctx context.Context, id model.ChangeFeedI
 	return c.controller.ResumeChangefeed(ctx, id, newCheckpointTs)
 }
 
-func (c *coordinator) UpdateChangefeed(ctx context.Context, change *model.ChangeFeedInfo) error {
+func (c *coordinator) UpdateChangefeed(ctx context.Context, change *config.ChangeFeedInfo) error {
 	return c.controller.UpdateChangefeed(ctx, change)
 }
 
-func (c *coordinator) ListChangefeeds(ctx context.Context) ([]*model.ChangeFeedInfo, []*model.ChangeFeedStatus, error) {
+func (c *coordinator) ListChangefeeds(ctx context.Context) ([]*config.ChangeFeedInfo, []*model.ChangeFeedStatus, error) {
 	return c.controller.ListChangefeeds(ctx)
 }
 
-func (c *coordinator) GetChangefeed(ctx context.Context, id model.ChangeFeedID) (*model.ChangeFeedInfo, *model.ChangeFeedStatus, error) {
+func (c *coordinator) GetChangefeed(ctx context.Context, id model.ChangeFeedID) (*config.ChangeFeedInfo, *model.ChangeFeedStatus, error) {
 	return c.controller.GetChangefeed(ctx, id)
 }
 

downstreamadapter/dispatchermanager/event_dispatcher_manager.go

@@ -38,7 +38,6 @@ import (
 	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/flowbehappy/tigate/pkg/metrics"
 	"github.com/pingcap/tiflow/cdc/model"
-	cfg "github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/util"
 	"go.uber.org/zap"
 )
@@ -134,7 +133,7 @@ func NewEventDispatcherManager(changefeedID model.ChangeFeedID,
 
 	// Set Filter
 	// TODO: 最后去更新一下 filter 的内部 NewFilter 函数，现在是在套壳适配
-	replicaConfig := cfg.ReplicaConfig{Filter: cfConfig.Filter}
+	replicaConfig := config.ReplicaConfig{Filter: cfConfig.Filter}
 	filter, err := filter.NewFilter(replicaConfig.Filter, cfConfig.TimeZone, replicaConfig.CaseSensitive)
 	if err != nil {
 		return nil, 0, apperror.ErrCreateEventDispatcherManagerFailed.Wrap(err).GenWithStackByArgs("create filter failed")
@@ -463,7 +462,7 @@ func (e *EventDispatcherManager) cleanTableEventDispatcher(id common.DispatcherI
 	log.Info("table event dispatcher completely stopped, and delete it from event dispatcher manager", zap.Any("dispatcher id", id))
 }
 
-func toFilterConfigPB(filter *cfg.FilterConfig) *eventpb.FilterConfig {
+func toFilterConfigPB(filter *config.FilterConfig) *eventpb.FilterConfig {
 	filterConfig := &eventpb.FilterConfig{
 		Rules:            filter.Rules,
 		IgnoreTxnStartTs: filter.IgnoreTxnStartTs,

downstreamadapter/sink/helper/eventrouter/event_router.go

@@ -18,10 +18,9 @@ import (
 	"github.com/flowbehappy/tigate/downstreamadapter/sink/helper/eventrouter/topic"
 	"github.com/flowbehappy/tigate/pkg/common"
 	commonEvent "github.com/flowbehappy/tigate/pkg/common/event"
-	ticonfig "github.com/flowbehappy/tigate/pkg/config"
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/pingcap/log"
 	tableFilter "github.com/pingcap/tidb/pkg/util/table-filter"
-	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 )
 
@@ -39,11 +38,11 @@ type EventRouter struct {
 }
 
 // NewEventRouter creates a new EventRouter.
-func NewEventRouter(sinkConfig *ticonfig.SinkConfig, protocol config.Protocol, defaultTopic, scheme string) (*EventRouter, error) {
+func NewEventRouter(sinkConfig *config.SinkConfig, protocol config.Protocol, defaultTopic, scheme string) (*EventRouter, error) {
 	// If an event does not match any dispatching rules in the config file,
 	// it will be dispatched by the default partition dispatcher and
 	// static topic dispatcher because it matches *.* rule.
-	ruleConfigs := append(sinkConfig.DispatchRules, &ticonfig.DispatchRule{
+	ruleConfigs := append(sinkConfig.DispatchRules, &config.DispatchRule{
 		Matcher:       []string{"*.*"},
 		PartitionRule: "default",
 		TopicRule:     "",

downstreamadapter/sink/helper/eventrouter/topic/expression.go

@@ -4,7 +4,7 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/flowbehappy/tigate/pkg/config"
 	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/sink"
 )

downstreamadapter/sink/helper/eventrouter/topic/topic.go

@@ -1,7 +1,7 @@
 package topic
 
 import (
-	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/flowbehappy/tigate/pkg/config"
 )
 
 type TopicGeneratorType int

downstreamadapter/sink/helper/helper.go

@@ -1,9 +1,23 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package helper
 
 import (
 	"net/url"
 	"strings"
 
+	"github.com/flowbehappy/tigate/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 )
 
@@ -17,3 +31,30 @@ func GetTopic(sinkURI *url.URL) (string, error) {
 	}
 	return topic, nil
 }
+
+// GetProtocol returns the protocol from the sink URI.
+func GetProtocol(protocolStr string) (config.Protocol, error) {
+	protocol, err := config.ParseSinkProtocolFromString(protocolStr)
+	if err != nil {
+		return protocol, cerror.WrapError(cerror.ErrKafkaInvalidConfig, err)
+	}
+
+	return protocol, nil
+}
+
+// GetFileExtension returns the extension for specific protocol
+func GetFileExtension(protocol config.Protocol) string {
+	switch protocol {
+	case config.ProtocolAvro, config.ProtocolCanalJSON, config.ProtocolMaxwell,
+		config.ProtocolOpen, config.ProtocolSimple:
+		return ".json"
+	case config.ProtocolCraft:
+		return ".craft"
+	case config.ProtocolCanal:
+		return ".canal"
+	case config.ProtocolCsv:
+		return ".csv"
+	default:
+		return ".unknown"
+	}
+}