config

Documentation

Index

• Variables
• func ValidatePipelineConfig(cfg *PipelineConfigV2) error
• type Config
  • func LoadConfigFromFile(path string) *Config
  • func NewConfig() *Config
  • func NewConfigFromString(configString string) (*Config, error)
  • func (c *Config) ConfigFromFile(path string) error
  • func (c *Config) ParseCmd(arguments []string) error
• type GtmConfig
• type MongoConfigs
• type MongoConnConfig
• type MongoPosition
• type MongoSource
• type MySQLConfig
• type PipelineConfig
• type PipelineConfigV2
  • func DecodeTomlString(s string) (*PipelineConfigV2, error)
• type SourceKafkaConfig
• type SourceProbeCfg
• type SourceTiDBConfig
• type TargetMySQLWorkerConfig

Variables

var DefaultBinlogSyncerTimeout = "10s"

Functions

func ValidatePipelineConfig

func ValidatePipelineConfig(cfg *PipelineConfigV2) error

Types

type Config

type Config struct {
	*flag.FlagSet `json:"-"`

	EtcdEndpoints string `toml:"etcd-endpoints" json:"etcd-endpoints"`

	PipelineConfig *PipelineConfigV2 `toml:"pipeline" json:"pipeline"`

	// Log related configuration.
	Log logutil.LogConfig `toml:"log" json:"log"`

	HttpAddr string `toml:"http-addr" json:"http-addr"`

	PositionFile  string `toml:"position-file" json:"position-file"`
	ConfigFile    string `toml:"-" json:"-"`
	ClearPosition bool   `toml:"-" json:"-"`
	Version       bool
}

Config is the configuration.

func LoadConfigFromFile

func LoadConfigFromFile(path string) *Config

func NewConfig

func NewConfig() *Config

NewConfig creates a new config.

func NewConfigFromString

func NewConfigFromString(configString string) (*Config, error)

func (*Config) ConfigFromFile

func (c *Config) ConfigFromFile(path string) error

ConfigFromFile loads config from file.

func (*Config) ParseCmd

func (c *Config) ParseCmd(arguments []string) error

ParseCmd parses flag definitions from argument list

type GtmConfig

type GtmConfig struct {
	UseBufferDuration bool `mapstructure:"use-buffer-duration" toml:"use-buffer-duration" json:"use-buffer-duration"`
	BufferSize        int  `mapstructure:"buffer-size" toml:"buffer-size" json:"buffer-size"`
	ChannelSize       int  `mapstructure:"channel-size" toml:"channel-size" json:"channel-size"`
	BufferDurationMs  int  `mapstructure:"buffer-duration-ms" toml:"buffer-duration-ms" json:"buffer-duration-ms"`
}

type MongoConfigs

type MongoConfigs struct {
	MongoSources   []MongoSource    `toml:"mongo-sources" json:"mongo-sources"`
	PositionSource *MongoConnConfig `toml:"position-conn" json:"position-conn"`
	GtmConfig      *GtmConfig       `toml:"gtm-config" json:"gtm-config"`
}

type MongoConnConfig

type MongoConnConfig struct {
	Host     string `mapstructure:"host" toml:"host" json:"host"`
	Port     int    `mapstructure:"port" toml:"port" json:"port"`
	Username string `mapstructure:"username" toml:"username" json:"username"`
	Password string `mapstructure:"password" toml:"password" json:"password"`
	Database string `mapstructure:"database" toml:"database" json:"database"`
	Direct   bool   `mapstructure:"-" toml:"-" json:"-"`
}

type MongoPosition

type MongoPosition bson.MongoTimestamp

type MongoSource

type MongoSource struct {
	MongoConnConfig *MongoConnConfig `mapstructure:"source" toml:"source" json:"source"`
	StartPosition   *MongoPosition   `mapstructure:"start-position" toml:"start-position" json:"start-position"`
}

type MySQLConfig

type MySQLConfig struct {
	IgnoreBiDirectionalData bool                       `mapstructure:"ignore-bidirectional-data" toml:"ignore-bidirectional-data" json:"ignore-bidirectional-data"`
	Source                  *utils.DBConfig            `mapstructure:"source" toml:"source" json:"source"`
	SourceSlave             *utils.DBConfig            `mapstructure:"source-slave" toml:"source-slave" json:"source-slave"`
	StartPosition           *utils.MySQLBinlogPosition `mapstructure:"start-position" toml:"start-position" json:"start-position"`
}

type PipelineConfig

type PipelineConfig struct {
	PipelineName string `toml:"name" json:"name"`

	// Deprecated!
	// DetectTxn txn is used in: bi-directional transfer, dynamic route
	DetectTxn bool `toml:"detect-txn" json:"detect-txn"`

	// UniqueSourceName name of the server
	UniqueSourceName string `toml:"unique-source-name" json:"unique-source-name"`

	Input        string `toml:"input" json:"input"`
	Output       string `toml:"output" json:"output"`
	OutputFormat string `toml:"output-format" json:"output-format"`

	MongoConfig      *MongoConfigs     `toml:"mongo" json:"mongo"`
	MySQLConfig      *MySQLConfig      `toml:"mysql" json:"mysql"`
	SourceTiDBConfig *SourceTiDBConfig `toml:"source-tidb" json:"source-tidb"`
	SourceProbeCfg   *SourceProbeCfg   `toml:"source-probe-config" json:"source-probe-config"`

	KafkaGlobalConfig *config.KafkaGlobalConfig `toml:"kafka-global" json:"kafka-global"`

	//
	// RouteMode, DynamicKafkaRouteConfig, StaticKafkaRouteConfig, and DBRoutes
	// are route related configuration
	RouteMode string `toml:"route-mode" json:"route-mode"`

	TableConfig []*config.TableConfig `toml:"table-config" json:"table-config"`

	TargetMySQL *utils.DBConfig `toml:"target-mysql" json:"target-mysql"`

	TargetMySQLWorkerCfg *TargetMySQLWorkerConfig `toml:"target-mysql-worker" json:"target-mysql-worker"`

	//
	// internal configurations that is not exposed to users
	//
	DisableBinlogChecker bool   `toml:"-" json:"-"`
	DebugBinlog          bool   `toml:"-" json:"-"`
	BinlogSyncerTimeout  string `toml:"-" json:"-"`
}

type PipelineConfigV2

type PipelineConfigV2 struct {
	PipelineName     string                 `mapstructure:"name" toml:"name" json:"name"`
	InputPlugins     map[string]interface{} `toml:"input" json:"input" mapstructure:"input"`
	FilterPlugins    []interface{}          `mapstructure:"filters" toml:"filters" json:"filters,omitempty"`
	OutputPlugins    map[string]interface{} `mapstructure:"output" toml:"output" json:"output"`
	SchedulerPlugins map[string]interface{} `mapstructure:"scheduler" toml:"scheduler" json:"scheduler,omitempty"`
}

func DecodeTomlString

func DecodeTomlString(s string) (*PipelineConfigV2, error)

type SourceKafkaConfig

type SourceKafkaConfig struct {
	BrokerConfig config.KafkaGlobalConfig    `mapstructure:"brokers" toml:"brokers" json:"brokers"`
	GroupID      string                      `mapstructure:"group-id" toml:"group-id" json:"group-id"`
	Topics       []string                    `mapstructure:"topic" toml:"topics" json:"topics"`
	ConsumeFrom  string                      `mapstructure:"consume-from" toml:"consume-from" json:"consume-from"`
	Common       config.KafkaCommonConfig    `mapstructure:"common" toml:"common" json:"common"`
	Consumer     *config.KafkaConsumerConfig `mapstructure:"consumer" toml:"consumer" json:"consumer"`
}

type SourceProbeCfg

type SourceProbeCfg struct {
	SourceMySQL *utils.DBConfig `mapstructure:"mysql" toml:"mysql" json:"mysql"`
	Annotation  string          `mapstructure:"annotation" toml:"annotation" json:"annotation"`
}

type SourceTiDBConfig

type SourceTiDBConfig struct {
	SourceDB          *utils.DBConfig    `mapstructure:"source-db" toml:"source-db" json:"source-db"`
	SourceKafka       *SourceKafkaConfig `mapstructure:"source-kafka" toml:"source-kafka" json:"source-kafka"`
	OffsetStoreConfig *SourceProbeCfg    `mapstructure:"offset-store" toml:"offset-store" json:"offset-store"`
}

type TargetMySQLWorkerConfig

type TargetMySQLWorkerConfig struct {
	EnableDDL          bool     `toml:"enable-ddl" json:"enable-ddl"`
	UseBidirection     bool     `toml:"use-bidirection" json:"use-bidirection"`
	UseShadingProxy    bool     `toml:"use-shading-proxy" json:"use-shading-proxy"`
	SQLExecutionEngine string   `toml:"sql-execution-engine" json:"sql-execution-engine"`
	Plugins            []string `toml:"plugins" json:"plugins"`
}