func init() {
	servenv.OnRun(func() {
		if servenv.GRPCCheckServiceMap("throttler") {
			StartServer(servenv.GRPCServer, throttler.GlobalManager)
		}
	})
}

func init() {
	servenv.OnRun(func() {
		if *allowedReplicationLag > 0 {
			health.Register("replication_reporter", mysqlctl.MySQLReplicationLag(agent.Mysqld, *allowedReplicationLag))
		}
	})
}

func init() {
	servenv.OnRun(func() {
		servenv.AddStatusPart("Topology Cache", topoTemplate, func() interface{} {
			return resilientSrvTopoServer.CacheStatus()
		})
	})
}

func init() {
	servenv.OnRun(func() {
		servenv.AddStatusPart("Tablet", tabletTemplate, func() interface{} {
			return map[string]interface{}{
				"Tablet":            agent.Tablet(),
				"BlacklistedTables": agent.BlacklistedTables(),
			}
		})
		if agent.IsRunningHealthCheck() {
			servenv.AddStatusFuncs(template.FuncMap{
				"github_com_youtube_vitess_health_html_name": healthHTMLName,
			})
			servenv.AddStatusPart("Health", healthTemplate, func() interface{} {
				return &healthStatus{Records: agent.History.Records()}
			})
		}
		tabletserver.AddStatusPart()
		servenv.AddStatusPart("Binlog Player", binlogTemplate, func() interface{} {
			return agent.BinlogPlayerMap.Status()
		})
		if onStatusRegistered != nil {
			onStatusRegistered()
		}
	})
}

func init() {
	servenv.OnRun(func() {
		if *disableActiveReparents {
			return
		}

		addCommand("Tablets", command{
			"DemoteMaster",
			commandDemoteMaster,
			"<tablet alias>",
			"Demotes a master tablet."})
		addCommand("Tablets", command{
			"ReparentTablet",
			commandReparentTablet,
			"<tablet alias>",
			"Reparent a tablet to the current master in the shard. This only works if the current slave position matches the last known reparent action."})

		addCommand("Shards", command{
			"InitShardMaster",
			commandInitShardMaster,
			"[-force] [-wait_slave_timeout=<duration>] <keyspace/shard> <tablet alias>",
			"Sets the initial master for a shard. Will make all other tablets in the shard slaves of the provided master. WARNING: this could cause data loss on an already replicating shard. PlannedReparentShard or EmergencyReparentShard should be used instead."})
		addCommand("Shards", command{
			"PlannedReparentShard",
			commandPlannedReparentShard,
			"-keyspace_shard=<keyspace/shard> [-new_master=<tablet alias>] [-avoid_master=<tablet alias>]",
			"Reparents the shard to the new master, or away from old master. Both old and new master need to be up and running."})
		addCommand("Shards", command{
			"EmergencyReparentShard",
			commandEmergencyReparentShard,
			"-keyspace_shard=<keyspace/shard> -new_master=<tablet alias>",
			"Reparents the shard to the new master. Assumes the old master is dead and not responsding."})
	})
}

func init() {
	servenv.OnRun(func() {
		if servenv.GRPCCheckServiceMap("vtctl") {
			grpcvtctlserver.StartServer(servenv.GRPCServer, ts)
		}
	})
}

func main() {
	defer exit.Recover()

	flag.Parse()
	servenv.Init()

	if initFakeZK != nil {
		initFakeZK()
	}
	ts := topo.GetServer()
	defer topo.CloseServers()

	resilientSrvTopoServer = vtgate.NewResilientSrvTopoServer(ts, "ResilientSrvTopoServer")

	healthCheck = discovery.NewHealthCheck(*connTimeoutTotal, *healthCheckRetryDelay, *healthCheckTimeout, "" /* statsSuffix */)

	tabletTypes := make([]topodatapb.TabletType, 0, 1)
	if len(*tabletTypesToWait) != 0 {
		for _, ttStr := range strings.Split(*tabletTypesToWait, ",") {
			tt, err := topoproto.ParseTabletType(ttStr)
			if err != nil {
				log.Errorf("unknown tablet type: %v", ttStr)
				continue
			}
			tabletTypes = append(tabletTypes, tt)
		}
	}
	vtg := vtgate.Init(context.Background(), healthCheck, ts, resilientSrvTopoServer, *cell, *retryDelay, *retryCount, *connTimeoutTotal, *connTimeoutPerConn, *connLife, tabletTypes, *maxInFlight, *testGateway)

	servenv.OnRun(func() {
		addStatusParts(vtg)
	})
	servenv.RunDefault()
}

func main() {
	defer exit.Recover()

	flag.Parse()
	servenv.Init()

	ts := topo.GetServer()
	defer topo.CloseServers()

	resilientSrvTopoServer = vtgate.NewResilientSrvTopoServer(ts, "ResilientSrvTopoServer")

	healthCheck = discovery.NewHealthCheck(*healthCheckConnTimeout, *healthCheckRetryDelay, *healthCheckTimeout)
	healthCheck.RegisterStats()

	tabletTypes := make([]topodatapb.TabletType, 0, 1)
	if len(*tabletTypesToWait) != 0 {
		for _, ttStr := range strings.Split(*tabletTypesToWait, ",") {
			tt, err := topoproto.ParseTabletType(ttStr)
			if err != nil {
				log.Errorf("unknown tablet type: %v", ttStr)
				continue
			}
			tabletTypes = append(tabletTypes, tt)
		}
	}
	l2vtg := l2vtgate.Init(healthCheck, ts, resilientSrvTopoServer, *cell, *retryCount, tabletTypes)

	servenv.OnRun(func() {
		addStatusParts(l2vtg)
	})
	servenv.RunDefault()
}

func init() {
	// Wait until flags are parsed, so we can check which topo server is in use.
	servenv.OnRun(func() {
		if etcdServer, ok := topo.GetServer().Impl.(*etcdtopo.Server); ok {
			vtctld.HandleExplorer("etcd", etcdtopo.NewExplorer(etcdServer))
		}
	})
}

func init() {
	servenv.ServiceMap["grpc-mysqlctl"] = true
	servenv.OnRun(func() {
		if servenv.GRPCCheckServiceMap("mysqlctl") {
			grpcmysqlctlserver.StartServer(servenv.GRPCServer, mysqld)
		}
	})
}

func init() {
	servenv.RegisterGRPCFlags()
	servenv.OnRun(func() {
		if servenv.GRPCCheckServiceMap("vtworker") {
			grpcvtworkerserver.StartServer(servenv.GRPCServer, wi)
		}
	})
}

func init() {
	// Wait until flags are parsed, so we can check which topo server is in use.
	servenv.OnRun(func() {
		if zkServer, ok := topo.GetServer().Impl.(*zktopo.Server); ok {
			HandleExplorer("zk", "/zk/", "zk.html", NewZkExplorer(zkServer.GetZConn()))
		}
	})
}

func init() {
	servenv.InitServiceMap("grpc", "mysqlctl")
	servenv.OnRun(func() {
		if servenv.GRPCCheckServiceMap("mysqlctl") {
			grpcmysqlctlserver.StartServer(servenv.GRPCServer, mysqld)
		}
	})
}

func main() {
	defer exit.Recover()

	flag.Parse()
	servenv.Init()

	if initFakeZK != nil {
		initFakeZK()
	}
	ts := topo.GetServer()
	defer topo.CloseServers()

	var schema *planbuilder.Schema
	if *schemaFile != "" {
		var err error
		if schema, err = planbuilder.LoadFile(*schemaFile); err != nil {
			log.Error(err)
			exit.Return(1)
		}
		log.Infof("v3 is enabled: loaded schema from file: %v", *schemaFile)
	} else {
		ctx := context.Background()
		schemaJSON, err := ts.GetVSchema(ctx)
		if err != nil {
			log.Warningf("Skipping v3 initialization: GetVSchema failed: %v", err)
			goto startServer
		}
		schema, err = planbuilder.NewSchema([]byte(schemaJSON))
		if err != nil {
			log.Warningf("Skipping v3 initialization: GetVSchema failed: %v", err)
			goto startServer
		}
		log.Infof("v3 is enabled: loaded schema from topo")
	}

startServer:
	resilientSrvTopoServer = vtgate.NewResilientSrvTopoServer(ts, "ResilientSrvTopoServer")

	healthCheck = discovery.NewHealthCheck(*connTimeoutTotal, *healthCheckRetryDelay, *healthCheckTimeout, "" /* statsSuffix */)

	tabletTypes := make([]topodatapb.TabletType, 0, 1)
	if len(*tabletTypesToWait) != 0 {
		for _, ttStr := range strings.Split(*tabletTypesToWait, ",") {
			tt, err := topoproto.ParseTabletType(ttStr)
			if err != nil {
				log.Errorf("unknown tablet type: %v", ttStr)
				continue
			}
			tabletTypes = append(tabletTypes, tt)
		}
	}
	vtg := vtgate.Init(healthCheck, ts, resilientSrvTopoServer, schema, *cell, *retryDelay, *retryCount, *connTimeoutTotal, *connTimeoutPerConn, *connLife, tabletTypes, *maxInFlight, *testGateway)

	servenv.OnRun(func() {
		addStatusParts(vtg)
	})
	servenv.RunDefault()
}

func init() {
	servenv.OnRun(func() {
		if !*enableQueries {
			return
		}

		addCommandGroup(queriesGroupName)

		// VtGate commands
		addCommand(queriesGroupName, command{
			"VtGateExecute",
			commandVtGateExecute,
			"-server <vtgate> [-bind_variables <JSON map>] [-connect_timeout <connect timeout>] [-keyspace <default keyspace>] [-tablet_type <tablet type>] [-json] <sql>",
			"Executes the given SQL query with the provided bound variables against the vtgate server."})
		addCommand(queriesGroupName, command{
			"VtGateExecuteShards",
			commandVtGateExecuteShards,
			"-server <vtgate> -keyspace <keyspace> -shards <shard0>,<shard1>,... [-bind_variables <JSON map>] [-connect_timeout <connect timeout>] [-tablet_type <tablet type>] [-json] <sql>",
			"Executes the given SQL query with the provided bound variables against the vtgate server. It is routed to the provided shards."})
		addCommand(queriesGroupName, command{
			"VtGateExecuteKeyspaceIds",
			commandVtGateExecuteKeyspaceIds,
			"-server <vtgate> -keyspace <keyspace> -keyspace_ids <ks1 in hex>,<k2 in hex>,... [-bind_variables <JSON map>] [-connect_timeout <connect timeout>] [-tablet_type <tablet type>] [-json] <sql>",
			"Executes the given SQL query with the provided bound variables against the vtgate server. It is routed to the shards that contain the provided keyspace ids."})
		addCommand(queriesGroupName, command{
			"VtGateSplitQuery",
			commandVtGateSplitQuery,
			"-server <vtgate> -keyspace <keyspace> [-split_column <split_column>] -split_count <split_count> [-bind_variables <JSON map>] [-connect_timeout <connect timeout>] <sql>",
			"Executes the SplitQuery computation for the given SQL query with the provided bound variables against the vtgate server (this is the base query for Map-Reduce workloads, and is provided here for debug / test purposes)."})

		// VtTablet commands
		addCommand(queriesGroupName, command{
			"VtTabletExecute",
			commandVtTabletExecute,
			"[-bind_variables <JSON map>] [-connect_timeout <connect timeout>] [-transaction_id <transaction_id>] [-tablet_type <tablet_type>] [-json] -keyspace <keyspace> -shard <shard> <tablet alias> <sql>",
			"Executes the given query on the given tablet."})
		addCommand(queriesGroupName, command{
			"VtTabletBegin",
			commandVtTabletBegin,
			"[-connect_timeout <connect timeout>] [-tablet_type <tablet_type>] -keyspace <keyspace> -shard <shard> <tablet alias>",
			"Starts a transaction on the provided server."})
		addCommand(queriesGroupName, command{
			"VtTabletCommit",
			commandVtTabletCommit,
			"[-connect_timeout <connect timeout>] [-tablet_type <tablet_type>] -keyspace <keyspace> -shard <shard> <tablet alias> <transaction_id>",
			"Commits a transaction on the provided server."})
		addCommand(queriesGroupName, command{
			"VtTabletRollback",
			commandVtTabletRollback,
			"[-connect_timeout <connect timeout>] [-tablet_type <tablet_type>] -keyspace <keyspace> -shard <shard> <tablet alias> <transaction_id>",
			"Rollbacks a transaction on the provided server."})
		addCommand(queriesGroupName, command{
			"VtTabletStreamHealth",
			commandVtTabletStreamHealth,
			"[-count <count, default 1>] [-connect_timeout <connect timeout>] <tablet alias>",
			"Executes the StreamHealth streaming query to a vttablet process. Will stop after getting <count> answers."})
	})
}

func init() {
	servenv.OnRun(func() {
		if agent.IsRunningHealthCheck() {
			servenv.AddStatusFuncs(template.FuncMap{
				"github_com_youtube_vitess_health_html_name": healthHTMLName,
			})
			servenv.AddStatusPart("Health", healthTemplate, func() interface{} {
				return &healthStatus{Records: agent.History.Records()}
			})
		}
		tabletserver.AddStatusPart()
	})
}

func init() {
	servenv.OnRun(func() {
		servenv.AddStatusPart("Topology Cache", topoTemplate, func() interface{} {
			return resilientSrvTopoServer.CacheStatus()
		})
		servenv.AddStatusPart("Stats", statsTemplate, func() interface{} {
			return nil
		})
		if onStatusRegistered != nil {
			onStatusRegistered()
		}
	})
}

func init() {
	servenv.OnRun(func() {
		http.Handle("/healthz", http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
			if _, err := agent.Healthy(); err != nil {
				http.Error(rw, fmt.Sprintf("500 internal server error: agent not healthy: %v", err), http.StatusInternalServerError)
				return
			}

			rw.Header().Set("Content-Length", fmt.Sprintf("%v", len(okMessage)))
			rw.WriteHeader(http.StatusOK)
			rw.Write(okMessage)
		}))
	})
}

// Init initializes VTGate server.
func Init(ctx context.Context, hc discovery.HealthCheck, topoServer topo.Server, serv topo.SrvTopoServer, cell string, retryCount int, tabletTypesToWait []topodatapb.TabletType) *VTGate {
	if rpcVTGate != nil {
		log.Fatalf("VTGate already initialized")
	}
	rpcVTGate = &VTGate{
		resolver:     NewResolver(hc, topoServer, serv, "VttabletCall", cell, retryCount, tabletTypesToWait),
		timings:      stats.NewMultiTimings("VtgateApi", []string{"Operation", "Keyspace", "DbType"}),
		rowsReturned: stats.NewMultiCounters("VtgateApiRowsReturned", []string{"Operation", "Keyspace", "DbType"}),

		logExecute:                  logutil.NewThrottledLogger("Execute", 5*time.Second),
		logExecuteShards:            logutil.NewThrottledLogger("ExecuteShards", 5*time.Second),
		logExecuteKeyspaceIds:       logutil.NewThrottledLogger("ExecuteKeyspaceIds", 5*time.Second),
		logExecuteKeyRanges:         logutil.NewThrottledLogger("ExecuteKeyRanges", 5*time.Second),
		logExecuteEntityIds:         logutil.NewThrottledLogger("ExecuteEntityIds", 5*time.Second),
		logExecuteBatchShards:       logutil.NewThrottledLogger("ExecuteBatchShards", 5*time.Second),
		logExecuteBatchKeyspaceIds:  logutil.NewThrottledLogger("ExecuteBatchKeyspaceIds", 5*time.Second),
		logStreamExecute:            logutil.NewThrottledLogger("StreamExecute", 5*time.Second),
		logStreamExecuteKeyspaceIds: logutil.NewThrottledLogger("StreamExecuteKeyspaceIds", 5*time.Second),
		logStreamExecuteKeyRanges:   logutil.NewThrottledLogger("StreamExecuteKeyRanges", 5*time.Second),
		logStreamExecuteShards:      logutil.NewThrottledLogger("StreamExecuteShards", 5*time.Second),
		logUpdateStream:             logutil.NewThrottledLogger("UpdateStream", 5*time.Second),
	}

	// vschemaCounters needs to be initialized before planner to
	// catch the initial load stats.
	vschemaCounters = stats.NewCounters("VtgateVSchemaCounts")

	// Resuse resolver's scatterConn.
	rpcVTGate.router = NewRouter(ctx, serv, cell, "VTGateRouter", rpcVTGate.resolver.scatterConn)
	normalErrors = stats.NewMultiCounters("VtgateApiErrorCounts", []string{"Operation", "Keyspace", "DbType"})
	infoErrors = stats.NewCounters("VtgateInfoErrorCounts")
	internalErrors = stats.NewCounters("VtgateInternalErrorCounts")

	qpsByOperation = stats.NewRates("QPSByOperation", stats.CounterForDimension(rpcVTGate.timings, "Operation"), 15, 1*time.Minute)
	qpsByKeyspace = stats.NewRates("QPSByKeyspace", stats.CounterForDimension(rpcVTGate.timings, "Keyspace"), 15, 1*time.Minute)
	qpsByDbType = stats.NewRates("QPSByDbType", stats.CounterForDimension(rpcVTGate.timings, "DbType"), 15, 1*time.Minute)

	errorsByOperation = stats.NewRates("ErrorsByOperation", stats.CounterForDimension(normalErrors, "Operation"), 15, 1*time.Minute)
	errorsByKeyspace = stats.NewRates("ErrorsByKeyspace", stats.CounterForDimension(normalErrors, "Keyspace"), 15, 1*time.Minute)
	errorsByDbType = stats.NewRates("ErrorsByDbType", stats.CounterForDimension(normalErrors, "DbType"), 15, 1*time.Minute)

	servenv.OnRun(func() {
		for _, f := range RegisterVTGates {
			f(rpcVTGate)
		}
	})
	vtgateOnce.Do(rpcVTGate.registerDebugHealthHandler)
	return rpcVTGate
}

func main() {
	defer exit.Recover()

	flag.Parse()
	servenv.Init()

	// The implementation chain.
	servenv.OnRun(func() {
		s := services.CreateServices()
		for _, f := range vtgate.RegisterVTGates {
			f(s)
		}
	})

	servenv.RunDefault()
}