VTShovel - VReplication support for external databases

go/mysql/binlog_event_filepos.go

@@ -106,9 +106,11 @@ func (ev filePosQueryEvent) Query(BinlogFormat) (Query, error) {
 	}, nil
 }
 
-//----------------------------------------------------------------------------
+func (ev filePosQueryEvent) StripChecksum(f BinlogFormat) (BinlogEvent, []byte, error) {
+	return ev, nil, nil
+}
 
-var _ BinlogEvent = filePosFakeEvent{}
+//----------------------------------------------------------------------------
 
 // filePosFakeEvent is the base class for fake events.
 type filePosFakeEvent struct {
@@ -207,10 +209,6 @@ func (ev filePosFakeEvent) Rows(BinlogFormat, *TableMap) (Rows, error) {
 	return Rows{}, nil
 }
 
-func (ev filePosFakeEvent) StripChecksum(f BinlogFormat) (BinlogEvent, []byte, error) {
-	return ev, nil, nil
-}
-
 func (ev filePosFakeEvent) IsPseudo() bool {
 	return false
 }
@@ -239,6 +237,10 @@ func (ev filePosGTIDEvent) IsGTID() bool {
 	return true
 }
 
+func (ev filePosGTIDEvent) StripChecksum(f BinlogFormat) (BinlogEvent, []byte, error) {
+	return ev, nil, nil
+}
+
 func (ev filePosGTIDEvent) GTID(BinlogFormat) (GTID, bool, error) {
 	return ev.gtid, false, nil
 }

go/stats/prometheusbackend/prometheusbackend.go

@@ -82,7 +82,7 @@ func (be PromBackend) publishPrometheusMetric(name string, v expvar.Var) {
 		newMultiTimingsCollector(st, be.buildPromName(name))
 	case *stats.Histogram:
 		newHistogramCollector(st, be.buildPromName(name))
-	case *stats.String, stats.StringFunc, stats.StringMapFunc, *stats.Rates:
+	case *stats.String, stats.StringFunc, stats.StringMapFunc, *stats.Rates, *stats.RatesFunc:
 		// Silently ignore these types since they don't make sense to
 		// export to Prometheus' data model.
 	default:

go/stats/rates.go

@@ -182,3 +182,32 @@ func (rt *Rates) String() string {
 	}
 	return string(data)
 }
+
+type RatesFunc struct {
+	F    func() map[string][]float64
+	help string
+}
+
+func NewRateFunc(name string, help string, f func() map[string][]float64) *RatesFunc {
+	c := &RatesFunc{
+		F:    f,
+		help: help,
+	}
+
+	if name != "" {
+		publish(name, c)
+	}
+	return c
+}
+
+func (rf *RatesFunc) Help() string {
+	return rf.help
+}
+
+func (rf *RatesFunc) String() string {
+	data, err := json.Marshal(rf.F())
+	if err != nil {
+		data, _ = json.Marshal(err.Error())
+	}
+	return string(data)
+}

go/vt/binlog/binlogplayer/binlog_player.go

@@ -111,7 +111,7 @@ func (bps *Stats) MessageHistory() []string {
 func NewStats() *Stats {
 	bps := &Stats{}
 	bps.Timings = stats.NewTimings("", "", "")
-	bps.Rates = stats.NewRates("", bps.Timings, 15, 60e9)
+	bps.Rates = stats.NewRates("", bps.Timings, 15*60/5, 5*time.Second)
 	bps.History = history.New(3)
 	bps.SecondsBehindMaster.Set(math.MaxInt64)
 	return bps
@@ -202,6 +202,7 @@ func (blp *BinlogPlayer) applyEvents(ctx context.Context) error {
 		log.Error(err)
 		return err
 	}
+
 	blp.position = settings.StartPos
 	blp.stopPosition = settings.StopPos
 	t, err := throttler.NewThrottler(

go/vt/dbconfigs/dbconfigs.go

@@ -69,14 +69,15 @@ const (
 	// AllPrivs user should have more privileges than App (should include possibility to do
 	// schema changes and write to internal Vitess tables), but it shouldn't have SUPER
 	// privilege like Dba has.
-	AllPrivs = "allprivs"
-	Dba      = "dba"
-	Filtered = "filtered"
-	Repl     = "repl"
+	AllPrivs     = "allprivs"
+	Dba          = "dba"
+	Filtered     = "filtered"
+	Repl         = "repl"
+	ExternalRepl = "erepl"
 )
 
 // All can be used to register all flags: RegisterFlags(All...)
-var All = []string{App, AppDebug, AllPrivs, Dba, Filtered, Repl}
+var All = []string{App, AppDebug, AllPrivs, Dba, Filtered, Repl, ExternalRepl}
 
 // RegisterFlags registers the flags for the given DBConfigFlag.
 // For instance, vttablet will register client, dba and repl.
@@ -128,6 +129,7 @@ func registerPerUserFlags(dbc *userConfig, userKey string) {
 	flag.StringVar(&dbc.param.SslCert, "db-config-"+userKey+"-ssl-cert", "", "deprecated: use db_ssl_cert")
 	flag.StringVar(&dbc.param.SslKey, "db-config-"+userKey+"-ssl-key", "", "deprecated: use db_ssl_key")
 	flag.StringVar(&dbc.param.ServerName, "db-config-"+userKey+"-server_name", "", "deprecated: use db_server_name")
+	flag.StringVar(&dbc.param.Flavor, "db-config-"+userKey+"-flavor", "", "deprecated: use db_flavor")
 
 	flag.StringVar(&dbc.param.DeprecatedDBName, "db-config-"+userKey+"-dbname", "", "deprecated: dbname does not need to be explicitly configured")
 
@@ -158,16 +160,33 @@ func (dbcfgs *DBConfigs) DbaWithDB() *mysql.ConnParams {
 	return dbcfgs.makeParams(Dba, true)
 }
 
-// FilteredWithDB returns connection parameters for appdebug with dbname set.
+// FilteredWithDB returns connection parameters for filtered with dbname set.
 func (dbcfgs *DBConfigs) FilteredWithDB() *mysql.ConnParams {
 	return dbcfgs.makeParams(Filtered, true)
 }
 
-// Repl returns connection parameters for appdebug with no dbname set.
+// Repl returns connection parameters for repl with no dbname set.
 func (dbcfgs *DBConfigs) Repl() *mysql.ConnParams {
 	return dbcfgs.makeParams(Repl, false)
 }
 
+// ExternalRepl returns connection parameters for repl with no dbname set.
+func (dbcfgs *DBConfigs) ExternalRepl() *mysql.ConnParams {
+	return dbcfgs.makeParams(ExternalRepl, true)
+}
+
+// ExternalReplWithDB returns connection parameters for repl with dbname set.
+func (dbcfgs *DBConfigs) ExternalReplWithDB() *mysql.ConnParams {
+	params := dbcfgs.makeParams(ExternalRepl, true)
+	// TODO @rafael: This is a hack to allows to configure external databases by providing
+	// db-config-erepl-dbname.
+	if params.DeprecatedDBName != "" {
+		params.DbName = params.DeprecatedDBName
+		return params
+	}
+	return params
+}
+
 // AppWithDB returns connection parameters for app with dbname set.
 func (dbcfgs *DBConfigs) makeParams(userKey string, withDB bool) *mysql.ConnParams {
 	orig := dbcfgs.userConfigs[userKey]
@@ -238,8 +257,13 @@ func HasConnectionParams() bool {
 // is used to initialize the per-user conn params.
 func Init(defaultSocketFile string) (*DBConfigs, error) {
 	// The new base configs, if set, supersede legacy settings.
-	for _, uc := range dbConfigs.userConfigs {
-		if HasConnectionParams() {
+	for user, uc := range dbConfigs.userConfigs {
+		// TODO @rafael: For ExternalRepl we need to respect the provided host / port
+		// At the moment this is an snowflake user connection type that it used by
+		// vreplication to connect to external mysql hosts that are not part of a vitess
+		// cluster. In the future we need to refactor all dbconfig to support custom users
+		// in a more flexible way.
+		if HasConnectionParams() && user != ExternalRepl {
 			uc.param.Host = baseConfig.Host
 			uc.param.Port = baseConfig.Port
 			uc.param.UnixSocket = baseConfig.UnixSocket
@@ -253,7 +277,9 @@ func Init(defaultSocketFile string) (*DBConfigs, error) {
 		if baseConfig.Flags != 0 {
 			uc.param.Flags = baseConfig.Flags
 		}
-		uc.param.Flavor = baseConfig.Flavor
+		if user != ExternalRepl {
+			uc.param.Flavor = baseConfig.Flavor
+		}
 		if uc.useSSL {
 			uc.param.SslCa = baseConfig.SslCa
 			uc.param.SslCaPath = baseConfig.SslCaPath
@@ -282,12 +308,13 @@ func Init(defaultSocketFile string) (*DBConfigs, error) {
 func NewTestDBConfigs(genParams, appDebugParams mysql.ConnParams, dbName string) *DBConfigs {
 	dbcfgs := &DBConfigs{
 		userConfigs: map[string]*userConfig{
-			App:      {param: genParams},
-			AppDebug: {param: appDebugParams},
-			AllPrivs: {param: genParams},
-			Dba:      {param: genParams},
-			Filtered: {param: genParams},
-			Repl:     {param: genParams},
+			App:          {param: genParams},
+			AppDebug:     {param: appDebugParams},
+			AllPrivs:     {param: genParams},
+			Dba:          {param: genParams},
+			Filtered:     {param: genParams},
+			Repl:         {param: genParams},
+			ExternalRepl: {param: genParams},
 		},
 	}
 	dbcfgs.DBName.Set(dbName)