func TestQueryShard(t *testing.T) {
	reflected, err := bson.Marshal(&reflectQueryShard{
		Sql:           "query",
		BindVariables: map[string]interface{}{"val": int64(1)},
		Keyspace:      "keyspace",
		Shards:        []string{"shard1", "shard2"},
		TabletType:    topo.TabletType("replica"),
		Session:       &commonSession,
	})
	if err != nil {
		t.Error(err)
	}
	want := string(reflected)

	custom := QueryShard{
		Sql:           "query",
		BindVariables: map[string]interface{}{"val": int64(1)},
		Keyspace:      "keyspace",
		Shards:        []string{"shard1", "shard2"},
		TabletType:    topo.TabletType("replica"),
		Session:       &commonSession,
	}
	encoded, err := bson.Marshal(&custom)
	if err != nil {
		t.Error(err)
	}
	got := string(encoded)
	if want != got {
		t.Errorf("want\n%#v, got\n%#v", want, got)
	}

	var unmarshalled QueryShard
	err = bson.Unmarshal(encoded, &unmarshalled)
	if err != nil {
		t.Error(err)
	}
	if !reflect.DeepEqual(custom, unmarshalled) {
		t.Errorf("want \n%#v, got \n%#v", custom, unmarshalled)
	}

	unexpected, err := bson.Marshal(&badQueryShard{})
	if err != nil {
		t.Error(err)
	}
	err = bson.Unmarshal(unexpected, &unmarshalled)
	want = "Unrecognized tag Extra"
	if err == nil || want != err.Error() {
		t.Errorf("want %v, got %v", want, err)
	}
}

func TestSession(t *testing.T) {
	reflected, err := bson.Marshal(&reflectSession{
		InTransaction: true,
		ShardSessions: []*ShardSession{{
			Keyspace:      "a",
			Shard:         "0",
			TabletType:    topo.TabletType("replica"),
			TransactionId: 1,
		}, {
			Keyspace:      "b",
			Shard:         "1",
			TabletType:    topo.TabletType("master"),
			TransactionId: 2,
		}},
	})
	if err != nil {
		t.Error(err)
	}
	want := string(reflected)

	custom := commonSession
	encoded, err := bson.Marshal(&custom)
	if err != nil {
		t.Error(err)
	}
	got := string(encoded)
	if want != got {
		t.Errorf("want\n%#v, got\n%#v", want, got)
	}

	var unmarshalled Session
	err = bson.Unmarshal(encoded, &unmarshalled)
	if err != nil {
		t.Error(err)
	}
	if !reflect.DeepEqual(custom, unmarshalled) {
		t.Errorf("want \n%#v, got \n%#v", custom, unmarshalled)
	}

	unexpected, err := bson.Marshal(&badSession{})
	if err != nil {
		t.Error(err)
	}
	err = bson.Unmarshal(unexpected, &unmarshalled)
	want = "Unrecognized tag Extra"
	if err == nil || want != err.Error() {
		t.Errorf("want %v, got %v", want, err)
	}
}

func getSrvKeyspace(rpcClient *rpcplus.Client, cell, keyspace string, verbose bool) {
	req := &topo.GetSrvKeyspaceArgs{
		Cell:     cell,
		Keyspace: keyspace,
	}
	reply := &topo.SrvKeyspace{}
	if err := rpcClient.Call("TopoReader.GetSrvKeyspace", req, reply); err != nil {
		log.Fatalf("TopoReader.GetSrvKeyspace error: %v", err)
	}
	if verbose {
		tabletTypes := make([]string, 0, len(reply.Partitions))
		for t, _ := range reply.Partitions {
			tabletTypes = append(tabletTypes, string(t))
		}
		sort.Strings(tabletTypes)
		for _, t := range tabletTypes {
			println(fmt.Sprintf("Partitions[%v] =", t))
			for i, s := range reply.Partitions[topo.TabletType(t)].Shards {
				println(fmt.Sprintf("  Shards[%v]=%v", i, s.KeyRange.String()))
			}
		}
		for i, s := range reply.Shards {
			println(fmt.Sprintf("Shards[%v]=%v", i, s.KeyRange.String()))
		}
		for i, t := range reply.TabletTypes {
			println(fmt.Sprintf("TabletTypes[%v] = %v", i, t))
		}
	}
}

// UnmarshalBson unmarshals QueryShard from buf.
func (qrs *QueryShard) UnmarshalBson(buf *bytes.Buffer) {
	bson.Next(buf, 4)

	kind := bson.NextByte(buf)
	for kind != bson.EOO {
		key := bson.ReadCString(buf)
		switch key {
		case "Sql":
			qrs.Sql = bson.DecodeString(buf, kind)
		case "BindVariables":
			qrs.BindVariables = tproto.DecodeBindVariablesBson(buf, kind)
		case "Keyspace":
			qrs.Keyspace = bson.DecodeString(buf, kind)
		case "TabletType":
			qrs.TabletType = topo.TabletType(bson.DecodeString(buf, kind))
		case "Shards":
			qrs.Shards = bson.DecodeStringArray(buf, kind)
		case "Session":
			qrs.Session = new(Session)
			qrs.Session.UnmarshalBson(buf)
		default:
			panic(bson.NewBsonError("Unrecognized tag %s", key))
		}
		kind = bson.NextByte(buf)
	}
}

func (sqs *StreamQueryKeyRange) UnmarshalBson(buf *bytes.Buffer) {
	bson.Next(buf, 4)

	kind := bson.NextByte(buf)
	for kind != bson.EOO {
		keyName := bson.ReadCString(buf)
		switch keyName {
		case "Sql":
			sqs.Sql = bson.DecodeString(buf, kind)
		case "BindVariables":
			sqs.BindVariables = tproto.DecodeBindVariablesBson(buf, kind)
		case "Keyspace":
			sqs.Keyspace = bson.DecodeString(buf, kind)
		case "KeyRange":
			sqs.KeyRange = bson.DecodeString(buf, kind)
		case "TabletType":
			sqs.TabletType = topo.TabletType(bson.DecodeString(buf, kind))
		case "Session":
			sqs.Session = new(Session)
			sqs.Session.UnmarshalBson(buf)
		default:
			panic(bson.NewBsonError("Unrecognized tag %s", keyName))
		}
		kind = bson.NextByte(buf)
	}
}

// for direct zk connection: vtzk://host:port/cell/keyspace/tabletType
// we always use a MetaConn, host and port are ignored.
// the driver name dictates if we use zk or zkocc, and streaming or not
//
// if user and password are specified in the URL, they will be used
// for each DB connection to the tablet's vttablet processes
func (driver *sDriver) Open(name string) (sc db.Conn, err error) {
	if !strings.HasPrefix(name, "vtzk://") {
		// add a default protocol talking to zk
		name = "vtzk://" + name
	}
	u, err := url.Parse(name)
	if err != nil {
		return nil, err
	}

	dbi, tabletType := path.Split(u.Path)
	dbi = strings.Trim(dbi, "/")
	tabletType = strings.Trim(tabletType, "/")
	cell, keyspace := path.Split(dbi)
	cell = strings.Trim(cell, "/")
	keyspace = strings.Trim(keyspace, "/")
	var user, password string
	if u.User != nil {
		user = u.User.Username()
		var ok bool
		password, ok = u.User.Password()
		if !ok {
			return nil, fmt.Errorf("vt: need a password if a user is specified")
		}
	}
	return Dial(driver.ts, cell, keyspace, topo.TabletType(tabletType), driver.stream, tablet.DefaultTimeout, user, password)
}

// UnmarshalBson unmarshals BatchQueryShard from buf.
func (bqs *BatchQueryShard) UnmarshalBson(buf *bytes.Buffer, kind byte) {
	bson.VerifyObject(kind)
	bson.Next(buf, 4)

	kind = bson.NextByte(buf)
	for kind != bson.EOO {
		keyName := bson.ReadCString(buf)
		switch keyName {
		case "Queries":
			bqs.Queries = tproto.DecodeQueriesBson(buf, kind)
		case "Keyspace":
			bqs.Keyspace = bson.DecodeString(buf, kind)
		case "Shards":
			bqs.Shards = bson.DecodeStringArray(buf, kind)
		case "TabletType":
			bqs.TabletType = topo.TabletType(bson.DecodeString(buf, kind))
		case "Session":
			if kind != bson.Null {
				bqs.Session = new(Session)
				bqs.Session.UnmarshalBson(buf, kind)
			}
		default:
			bson.Skip(buf, kind)
		}
		kind = bson.NextByte(buf)
	}
}

func TestHashResolve(t *testing.T) {
	hind := NewHashIndex(TEST_SHARDED, new(sandboxTopo), "")
	nn, _ := sqltypes.BuildNumeric("11")
	ks, shards, err := hind.Resolve(topo.TabletType("master"), []interface{}{1, int32(2), int64(3), uint(4), uint32(5), uint64(6), nn})
	if err != nil {
		t.Error(err)
	}
	want := []string{"-20", "-20", "40-60", "c0-e0", "60-80", "e0-", "a0-c0"}
	if !reflect.DeepEqual(shards, want) {
		t.Errorf("got\n%#v, want\n%#v", shards, want)
	}
	if ks != TEST_SHARDED {
		t.Errorf("got %v, want TEST_SHARDED", ks)
	}
	_, _, err = hind.Resolve(topo.TabletType("master"), []interface{}{"aa"})
	wantErr := "unexpected type for aa: string"
	if err == nil || err.Error() != wantErr {
		t.Errorf("got %v, want %v", err, wantErr)
	}
}

func (agent *ActionAgent) initHeathCheck() {
	if !agent.IsRunningHealthCheck() {
		log.Infof("No target_tablet_type specified, disabling any health check")
		return
	}

	log.Infof("Starting periodic health check every %v with target_tablet_type=%v", *healthCheckInterval, *targetTabletType)
	t := timer.NewTimer(*healthCheckInterval)
	servenv.OnTerm(func() {
		// When we enter lameduck mode, we want to not call
		// the health check any more. After this returns, we
		// are guaranteed to not call it.
		log.Info("Stopping periodic health check timer")
		t.Stop()

		// Now we can finish up and force ourselves to not healthy.
		agent.terminateHealthChecks(topo.TabletType(*targetTabletType))
	})
	t.Start(func() {
		agent.runHealthCheck(topo.TabletType(*targetTabletType))
	})
}

// handleExplorerRedirect returns the redirect target URL.
func handleExplorerRedirect(r *http.Request) (string, error) {
	keyspace := r.FormValue("keyspace")
	shard := r.FormValue("shard")
	cell := r.FormValue("cell")

	switch r.FormValue("type") {
	case "keyspace":
		if keyspace == "" {
			return "", errors.New("keyspace is required for this redirect")
		}
		return explorer.GetKeyspacePath(keyspace), nil
	case "shard":
		if keyspace == "" || shard == "" {
			return "", errors.New("keyspace and shard are required for this redirect")
		}
		return explorer.GetShardPath(keyspace, shard), nil
	case "srv_keyspace":
		if keyspace == "" || cell == "" {
			return "", errors.New("keyspace and cell are required for this redirect")
		}
		return explorer.GetSrvKeyspacePath(cell, keyspace), nil
	case "srv_shard":
		if keyspace == "" || shard == "" || cell == "" {
			return "", errors.New("keyspace, shard, and cell are required for this redirect")
		}
		return explorer.GetSrvShardPath(cell, keyspace, shard), nil
	case "srv_type":
		tabletType := r.FormValue("tablet_type")
		if keyspace == "" || shard == "" || cell == "" || tabletType == "" {
			return "", errors.New("keyspace, shard, cell, and tablet_type are required for this redirect")
		}
		return explorer.GetSrvTypePath(cell, keyspace, shard, topo.TabletType(tabletType)), nil
	case "tablet":
		alias := r.FormValue("alias")
		if alias == "" {
			return "", errors.New("alias is required for this redirect")
		}
		tabletAlias, err := topo.ParseTabletAliasString(alias)
		if err != nil {
			return "", fmt.Errorf("bad tablet alias %q: %v", alias, err)
		}
		return explorer.GetTabletPath(tabletAlias), nil
	case "replication":
		if keyspace == "" || shard == "" || cell == "" {
			return "", errors.New("keyspace, shard, and cell are required for this redirect")
		}
		return explorer.GetReplicationSlaves(cell, keyspace, shard), nil
	default:
		return "", errors.New("bad redirect type")
	}
}

func initHeathCheck(agent *tabletmanager.ActionAgent) {
	if *targetTabletType == "" {
		log.Infof("No target_tablet_type specified, disabling any health check")
		return
	}

	log.Infof("Starting up periodic health check every %v with target_tablet_type=%v", *healthCheckInterval, *targetTabletType)
	go func() {
		t := time.NewTicker(*healthCheckInterval)
		for _ = range t.C {
			agent.RunHealthCheck(topo.TabletType(*targetTabletType))
		}
	}()
}

func (zkts *Server) GetSrvTabletTypesPerShard(cell, keyspace, shard string) ([]topo.TabletType, error) {
	zkSgShardPath := zkPathForVtShard(cell, keyspace, shard)
	children, _, err := zkts.zconn.Children(zkSgShardPath)
	if err != nil {
		if zookeeper.IsError(err, zookeeper.ZNONODE) {
			err = topo.ErrNoNode
		}
		return nil, err
	}
	result := make([]topo.TabletType, len(children))
	for i, tt := range children {
		result[i] = topo.TabletType(tt)
	}
	return result, nil
}

func (spm *SessionParams) UnmarshalBson(buf *bytes.Buffer) {
	bson.Next(buf, 4)

	kind := bson.NextByte(buf)
	for kind != bson.EOO {
		key := bson.ReadCString(buf)
		switch key {
		case "TabletType":
			spm.TabletType = topo.TabletType(bson.DecodeString(buf, kind))
		default:
			panic(bson.NewBsonError("Unrecognized tag %s", key))
		}
		kind = bson.NextByte(buf)
	}
}

func TestSessionParams(t *testing.T) {
	reflected, err := bson.Marshal(&reflectSessionParams{topo.TabletType("master")})
	if err != nil {
		t.Error(err)
	}
	want := string(reflected)

	custom := SessionParams{topo.TabletType("master")}
	encoded, err := bson.Marshal(&custom)
	if err != nil {
		t.Error(err)
	}
	got := string(encoded)
	if want != got {
		t.Errorf("want\n%#v, got\n%#v", want, got)
	}

	var unmarshalled SessionParams
	err = bson.Unmarshal(encoded, &unmarshalled)
	if err != nil {
		t.Error(err)
	}
	if custom != unmarshalled {
		t.Errorf("want %v, got %#v", custom, unmarshalled)
	}

	unexpected, err := bson.Marshal(&badSessionParams{})
	if err != nil {
		t.Error(err)
	}
	err = bson.Unmarshal(unexpected, &unmarshalled)
	want = "Unrecognized tag Extra"
	if err == nil || want != err.Error() {
		t.Errorf("want %v, got %v", want, err)
	}
}

func getEndPoints(rpcClient *rpcplus.Client, cell, keyspace, shard, tabletType string, verbose bool) {
	req := &topo.GetEndPointsArgs{
		Cell:       cell,
		Keyspace:   keyspace,
		Shard:      shard,
		TabletType: topo.TabletType(tabletType),
	}
	reply := &topo.EndPoints{}
	if err := rpcClient.Call("TopoReader.GetEndPoints", req, reply); err != nil {
		log.Fatalf("TopoReader.GetEndPoints error: %v", err)
	}
	if verbose {
		for i, e := range reply.Entries {
			println(fmt.Sprintf("Entries[%v] = %v %v", i, e.Uid, e.Host))
		}
	}
}

// for direct zk connection: vtzk://host:port/cell/keyspace/tabletType
// we always use a MetaConn, host and port are ignored.
// the driver name dictates if we use zk or zkocc, and streaming or not
func (driver *sDriver) Open(name string) (sc db.Conn, err error) {
	if !strings.HasPrefix(name, "vtzk://") {
		// add a default protocol talking to zk
		name = "vtzk://" + name
	}
	u, err := url.Parse(name)
	if err != nil {
		return nil, err
	}

	dbi, tabletType := path.Split(u.Path)
	dbi = strings.Trim(dbi, "/")
	tabletType = strings.Trim(tabletType, "/")
	cell, keyspace := path.Split(dbi)
	cell = strings.Trim(cell, "/")
	keyspace = strings.Trim(keyspace, "/")
	return Dial(driver.ts, cell, keyspace, topo.TabletType(tabletType), driver.stream, tablet.DefaultTimeout)
}

func (zkts *Server) GetSrvTabletTypesPerShard(cell, keyspace, shard string) ([]topo.TabletType, error) {
	zkSgShardPath := zkPathForVtShard(cell, keyspace, shard)
	children, _, err := zkts.zconn.Children(zkSgShardPath)
	if err != nil {
		if zookeeper.IsError(err, zookeeper.ZNONODE) {
			err = topo.ErrNoNode
		}
		return nil, err
	}
	result := make([]topo.TabletType, 0, len(children))
	for _, tt := range children {
		// these two are used for locking
		if tt == "action" || tt == "actionlog" {
			continue
		}
		result = append(result, topo.TabletType(tt))
	}
	return result, nil
}

// GetSrvTabletTypesPerShard implements topo.Server.
func (s *Server) GetSrvTabletTypesPerShard(ctx context.Context, cellName, keyspace, shard string) ([]topo.TabletType, error) {
	cell, err := s.getCell(cellName)
	if err != nil {
		return nil, err
	}

	resp, err := cell.Get(srvShardDirPath(keyspace, shard), false /* sort */, false /* recursive */)
	if err != nil {
		return nil, convertError(err)
	}
	if resp.Node == nil {
		return nil, ErrBadResponse
	}

	tabletTypes := make([]topo.TabletType, 0, len(resp.Node.Nodes))
	for _, n := range resp.Node.Nodes {
		tabletTypes = append(tabletTypes, topo.TabletType(path.Base(n.Key)))
	}
	return tabletTypes, nil
}

//.........这里部分代码省略.........
			}
			return ts.GetTabletsByCell(ctx, cell)
		}

		// Get tablet health.
		if parts := strings.Split(tabletPath, "/"); len(parts) == 2 && parts[1] == "health" {
			tabletAlias, err := topo.ParseTabletAliasString(parts[0])
			if err != nil {
				return nil, err
			}
			return tabletHealthCache.Get(ctx, tabletAlias)
		}

		tabletAlias, err := topo.ParseTabletAliasString(tabletPath)
		if err != nil {
			return nil, err
		}

		// Perform an action on a tablet.
		if r.Method == "POST" {
			if err := r.ParseForm(); err != nil {
				return nil, err
			}
			action := r.FormValue("action")
			if action == "" {
				return nil, errors.New("must specify action")
			}
			return actions.ApplyTabletAction(ctx, action, tabletAlias, r), nil
		}

		// Get the tablet record.
		return ts.GetTablet(ctx, tabletAlias)
	})

	// EndPoints
	handleCollection("endpoints", func(r *http.Request) (interface{}, error) {
		// We expect cell/keyspace/shard/tabletType.
		epPath := getItemPath(r.URL.Path)
		parts := strings.Split(epPath, "/")
		if len(parts) != 4 {
			return nil, fmt.Errorf("invalid cell/keyspace/shard/tabletType: %q", epPath)
		}

		if parts[3] == "" {
			// tabletType is empty, so list the tablet types.
			return ts.GetSrvTabletTypesPerShard(ctx, parts[0], parts[1], parts[2])
		}

		// Get the endpoints object for a specific type.
		ep, _, err := ts.GetEndPoints(ctx, parts[0], parts[1], parts[2], topo.TabletType(parts[3]))
		return ep, err
	})

	// Schema Change
	http.HandleFunc(apiPrefix+"schema/apply", func(w http.ResponseWriter, r *http.Request) {
		req := struct{ Keyspace, SQL string }{}
		if err := unmarshalRequest(r, &req); err != nil {
			httpErrorf(w, r, "can't unmarshal request: %v", err)
			return
		}

		executor := schemamanager.NewTabletExecutor(
			tmclient.NewTabletManagerClient(),
			ts)

		schemamanager.Run(ctx,
			schemamanager.NewUIController(req.SQL, req.Keyspace, w), executor)
	})

	// VSchema
	http.HandleFunc(apiPrefix+"vschema/", func(w http.ResponseWriter, r *http.Request) {
		schemafier, ok := ts.(topo.Schemafier)
		if !ok {
			httpErrorf(w, r, "%T doesn't support schemafier API", ts)
			return
		}

		// Save VSchema
		if r.Method == "POST" {
			vschema, err := ioutil.ReadAll(r.Body)
			if err != nil {
				httpErrorf(w, r, "can't read request body: %v", err)
				return
			}
			if err := schemafier.SaveVSchema(ctx, string(vschema)); err != nil {
				httpErrorf(w, r, "can't save vschema: %v", err)
			}
			return
		}

		// Get VSchema
		vschema, err := schemafier.GetVSchema(ctx)
		if err != nil {
			httpErrorf(w, r, "can't get vschema: %v", err)
			return
		}
		w.Header().Set("Content-Type", jsonContentType)
		w.Write([]byte(vschema))
	})
}

func TestKeyspaceIdBatchQuery(t *testing.T) {
	reflected, err := bson.Marshal(&reflectKeyspaceIdBatchQuery{
		Queries: []reflectBoundKeyspaceIdQuery{{
			Sql:           "query",
			BindVariables: map[string]interface{}{"val": int64(1)},
			Keyspace:      "keyspace",
			KeyspaceIds:   []kproto.KeyspaceId{kproto.KeyspaceId("10"), kproto.KeyspaceId("20")},
		}},
		Session: &Session{InTransaction: true,
			ShardSessions: []*ShardSession{{
				Keyspace:      "a",
				Shard:         "0",
				TabletType:    topo.TabletType("replica"),
				TransactionId: 1,
			}, {
				Keyspace:      "b",
				Shard:         "1",
				TabletType:    topo.TabletType("master"),
				TransactionId: 2,
			}},
		},
	})
	if err != nil {
		t.Error(err)
	}
	want := string(reflected)

	custom := KeyspaceIdBatchQuery{
		Queries: []BoundKeyspaceIdQuery{{
			Sql:           "query",
			BindVariables: map[string]interface{}{"val": int64(1)},
			Keyspace:      "keyspace",
			KeyspaceIds:   []kproto.KeyspaceId{kproto.KeyspaceId("10"), kproto.KeyspaceId("20")},
		}},
		Session: &commonSession,
	}
	encoded, err := bson.Marshal(&custom)
	if err != nil {
		t.Error(err)
	}
	got := string(encoded)
	if want != got {
		t.Errorf("want\n%+v, got\n%+v", want, got)
	}

	var unmarshalled KeyspaceIdBatchQuery
	err = bson.Unmarshal(encoded, &unmarshalled)
	if err != nil {
		t.Error(err)
	}
	if !reflect.DeepEqual(custom, unmarshalled) {
		t.Errorf("want \n%+v, got \n%+v", custom, unmarshalled)
	}

	extra, err := bson.Marshal(&extraKeyspaceIdBatchQuery{})
	if err != nil {
		t.Error(err)
	}
	err = bson.Unmarshal(extra, &unmarshalled)
	if err != nil {
		t.Error(err)
	}
}