VTShovel - VReplication support for external databases

go/cmd/vtcombo/main.go

@@ -118,7 +118,7 @@ func main() {
  }
 
  // vtgate configuration and init
- resilientServer := srvtopo.NewResilientServer(ts, "ResilientSrvTopoServer")
+ resilientServer := srvtopo.NewResilientServer(ts, "ResilientSrvTopoServer", true)
  healthCheck := discovery.NewHealthCheck(1*time.Millisecond /*retryDelay*/, 1*time.Hour /*healthCheckTimeout*/)
  tabletTypesToWait := []topodatapb.TabletType{
  topodatapb.TabletType_MASTER,

go/cmd/vtgate/vtgate.go

@@ -66,7 +66,7 @@ func main() {
  ts := topo.Open()
  defer ts.Close()
 
- resilientServer = srvtopo.NewResilientServer(ts, "ResilientSrvTopoServer")
+ resilientServer = srvtopo.NewResilientServer(ts, "ResilientSrvTopoServer", true)
 
  healthCheck = discovery.NewHealthCheck(*healthCheckRetryDelay, *healthCheckTimeout)
  healthCheck.RegisterStats()

go/mysql/flavor.go

@@ -62,6 +62,9 @@ type flavor interface {
  // stopSlave returns the command to stop the slave.
  stopSlaveCommand() string
 
+ // sendBinlogFileDumpCommand sends the packet required to start streaming from file:post
+ sendBinlogFileDumpCommand(c *Conn, slaveID uint32, binlogFilename string, pos uint32) error
+
  // sendBinlogDumpCommand sends the packet required to start
  // dumping binlogs from the specified location.
  sendBinlogDumpCommand(c *Conn, slaveID uint32, startPos Position) error
@@ -163,6 +166,13 @@ func (c *Conn) StopSlaveCommand() string {
  return c.flavor.stopSlaveCommand()
 }
 
+// SendBinlogFileDumpCommand sends the flavor-specific version of
+// the COM_BINLOG_DUMP command to start dumping raw binlog
+// events over a slave connection, starting at a given file position.
+func (c *Conn) SendBinlogFileDumpCommand(slaveID uint32, binlogFilename string, pos uint32) error {
+ return c.flavor.sendBinlogFileDumpCommand(c, slaveID, binlogFilename, pos)
+}
+
 // SendBinlogDumpCommand sends the flavor-specific version of
 // the COM_BINLOG_DUMP command to start dumping raw binlog
 // events over a slave connection, starting at a given GTID.

go/mysql/flavor_mariadb.go

@@ -55,6 +55,11 @@ func (mariadbFlavor) stopSlaveCommand() string {
  return "STOP SLAVE"
 }
 
+// sendBinlogFileDumpCommand is part of the Flavor interface.
+func (mariadbFlavor) sendBinlogFileDumpCommand(c *Conn, slaveID uint32, binlogFilename string, pos uint32) error {
+ panic("filename binglog not supported for mariadb")
+}
+
 // sendBinlogDumpCommand is part of the Flavor interface.
 func (mariadbFlavor) sendBinlogDumpCommand(c *Conn, slaveID uint32, startPos Position) error {
  // Tell the server that we understand GTIDs by setting our slave

go/mysql/flavor_mysql.go

@@ -53,6 +53,11 @@ func (mysqlFlavor) stopSlaveCommand() string {
  return "STOP SLAVE"
 }
 
+// sendBinlogDumpCommand is part of the Flavor interface.
+func (mysqlFlavor) sendBinlogFileDumpCommand(c *Conn, slaveID uint32, binlogFilename string, pos uint32) error {
+ return c.WriteComBinlogDump(slaveID, binlogFilename, pos, 0)
+}
+
 // sendBinlogDumpCommand is part of the Flavor interface.
 func (mysqlFlavor) sendBinlogDumpCommand(c *Conn, slaveID uint32, startPos Position) error {
  gtidSet, ok := startPos.GTIDSet.(Mysql56GTIDSet)

go/mysql/replication_position.go

@@ -19,6 +19,7 @@ package mysql
 import (
  "encoding/json"
  "fmt"
+ "strconv"
  "strings"
 
  "vitess.io/vitess/go/vt/proto/vtrpc"
@@ -35,6 +36,12 @@ const (
  MaximumPositionSize = 64000
 )
 
+// BinlogFilePos used to encode filename:pos.
+type BinlogFilePos struct {
+ Name string
+ Pos uint32
+}
+
 // Position represents the information necessary to describe which
 // transactions a server has seen, so that it can request a replication stream
 // from a new master that picks up where it left off.
@@ -120,6 +127,28 @@ func EncodePosition(rp Position) string {
  return fmt.Sprintf("%s/%s", rp.GTIDSet.Flavor(), rp.GTIDSet.String())
 }
 
+// ParseFilePosition converts a string in the format file:pos
+// to BinlogFilePos
+func ParseFilePosition(s string) (rp BinlogFilePos, err error) {
+ if s == "" {
+ return rp, vterrors.Errorf(vtrpc.Code_INTERNAL, "parse error: unknown file:pos format %#v", s)
+ }
+
+ parts := strings.SplitN(s, ":", 2)
+ if len(parts) != 2 {
+ return rp, vterrors.Errorf(vtrpc.Code_INTERNAL, "parse error: unknown file:pos format %#v", s)
+ }
+
+ pos, err := strconv.Atoi(parts[1])
+ if err != nil {
+ return rp, vterrors.Errorf(vtrpc.Code_INTERNAL, "parse error: pos is not a valid int %#v", s)
+ }
+
+ rp.Name = parts[0]
+ rp.Pos = uint32(pos)
+ return rp, nil
+}
+
 // DecodePosition converts a string in the format returned by
 // EncodePosition back into a Position value with the
 // correct underlying flavor.

go/mysql/replication_position_test.go

@@ -208,6 +208,48 @@ func TestPositionAppendToZero(t *testing.T) {
  }
 }
 
+func TestParseFilePositionInvalidInput(t *testing.T) {
+ input := "filenameinvalidpos"
+ rp, err := ParseFilePosition(input)
+ if err == nil {
+ t.Errorf("ParseFilePosition(%#v) expected error, got : %#v", input, rp)
+ }
+
+ want := `parse error: unknown file:pos format`
+ got, ok := err.(error)
+ if !ok || !strings.HasPrefix(got.Error(), want) {
+ t.Errorf("wrong error, got %#v, want %#v", got, want)
+ }
+}
+
+func TestParseFilePositionInvalidPos(t *testing.T) {
+ input := "filename:invalidpos"
+ rp, err := ParseFilePosition(input)
+ if err == nil {
+ t.Errorf("ParseFilePosition(%#v) expected error, got : %#v", input, rp)
+ }
+
+ want := `parse error: pos is not a valid`
+ got, ok := err.(error)
+ if !ok || !strings.HasPrefix(got.Error(), want) {
+ t.Errorf("wrong error, got %#v, want %#v", got, want)
+ }
+}
+
+func TestParseFilePosition(t *testing.T) {
+ input := "filename:2343"
+ want := BinlogFilePos{Name: "filename", Pos: 2343}
+ got, err := ParseFilePosition(input)
+ if err != nil {
+ t.Errorf("ParseFilePosition(%#v) unexpected error: %#v", input, err)
+ }
+
+ if got.Name != want.Name || got.Pos != want.Pos {
+ t.Errorf("ParseFilePosition(%#v) = %#v, want %#v", input, got, want)
+ }
+
+}
+
 func TestMustParsePosition(t *testing.T) {
  flavor := "fake flavor"
  gtidSetParsers[flavor] = func(s string) (GTIDSet, error) {

go/stats/prometheusbackend/prometheusbackend.go

@@ -66,7 +66,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,7 +202,10 @@ func (blp *BinlogPlayer) applyEvents(ctx context.Context) error {
  log.Error(err)
  return err
  }
- blp.position = settings.StartPos
+
+ if !settings.GtidStartPos.IsZero() {
+ blp.position = settings.GtidStartPos
+ }
  blp.stopPosition = settings.StopPos
  t, err := throttler.NewThrottler(
  fmt.Sprintf("BinlogPlayer/%d", blp.uid),
@@ -517,11 +520,12 @@ func SetVReplicationState(dbClient DBClient, uid uint32, state, message string)
 
 // VRSettings contains the settings of a vreplication table.
 type VRSettings struct {
- StartPos mysql.Position
+ StartPos string
  StopPos mysql.Position
  MaxTPS int64
  MaxReplicationLag int64
  State string
+ GtidStartPos mysql.Position
 }
 
 // ReadVRSettings retrieves the throttler settings for
@@ -546,17 +550,18 @@ func ReadVRSettings(dbClient DBClient, uid uint32) (VRSettings, error) {
  if err != nil {
  return VRSettings{}, fmt.Errorf("failed to parse max_replication_lag column: %v", err)
  }
- startPos, err := mysql.DecodePosition(vrRow[0].ToString())
- if err != nil {
-  return VRSettings{}, fmt.Errorf("failed to parse pos column: %v", err)
- }
+ startPos := vrRow[0].ToString()
+ // TODO: This will be removed when we start using filename:pos flavor and everythign will by a proper enconded mysql.Position
+ gtidStartPos, _ := mysql.DecodePosition(startPos)
+
  stopPos, err := mysql.DecodePosition(vrRow[1].ToString())
  if err != nil {
  return VRSettings{}, fmt.Errorf("failed to parse stop_pos column: %v", err)
  }
 
  return VRSettings{
  StartPos: startPos,
+ GtidStartPos: gtidStartPos,
  StopPos: stopPos,
  MaxTPS: maxTPS,
  MaxReplicationLag: maxReplicationLag,

go/vt/binlog/slave_connection.go

@@ -127,6 +127,26 @@ func (sc *SlaveConnection) StartBinlogDumpFromPosition(ctx context.Context, star
  return sc.streamEvents(ctx), nil
 }
 
+// StartBinlogDumpFromFilePosition requests a replication binlog dump from
+// the master mysqld at the given binlog filename:pos and then sends binlog
+// events to the provided channel.
+// The stream will continue in the background, waiting for new events if
+// necessary, until the connection is closed, either by the master or
+// by canceling the context.
+//
+// Note the context is valid and used until eventChan is closed.
+func (sc *SlaveConnection) StartBinlogDumpFromFilePosition(ctx context.Context, binlogFilename string, pos uint32) (<-chan mysql.BinlogEvent, error) {
+ ctx, sc.cancel = context.WithCancel(ctx)
+
+ log.Infof("sending binlog file dump command: binlogfilename=%v, pos=%v, slaveID=%v", binlogFilename, pos, sc.slaveID)
+ if err := sc.SendBinlogFileDumpCommand(sc.slaveID, binlogFilename, pos); err != nil {
+ log.Errorf("couldn't send binlog dump command: %v", err)
+ return nil, err
+ }
+
+ return sc.streamEvents(ctx), nil
+}
+
 // streamEvents returns a channel on which events are streamed.
 func (sc *SlaveConnection) streamEvents(ctx context.Context) chan mysql.BinlogEvent {
  // FIXME(alainjobart) I think we can use a buffered channel for better performance.

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.
@@ -157,16 +158,28 @@ 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, false)
+}
+
+// ExternalReplWithDB returns connection parameters for repl with dbname set.
+func (dbcfgs *DBConfigs) ExternalReplWithDB() *mysql.ConnParams {
+ params := dbcfgs.makeParams(ExternalRepl, false)
+ params.DbName = params.DeprecatedDBName
+ 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]
@@ -237,8 +250,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