Adding 'worker' tag to tablet.

When a worker is working on a tablet, add a URL that
points back to the worker.

go/vt/servenv/run.go

@@ -3,13 +3,19 @@ package servenv
 import (
 	"fmt"
 	"net/http"
+	"net/url"
+	"os"
 
 	log "github.com/golang/glog"
+	"github.com/youtube/vitess/go/netutil"
 	"github.com/youtube/vitess/go/proc"
 )
 
 var (
 	onCloseHooks hooks
+
+	// filled in when calling Run or RunSecure
+	ListeningUrl url.URL
 )
 
 // Run starts listening for RPC and HTTP requests on the given port,
@@ -37,12 +43,25 @@ func RunSecure(port int, securePort int, cert, key, caCert string) {
 		log.Infof("listening on secure port %v", securePort)
 		SecureServe(fmt.Sprintf(":%d", securePort), cert, key, caCert)
 	}
+	host, err := netutil.FullyQualifiedHostname()
+	if err != nil {
+		host, err = os.Hostname()
+		if err != nil {
+			log.Fatal("os.Hostname() failed: %v", err)
+		}
+	}
+	ListeningUrl = url.URL{
+		Scheme: "http",
+		Host:   fmt.Sprintf("%v:%v", host, port),
+		Path:   "/",
+	}
 	proc.Wait()
 	Close()
 }
 
 // Close runs any registered exit hooks in parallel.
 func Close() {
+	ListeningUrl = url.URL{}
 	onCloseHooks.Fire()
 }
 

go/vt/worker/diff_utils.go

@@ -86,7 +86,7 @@ func FullTableScan(ts topo.Server, tabletAlias topo.TabletAlias, tableDefinition
 	}
 
 	sql := fmt.Sprintf("SELECT %v FROM %v %vORDER BY (%v)", strings.Join(orderedColumns(tableDefinition), ", "), tableDefinition.Name, where, strings.Join(tableDefinition.PrimaryKeyColumns, ", "))
-	log.Infof("SQL query for %v: %v", tabletAlias, sql)
+	log.Infof("SQL query for %v/%v: %v", tabletAlias, tableDefinition.Name, sql)
 	req := &tproto.Query{
 		Sql:           sql,
 		BindVariables: make(map[string]interface{}),

go/vt/worker/split_diff.go

@@ -15,6 +15,7 @@ import (
 	"github.com/youtube/vitess/go/vt/concurrency"
 	"github.com/youtube/vitess/go/vt/key"
 	"github.com/youtube/vitess/go/vt/mysqlctl"
+	"github.com/youtube/vitess/go/vt/servenv"
 	tm "github.com/youtube/vitess/go/vt/tabletmanager"
 	"github.com/youtube/vitess/go/vt/topo"
 	"github.com/youtube/vitess/go/vt/wrangler"
@@ -207,7 +208,6 @@ func (sdw *SplitDiffWorker) init() error {
 // - find one rdonly per source shard
 // - find one rdonly in destination shard
 // - mark them all as 'checker' pointing back to us
-// TODO(alainjobart) add a tag pointing back to us to the checker instances
 
 func (sdw *SplitDiffWorker) findTarget(shard string) (topo.TabletAlias, error) {
 	endPoints, err := sdw.wr.TopoServer().GetEndPoints(sdw.cell, sdw.keyspace, shard, topo.TYPE_RDONLY)
@@ -226,6 +226,18 @@ func (sdw *SplitDiffWorker) findTarget(shard string) (topo.TabletAlias, error) {
 	if err := sdw.wr.ChangeType(tabletAlias, topo.TYPE_CHECKER, false /*force*/); err != nil {
 		return topo.TabletAlias{}, err
 	}
+	ourUrl := servenv.ListeningUrl.String()
+	log.Infof("Adding tag[worker]=%v to tablet %v", ourUrl, tabletAlias)
+	if err := sdw.wr.TopoServer().UpdateTabletFields(tabletAlias, func(tablet *topo.Tablet) error {
+		if tablet.Tags == nil {
+			tablet.Tags = make(map[string]string)
+		}
+		tablet.Tags["worker"] = ourUrl
+		return nil
+	}); err != nil {
+		return topo.TabletAlias{}, err
+	}
+	wrangler.RecordTabletTagAction(sdw.cleaner, tabletAlias, "worker", "")
 
 	// Record a clean-up action to take the tablet back to rdonly.
 	// We will alter this one later on and let the tablet go back to

go/vt/wrangler/cleaner.go

@@ -158,6 +158,45 @@ func (csta ChangeSlaveTypeAction) CleanUp(wr *Wrangler) error {
 	return wr.ChangeType(csta.TabletAlias, csta.TabletType, false)
 }
 
+//
+// TabletTagAction CleanerAction
+//
+
+// TabletTagAction will add / remove a tag to a tablet. If Value is
+// empty, will remove the tag.
+type TabletTagAction struct {
+	TabletAlias topo.TabletAlias
+	Name        string
+	Value       string
+}
+
+const TabletTagActionName = "TabletTagAction"
+
+// RecordTabletTagAction records a new TabletTagAction
+// into the specified Cleaner
+func RecordTabletTagAction(cleaner *Cleaner, tabletAlias topo.TabletAlias, name, value string) {
+	cleaner.Record(TabletTagActionName, tabletAlias.String(), &TabletTagAction{
+		TabletAlias: tabletAlias,
+		Name:        name,
+		Value:       value,
+	})
+}
+
+// CleanUp is part of CleanerAction interface.
+func (tta TabletTagAction) CleanUp(wr *Wrangler) error {
+	return wr.TopoServer().UpdateTabletFields(tta.TabletAlias, func(tablet *topo.Tablet) error {
+		if tablet.Tags == nil {
+			tablet.Tags = make(map[string]string)
+		}
+		if tta.Value != "" {
+			tablet.Tags[tta.Name] = tta.Value
+		} else {
+			delete(tablet.Tags, tta.Name)
+		}
+		return nil
+	})
+}
+
 //
 // StartSlaveAction CleanerAction
 //