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)