// TestParseJoinUsingAddrs verifies that JoinList is parsed
// correctly.
func TestParseJoinUsingAddrs(t *testing.T) {
	defer leaktest.AfterTest(t)()
	ctx := MakeContext()
	ctx.JoinList = []string{"localhost:12345,,localhost:23456", "localhost:34567"}
	ctx.Stores = StoreSpecList{Specs: []StoreSpec{{InMemory: true, SizeInBytes: minimumStoreSize * 100}}}
	stopper := stop.NewStopper()
	defer stopper.Stop()
	if err := ctx.InitStores(stopper); err != nil {
		t.Fatalf("Failed to initialize stores: %s", err)
	}
	if err := ctx.InitNode(); err != nil {
		t.Fatalf("Failed to initialize node: %s", err)
	}
	r1, err := resolver.NewResolver("localhost:12345")
	if err != nil {
		t.Fatal(err)
	}
	r2, err := resolver.NewResolver("localhost:23456")
	if err != nil {
		t.Fatal(err)
	}
	r3, err := resolver.NewResolver("localhost:34567")
	if err != nil {
		t.Fatal(err)
	}
	expected := []resolver.Resolver{r1, r2, r3}
	if !reflect.DeepEqual(ctx.GossipBootstrapResolvers, expected) {
		t.Fatalf("Unexpected bootstrap addresses: %v, expected: %v", ctx.GossipBootstrapResolvers, expected)
	}
}

// TestParseGossipBootstrapAddrs verifies that GossipBootstrap is
// parsed correctly.
func TestParseGossipBootstrapAddrs(t *testing.T) {
	defer leaktest.AfterTest(t)
	ctx := NewContext()
	ctx.GossipBootstrap = "localhost:12345,,localhost:23456"
	ctx.Stores = "mem=1"
	stopper := stop.NewStopper()
	defer stopper.Stop()
	if err := ctx.InitStores(stopper); err != nil {
		t.Fatalf("Failed to initialize stores: %s", err)
	}
	if err := ctx.InitNode(); err != nil {
		t.Fatalf("Failed to initialize node: %s", err)
	}
	r1, err := resolver.NewResolver(&ctx.Context, "tcp=localhost:12345")
	if err != nil {
		t.Fatal(err)
	}
	r2, err := resolver.NewResolver(&ctx.Context, "tcp=localhost:23456")
	if err != nil {
		t.Fatal(err)
	}
	expected := []resolver.Resolver{r1, r2}
	if !reflect.DeepEqual(ctx.GossipBootstrapResolvers, expected) {
		t.Fatalf("Unexpected bootstrap addresses: %v, expected: %v", ctx.GossipBootstrapResolvers, expected)
	}
}

// Start runs the RPC and HTTP servers, starts the gossip instance (if
// selfBootstrap is true, uses the rpc server's address as the gossip
// bootstrap), and starts the node using the supplied engines slice.
func (s *Server) Start(selfBootstrap bool) error {
	if err := s.rpc.Listen(); err != nil {
		return util.Errorf("could not listen on %s: %s", s.ctx.Addr, err)
	}

	// Handle self-bootstrapping case for a single node.
	if selfBootstrap {
		selfResolver, err := resolver.NewResolver(&s.ctx.Context, s.rpc.Addr().String())
		if err != nil {
			return err
		}
		s.gossip.SetResolvers([]resolver.Resolver{selfResolver})
	}
	s.gossip.Start(s.rpc, s.stopper)

	if err := s.node.start(s.rpc, s.ctx.Engines, s.ctx.NodeAttributes, s.stopper); err != nil {
		return err
	}

	// Begin recording runtime statistics.
	runtime := status.NewRuntimeStatRecorder(s.node.Descriptor.NodeID, s.clock)
	s.tsDB.PollSource(runtime, s.ctx.MetricsFrequency, ts.Resolution10s, s.stopper)

	// Begin recording time series data collected by the status monitor.
	s.recorder = status.NewNodeStatusRecorder(s.node.status, s.clock)
	s.tsDB.PollSource(s.recorder, s.ctx.MetricsFrequency, ts.Resolution10s, s.stopper)

	// Begin recording status summaries.
	s.startWriteSummaries()

	log.Infof("starting %s server at %s", s.ctx.HTTPRequestScheme(), s.rpc.Addr())
	s.initHTTP()
	s.rpc.Serve(s)
	return nil
}

// Start runs the RPC and HTTP servers, starts the gossip instance (if
// selfBootstrap is true, uses the rpc server's address as the gossip
// bootstrap), and starts the node using the supplied engines slice.
func (s *Server) Start(selfBootstrap bool) error {
	tlsConfig, err := s.ctx.GetServerTLSConfig()
	if err != nil {
		return err
	}

	unresolvedAddr := util.MakeUnresolvedAddr("tcp", s.ctx.Addr)
	ln, err := util.ListenAndServe(s.stopper, s, unresolvedAddr, tlsConfig)
	if err != nil {
		return err
	}

	s.listener = ln

	addr := ln.Addr()
	addrStr := addr.String()

	// Handle self-bootstrapping case for a single node.
	if selfBootstrap {
		selfResolver, err := resolver.NewResolver(&s.ctx.Context, addrStr)
		if err != nil {
			return err
		}
		s.gossip.SetResolvers([]resolver.Resolver{selfResolver})
	}
	s.gossip.Start(s.rpc, addr, s.stopper)

	if err := s.node.start(s.rpc, addr, s.ctx.Engines, s.ctx.NodeAttributes); err != nil {
		return err
	}

	// Begin recording runtime statistics.
	runtime := status.NewRuntimeStatRecorder(s.node.Descriptor.NodeID, s.clock)
	s.tsDB.PollSource(runtime, s.ctx.MetricsFrequency, ts.Resolution10s, s.stopper)

	// Begin recording time series data collected by the status monitor.
	s.recorder = status.NewNodeStatusRecorder(s.node.status, s.clock)
	s.tsDB.PollSource(s.recorder, s.ctx.MetricsFrequency, ts.Resolution10s, s.stopper)

	// Begin recording status summaries.
	s.startWriteSummaries()

	s.sqlServer.SetNodeID(s.node.Descriptor.NodeID)

	s.status = newStatusServer(s.db, s.gossip, s.metaRegistry, s.ctx)

	log.Infof("starting %s server at %s", s.ctx.HTTPRequestScheme(), addr)
	s.initHTTP()

	// TODO(tamird): pick a port here
	host, _, err := net.SplitHostPort(addrStr)
	if err != nil {
		return err
	}

	return s.pgServer.Start(util.MakeUnresolvedAddr("tcp", net.JoinHostPort(host, "0")))
}

func TestGossipGetNextBootstrapAddress(t *testing.T) {
	defer leaktest.AfterTest(t)()
	defer resolver.SetLookupTimeout(time.Minute)()

	// Set up an http server for testing the http load balancer.
	i := 0
	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		i++
		fmt.Fprintf(w, `{"address": {"network": "tcp", "address": "10.10.0.%d:12345"}}`, i)
	})
	s := httptest.NewServer(handler)
	defer s.Close()

	resolverSpecs := []string{
		"127.0.0.1:9000",
		"tcp=127.0.0.1:9001",
		"unix=/tmp/unix-socket12345",
		fmt.Sprintf("http-lb=%s", s.Listener.Addr()),
		"foo=127.0.0.1:9003", // error should not resolve.
		"http-lb=",           // error should not resolve.
		"localhost:9004",
	}

	resolvers := []resolver.Resolver{}
	for _, rs := range resolverSpecs {
		resolver, err := resolver.NewResolver(&base.Context{Insecure: true}, rs)
		if err == nil {
			resolvers = append(resolvers, resolver)
		}
	}
	if len(resolvers) != 5 {
		t.Errorf("expected 5 resolvers; got %d", len(resolvers))
	}
	g := New(nil, resolvers, nil)

	// Using specified resolvers, fetch bootstrap addresses 10 times
	// and verify the results match expected addresses.
	expAddresses := []string{
		"127.0.0.1:9000",
		"127.0.0.1:9001",
		"/tmp/unix-socket12345",
		"10.10.0.1:12345",
		"localhost:9004",
		"10.10.0.2:12345",
		"10.10.0.3:12345",
		"10.10.0.4:12345",
		"10.10.0.5:12345",
		"10.10.0.6:12345",
	}
	for i := 0; i < len(expAddresses); i++ {
		if addr := g.getNextBootstrapAddress(); addr == nil {
			t.Errorf("%d: unexpected nil addr when expecting %s", i, expAddresses[i])
		} else if addrStr := addr.String(); addrStr != expAddresses[i] {
			t.Errorf("%d: expected addr %s; got %s", i, expAddresses[i], addrStr)
		}
	}
}

// Start runs the RPC and HTTP servers, starts the gossip instance (if
// selfBootstrap is true, uses the rpc server's address as the gossip
// bootstrap), and starts the node using the supplied engines slice.
func (s *Server) Start(selfBootstrap bool) error {
	tlsConfig, err := s.ctx.GetServerTLSConfig()
	if err != nil {
		return err
	}

	unresolvedAddr := util.MakeUnresolvedAddr("tcp", s.ctx.Addr)
	ln, err := util.ListenAndServe(s.stopper, s, unresolvedAddr, tlsConfig)
	if err != nil {
		return err
	}

	s.listener = ln

	addr := ln.Addr()
	addrStr := addr.String()
	s.rpcContext.SetLocalServer(s.rpc, addrStr)

	// Handle self-bootstrapping case for a single node.
	if selfBootstrap {
		selfResolver, err := resolver.NewResolver(&s.ctx.Context, addrStr)
		if err != nil {
			return err
		}
		s.gossip.SetResolvers([]resolver.Resolver{selfResolver})
	}
	s.gossip.Start(s.rpc, addr)

	if err := s.node.start(s.rpc, addr, s.ctx.Engines, s.ctx.NodeAttributes); err != nil {
		return err
	}

	// Begin recording runtime statistics.
	runtime := status.NewRuntimeStatRecorder(s.node.Descriptor.NodeID, s.clock)
	s.tsDB.PollSource(runtime, s.ctx.MetricsFrequency, ts.Resolution10s, s.stopper)

	// Begin recording time series data collected by the status monitor.
	s.recorder = status.NewNodeStatusRecorder(s.node.status, s.clock)
	s.tsDB.PollSource(s.recorder, s.ctx.MetricsFrequency, ts.Resolution10s, s.stopper)

	// Begin recording status summaries.
	s.startWriteSummaries()

	s.sqlServer.SetNodeID(s.node.Descriptor.NodeID)
	// Create and start the schema change manager only after a NodeID
	// has been assigned.
	s.schemaChangeManager = sql.NewSchemaChangeManager(*s.db, s.gossip, s.leaseMgr)
	s.schemaChangeManager.Start(s.stopper)

	s.status = newStatusServer(s.db, s.gossip, s.metaRegistry, s.ctx)

	log.Infof("starting %s server at %s", s.ctx.HTTPRequestScheme(), addr)
	s.initHTTP()

	return s.pgServer.Start(util.MakeUnresolvedAddr("tcp", s.ctx.PGAddr))
}

func TestGossipGetNextBootstrapAddress(t *testing.T) {
	defer leaktest.AfterTest(t)
	resolverSpecs := []string{
		"127.0.0.1:9000",
		"tcp=127.0.0.1:9001",
		"unix=/tmp/unix-socket12345",
		"lb=127.0.0.1:9002",
		"foo=127.0.0.1:9003", // error should not resolve.
		"lb=",                // error should not resolve.
		"localhost:9004",
		"lb=127.0.0.1:9005",
	}

	resolvers := []resolver.Resolver{}
	for _, rs := range resolverSpecs {
		resolver, err := resolver.NewResolver(&base.Context{}, rs)
		if err == nil {
			resolvers = append(resolvers, resolver)
		}
	}
	if len(resolvers) != 6 {
		t.Errorf("expected 6 resolvers; got %d", len(resolvers))
	}
	g := New(nil, resolvers)

	// Using specified resolvers, fetch bootstrap addresses 10 times
	// and verify the results match expected addresses.
	expAddresses := []string{
		"127.0.0.1:9000",
		"127.0.0.1:9001",
		"/tmp/unix-socket12345",
		"127.0.0.1:9002",
		"localhost:9004",
		"127.0.0.1:9005",
		"127.0.0.1:9002",
		"127.0.0.1:9005",
		"127.0.0.1:9002",
		"127.0.0.1:9005",
	}
	for i := 0; i < len(expAddresses); i++ {
		log.Infof("getting next address")
		addr := g.getNextBootstrapAddress()
		if addr == nil {
			t.Errorf("%d: unexpected nil addr when expecting %s", i, expAddresses[i])
		} else if addr.String() != expAddresses[i] {
			t.Errorf("%d: expected addr %s; got %s", i, expAddresses[i], addr.String())
		}
	}
}

// TestClientRegisterInitNodeID verifies two client's gossip request with NodeID 0.
func TestClientRegisterWithInitNodeID(t *testing.T) {
	defer leaktest.AfterTest(t)()
	stopper := stop.NewStopper()
	defer stopper.Stop()

	// Create three gossip nodes, and connect to the first with NodeID 0.
	var g []*Gossip
	var gossipAddr string
	for i := 0; i < 3; i++ {
		clock := hlc.NewClock(hlc.UnixNano)
		RPCContext := rpc.NewContext(&base.Context{Insecure: true}, clock, stopper)

		addr := util.CreateTestAddr("tcp")
		server := grpc.NewServer()
		TLSConfig, err := RPCContext.GetServerTLSConfig()
		if err != nil {
			t.Fatal(err)
		}
		ln, err := util.ListenAndServe(stopper, server, addr, TLSConfig)
		if err != nil {
			t.Fatal(err)
		}

		// Connect to the first gossip node.
		if gossipAddr == "" {
			gossipAddr = ln.Addr().String()
		}

		var resolvers []resolver.Resolver
		resolver, _ := resolver.NewResolver(&RPCContext.Context, gossipAddr)
		resolvers = append(resolvers, resolver)
		gnode := New(RPCContext, resolvers, stopper)
		// node ID must be non-zero
		gnode.SetNodeID(roachpb.NodeID(i + 1))
		g = append(g, gnode)
		gnode.Start(server, ln.Addr())
	}

	util.SucceedsSoon(t, func() error {
		// The first gossip node should have two gossip client address
		// in nodeMap if these three gossip nodes registered success.
		g[0].mu.Lock()
		defer g[0].mu.Unlock()
		if a, e := len(g[0].nodeMap), 2; a != e {
			return util.Errorf("expected %s to contain %d nodes, got %d", g[0].nodeMap, e, a)
		}
		return nil
	})
}

// parseGossipBootstrapResolvers parses a comma-separated list of
// gossip bootstrap resolvers.
func (ctx *Context) parseGossipBootstrapResolvers() ([]resolver.Resolver, error) {
	var bootstrapResolvers []resolver.Resolver
	addresses := strings.Split(ctx.JoinUsing, ",")
	for _, address := range addresses {
		if len(address) == 0 {
			continue
		}
		resolver, err := resolver.NewResolver(&ctx.Context, address)
		if err != nil {
			return nil, err
		}
		bootstrapResolvers = append(bootstrapResolvers, resolver)
	}

	return bootstrapResolvers, nil
}

// TestClientRegisterInitNodeID verifies two client's gossip request with NodeID 0.
func TestClientRegisterWithInitNodeID(t *testing.T) {
	defer leaktest.AfterTest(t)
	stopper := stop.NewStopper()
	defer stopper.Stop()

	// Create three gossip nodes, and connect to the first with NodeID 0.
	var g []*Gossip
	var gossipAddr string
	for i := 0; i < 3; i++ {
		clock := hlc.NewClock(hlc.UnixNano)
		RPCContext := rpc.NewContext(&base.Context{Insecure: true}, clock, stopper)

		addr := util.CreateTestAddr("tcp")
		server := rpc.NewServer(RPCContext)
		TLSConfig, err := RPCContext.GetServerTLSConfig()
		if err != nil {
			t.Fatal(err)
		}
		ln, err := util.ListenAndServe(stopper, server, addr, TLSConfig)
		if err != nil {
			t.Fatal(err)
		}

		// Connect to the first gossip node.
		if gossipAddr == "" {
			gossipAddr = ln.Addr().String()
		}

		var resolvers []resolver.Resolver
		resolver, _ := resolver.NewResolver(&RPCContext.Context, gossipAddr)
		resolvers = append(resolvers, resolver)
		gnode := New(RPCContext, resolvers, stopper)
		g = append(g, gnode)
		gnode.Start(server, ln.Addr())
	}

	util.SucceedsWithin(t, 5*time.Second, func() error {
		// The first gossip node should have two gossip client address
		// in lAddrMap if these three gossip nodes registered success.
		g[0].mu.Lock()
		defer g[0].mu.Unlock()
		if len(g[0].lAddrMap) == 2 {
			return nil
		}
		return util.Errorf("gossip client register fail.")
	})
}

// TestClientRegisterInitNodeID verifies two client's gossip request with NodeID 0.
func TestClientRegisterWithInitNodeID(t *testing.T) {
	defer leaktest.AfterTest(t)()
	stopper := stop.NewStopper()
	defer stopper.Stop()

	// Create three gossip nodes, and connect to the first with NodeID 0.
	var g []*Gossip
	var gossipAddr string
	for i := 0; i < 3; i++ {
		RPCContext := rpc.NewContext(&base.Context{Insecure: true}, nil, stopper)

		server := rpc.NewServer(RPCContext)
		ln, err := netutil.ListenAndServeGRPC(stopper, server, util.TestAddr)
		if err != nil {
			t.Fatal(err)
		}

		// Connect to the first gossip node.
		if gossipAddr == "" {
			gossipAddr = ln.Addr().String()
		}

		var resolvers []resolver.Resolver
		resolver, err := resolver.NewResolver(gossipAddr)
		if err != nil {
			t.Fatal(err)
		}
		resolvers = append(resolvers, resolver)
		gnode := New(context.TODO(), RPCContext, server, resolvers, stopper, metric.NewRegistry())
		// node ID must be non-zero
		gnode.SetNodeID(roachpb.NodeID(i + 1))
		g = append(g, gnode)
		gnode.Start(ln.Addr())
	}

	util.SucceedsSoon(t, func() error {
		// The first gossip node should have two gossip client address
		// in nodeMap if these three gossip nodes registered success.
		g[0].mu.Lock()
		defer g[0].mu.Unlock()
		if a, e := len(g[0].mu.nodeMap), 2; a != e {
			return errors.Errorf("expected %s to contain %d nodes, got %d", g[0].mu.nodeMap, e, a)
		}
		return nil
	})
}

// parseGossipBootstrapResolvers parses list of gossip bootstrap resolvers.
func (ctx *Context) parseGossipBootstrapResolvers() ([]resolver.Resolver, error) {
	var bootstrapResolvers []resolver.Resolver
	for _, commaSeparatedAddresses := range ctx.JoinList {
		addresses := strings.Split(commaSeparatedAddresses, ",")
		for _, address := range addresses {
			if len(address) == 0 {
				continue
			}
			resolver, err := resolver.NewResolver(address)
			if err != nil {
				return nil, err
			}
			bootstrapResolvers = append(bootstrapResolvers, resolver)
		}
	}

	return bootstrapResolvers, nil
}

// parseGossipBootstrapResolvers parses a comma-separated list of
// gossip bootstrap resolvers.
func (ctx *Context) parseGossipBootstrapResolvers() ([]resolver.Resolver, error) {
	var bootstrapResolvers []resolver.Resolver
	addresses := strings.Split(ctx.GossipBootstrap, ",")
	for _, address := range addresses {
		if len(address) == 0 {
			continue
		}
		if strings.HasPrefix(address, SelfGossipAddr) {
			address = ctx.Addr
		}
		resolver, err := resolver.NewResolver(&ctx.Context, address)
		if err != nil {
			return nil, err
		}
		bootstrapResolvers = append(bootstrapResolvers, resolver)
	}

	return bootstrapResolvers, nil
}

func TestGossipGetNextBootstrapAddress(t *testing.T) {
	defer leaktest.AfterTest(t)()
	stopper := stop.NewStopper()
	defer stopper.Stop()

	resolverSpecs := []string{
		"127.0.0.1:9000",
		"127.0.0.1:9001",
		"localhost:9004",
	}

	resolvers := []resolver.Resolver{}
	for _, rs := range resolverSpecs {
		resolver, err := resolver.NewResolver(&base.Context{Insecure: true}, rs)
		if err == nil {
			resolvers = append(resolvers, resolver)
		}
	}
	if len(resolvers) != 3 {
		t.Errorf("expected 3 resolvers; got %d", len(resolvers))
	}
	server := rpc.NewServer(rpc.NewContext(&base.Context{Insecure: true}, nil, stopper))
	g := New(nil, server, resolvers, nil, metric.NewRegistry())

	// Using specified resolvers, fetch bootstrap addresses 3 times
	// and verify the results match expected addresses.
	expAddresses := []string{
		"127.0.0.1:9000",
		"127.0.0.1:9001",
		"localhost:9004",
	}
	for i := 0; i < len(expAddresses); i++ {
		if addr := g.getNextBootstrapAddress(); addr == nil {
			t.Errorf("%d: unexpected nil addr when expecting %s", i, expAddresses[i])
		} else if addrStr := addr.String(); addrStr != expAddresses[i] {
			t.Errorf("%d: expected addr %s; got %s", i, expAddresses[i], addrStr)
		}
	}
}

// parseGossipBootstrapResolvers parses a comma-separated list of
// gossip bootstrap resolvers.
func (ctx *Context) parseGossipBootstrapResolvers() ([]resolver.Resolver, error) {
	var bootstrapResolvers []resolver.Resolver
	addresses := strings.Split(ctx.GossipBootstrap, ",")
	for _, address := range addresses {
		if len(address) == 0 {
			continue
		}
		// Special case self= to pick a nice address that resolves
		// uniquely for use in Gossip. This avoids having to specify
		// the port for single-node clusters twice (once in --addr,
		// once in --gossip).
		if strings.HasPrefix(address, "self=") {
			address = ctx.Addr
		}
		resolver, err := resolver.NewResolver(&ctx.Context, address)
		if err != nil {
			return nil, err
		}
		bootstrapResolvers = append(bootstrapResolvers, resolver)
	}

	return bootstrapResolvers, nil
}