diff --git a/ctl/server.go b/ctl/server.go index 14a262470..8aa0edeb9 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -28,6 +28,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVarP(&srv.Config.Bind, "bind", "b", srv.Config.Bind, "Default URI on which pilosa should listen.") flags.StringVar(&srv.Config.BindGRPC, "bind-grpc", srv.Config.BindGRPC, "URI on which pilosa should listen for gRPC requests.") flags.StringVar(&srv.Config.Advertise, "advertise", srv.Config.Advertise, "Address to advertise externally.") + flags.StringVar(&srv.Config.AdvertiseGRPC, "advertise-grpc", srv.Config.AdvertiseGRPC, "Address to advertise externally for gRPC.") flags.IntVarP(&srv.Config.MaxWritesPerRequest, "max-writes-per-request", "", srv.Config.MaxWritesPerRequest, "Number of write commands per request.") flags.StringVar(&srv.Config.LogPath, "log-path", srv.Config.LogPath, "Log path") flags.BoolVar(&srv.Config.Verbose, "verbose", srv.Config.Verbose, "Enable verbose logging") diff --git a/server/config.go b/server/config.go index 2e59ed314..6d0181756 100644 --- a/server/config.go +++ b/server/config.go @@ -29,6 +29,11 @@ import ( "github.com/pkg/errors" ) +const ( + defaultBindPort = "10101" + defaultBindGRPCPort = "20101" +) + // TLSConfig contains TLS configuration type TLSConfig struct { // CertificatePath contains the path to the certificate (.crt or .pem file) @@ -60,6 +65,11 @@ type Config struct { // route to an interface that Bind is listening on. Advertise string `toml:"advertise"` + // AdvertiseGRPC is the address advertised by the server to other nodes + // in the cluster. It should be reachable by all other nodes and should + // route to an interface that BindGRPC is listening on. + AdvertiseGRPC string `toml:"advertise-grpc"` + // MaxWritesPerRequest limits the number of mutating commands that can be in // a single request to the server. This includes Set, Clear, // SetRowAttrs & SetColumnAttrs. @@ -162,8 +172,8 @@ type Config struct { func NewConfig() *Config { c := &Config{ DataDir: "~/.pilosa", - Bind: ":10101", - BindGRPC: ":20101", + Bind: ":" + defaultBindPort, + BindGRPC: ":" + defaultBindGRPCPort, MaxWritesPerRequest: 5000, // We default these Max File/Map counts very high. This is basically a @@ -223,25 +233,32 @@ func NewConfig() *Config { // indicate it's left unspecified. func (cfg *Config) validateAddrs(ctx context.Context) error { // Validate the advertise address. - advScheme, advHost, advPort, err := validateAdvertiseAddr(ctx, cfg.Advertise, cfg.Bind) + advScheme, advHost, advPort, err := validateAdvertiseAddr(ctx, cfg.Advertise, cfg.Bind, defaultBindPort) if err != nil { return errors.Wrapf(err, "validating advertise address") } cfg.Advertise = schemeHostPortString(advScheme, advHost, advPort) // Validate the listen address. - listenScheme, listenHost, listenPort, err := validateListenAddr(ctx, cfg.Bind) + listenScheme, listenHost, listenPort, err := validateListenAddr(ctx, cfg.Bind, defaultBindPort) if err != nil { return errors.Wrap(err, "validating listen address") } cfg.Bind = schemeHostPortString(listenScheme, listenHost, listenPort) + // Validate the gRPC advertise address. + _, grpcAdvHost, grpcAdvPort, err := validateAdvertiseAddr(ctx, cfg.AdvertiseGRPC, cfg.BindGRPC, defaultBindGRPCPort) + if err != nil { + return errors.Wrapf(err, "validating grpc advertise address") + } + cfg.AdvertiseGRPC = schemeHostPortString("grpc", grpcAdvHost, grpcAdvPort) + // Validate the gRPC listen address. - grpcListenScheme, grpcListenHost, grpcListenPort, err := validateListenAddr(ctx, cfg.BindGRPC) + _, grpcListenHost, grpcListenPort, err := validateListenAddr(ctx, cfg.BindGRPC, defaultBindGRPCPort) if err != nil { return errors.Wrap(err, "validating grpc listen address") } - cfg.BindGRPC = schemeHostPortString(grpcListenScheme, grpcListenHost, grpcListenPort) + cfg.BindGRPC = schemeHostPortString("grpc", grpcListenHost, grpcListenPort) return nil } @@ -251,8 +268,8 @@ func (cfg *Config) validateAddrs(ctx context.Context) error { // the configured listen address if any, otherwise it makes a best // guess at the outbound IP address. // Returns scheme, host, port as strings. -func validateAdvertiseAddr(ctx context.Context, advAddr, listenAddr string) (string, string, string, error) { - listenScheme, listenHost, listenPort, err := splitAddr(listenAddr) +func validateAdvertiseAddr(ctx context.Context, advAddr, listenAddr, defaultPort string) (string, string, string, error) { + listenScheme, listenHost, listenPort, err := splitAddr(listenAddr, defaultPort) if err != nil { return "", "", "", errors.Wrap(err, "getting listen address") } @@ -318,8 +335,8 @@ func outboundIP() net.IP { // the default (localhost) should be used. Rresolves host names to IP // addresses. // Returns scheme, host, port as strings. -func validateListenAddr(ctx context.Context, addr string) (string, string, string, error) { - scheme, host, port, err := splitAddr(addr) +func validateListenAddr(ctx context.Context, addr, defaultPort string) (string, string, string, error) { + scheme, host, port, err := splitAddr(addr, defaultPort) if err != nil { return "", "", "", errors.Wrap(err, "getting listen address") } @@ -348,7 +365,7 @@ func schemeHostPortString(scheme, host, port string) string { } // splitAddr returns scheme, host, port as strings. -func splitAddr(addr string) (string, string, string, error) { +func splitAddr(addr string, defaultPort string) (string, string, string, error) { scheme, hostPort := splitScheme(addr) host, port := "", "" if hostPort != "" { @@ -362,7 +379,7 @@ func splitAddr(addr string) (string, string, string, error) { // results in a port of 0, which causes Pilosa to listen on // a random port. if port == "" { - port = "10101" + port = defaultPort } return scheme, host, port, nil } diff --git a/server/config_internal_test.go b/server/config_internal_test.go index ca7548895..09d423e98 100644 --- a/server/config_internal_test.go +++ b/server/config_internal_test.go @@ -26,7 +26,6 @@ import ( type addrs struct{ bind, advertise string } func TestConfig_validateAddrs(t *testing.T) { - // Prepare some reference strings that will be checked in the // test below. outboundAddr := outboundIP().String() @@ -104,8 +103,9 @@ func TestConfig_validateAddrs(t *testing.T) { {"", addrs{"0.0.0.0:1234", ""}, addrs{"0.0.0.0:1234", outboundAddr + ":1234"}}, - // Expected errors. + // Expected errors. + // // Missing port number. {"missing port in address", addrs{"localhost", ""}, @@ -139,11 +139,10 @@ func TestConfig_validateAddrs(t *testing.T) { } else if err == nil && test.expErr != "" { t.Fatalf("expected error string to contain %s, but got no error", test.expErr) } else if err != nil && test.expErr != "" { - if strings.Contains(err.Error(), test.expErr) { - return - } else { + if !strings.Contains(err.Error(), test.expErr) { t.Fatalf("expected error string to contain %s, but got %s", test.expErr, err.Error()) } + return } if c.Bind != test.exp.bind { @@ -154,3 +153,132 @@ func TestConfig_validateAddrs(t *testing.T) { }) } } + +func TestConfig_validateAddrsGRPC(t *testing.T) { + // Prepare some reference strings that will be checked in the + // test below. + outboundAddr := outboundIP().String() + hostname, err := os.Hostname() + if err != nil { + t.Fatal(err) + } + hostAddr, err := lookupAddr(context.Background(), net.DefaultResolver, hostname) + if err != nil { + t.Fatal(err) + } + if strings.Contains(hostAddr, ":") { + hostAddr = "[" + hostAddr + "]" + } + + tests := []struct { + expErr string + in addrs + exp addrs + }{ + // Default values; addresses set empty. + {"", + addrs{"", ""}, + addrs{"grpc://:20101", "grpc://:20101"}}, + {"", + addrs{":", ""}, + addrs{"grpc://:20101", "grpc://:20101"}}, + {"", + addrs{"", ":"}, + addrs{"grpc://:20101", "grpc://:20101"}}, + {"", + addrs{":", ":"}, + addrs{"grpc://:20101", "grpc://:20101"}}, + // Listener :port. + {"", + addrs{":1234", ""}, + addrs{"grpc://:1234", "grpc://:1234"}}, + // Listener with host:port. + {"", + addrs{hostAddr + ":20101", ""}, + addrs{"grpc://" + hostAddr + ":20101", "grpc://" + hostAddr + ":20101"}}, + // Listener with host:. + {"", + addrs{hostAddr + ":", ""}, + addrs{"grpc://" + hostAddr + ":20101", "grpc://" + hostAddr + ":20101"}}, + // Listener with scheme:. + {"", + addrs{"http://" + hostAddr + ":", ""}, + addrs{"grpc://" + hostAddr + ":20101", "grpc://" + hostAddr + ":20101"}}, + // Listener with localhost:port. + {"", + addrs{"localhost:1234", ""}, + addrs{"grpc://localhost:1234", "grpc://localhost:1234"}}, + // Listener with localhost:. + {"", + addrs{"localhost:", ""}, + addrs{"grpc://localhost:20101", "grpc://localhost:20101"}}, + // Listener and advertise addresses. + {"", + addrs{hostAddr + ":1234", hostAddr + ":"}, + addrs{"grpc://" + hostAddr + ":1234", "grpc://" + hostAddr + ":1234"}}, + // Explicit port number in advertise addr. + {"", + addrs{hostAddr + ":1234", hostAddr + ":7890"}, + addrs{"grpc://" + hostAddr + ":1234", "grpc://" + hostAddr + ":7890"}}, + // Use a non-numeric port number. + {"", + addrs{":postgresql", ""}, + addrs{"grpc://:5432", "grpc://:5432"}}, + // Advertise port 0 means reuse listen port. + {"", + addrs{":1234", ":0"}, + addrs{"grpc://:1234", "grpc://:1234"}}, + // Listen on all interfaces. Determine advertise address. + {"", + addrs{"0.0.0.0:1234", ""}, + addrs{"grpc://0.0.0.0:1234", "grpc://" + outboundAddr + ":1234"}}, + + // Expected errors. + // + // Missing port number. + {"missing port in address", + addrs{"localhost", ""}, + addrs{}}, + {"missing port in address", + addrs{":1234", "localhost"}, + addrs{}}, + // Invalid port number. + {"invalid port", + addrs{"localhost:-1234", ""}, + addrs{}}, + {"validating grpc advertise address", + addrs{"localhost:foo", ""}, + addrs{}}, + {"no such host", + addrs{"333.333.333.333:1234", ""}, + addrs{}}, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + c := NewConfig() + + c.BindGRPC = test.in.bind + c.AdvertiseGRPC = test.in.advertise + + err := c.validateAddrs(context.Background()) + + if err != nil && test.expErr == "" { + t.Fatal(err) + } else if err == nil && test.expErr != "" { + t.Fatalf("expected error string to contain %s, but got no error", test.expErr) + } else if err != nil && test.expErr != "" { + if !strings.Contains(err.Error(), test.expErr) { + t.Fatalf("expected error string to contain %s, but got %s", test.expErr, err.Error()) + } + return + } + + if c.BindGRPC != test.exp.bind { + t.Fatalf("bind address: expected %s, but got %s", test.exp.bind, c.BindGRPC) + } else if c.AdvertiseGRPC != test.exp.advertise { + t.Fatalf("advertise address: expected %s, but got %s", test.exp.advertise, c.AdvertiseGRPC) + } + }) + } +} diff --git a/server/server.go b/server/server.go index f995b7ac2..8ef7b92fd 100644 --- a/server/server.go +++ b/server/server.go @@ -265,20 +265,6 @@ func (m *Command) SetupServer() error { return errors.Wrap(err, "creating grpc listener") } - // If grpc port is 0, get auto-allocated port from listener - if grpcURI.Port == 0 { - grpcURI.SetPort(uint16(m.grpcLn.Addr().(*net.TCPAddr).Port)) - } - - if grpcURI.Scheme == "http" { - grpcURI.Scheme = "grpc" - } - - // discover the address if not specified - if grpcURI.Host == "0.0.0.0" { - grpcURI.Host = outboundIP().String() - } - // Setup TLS if uri.Scheme == "https" { m.tlsConfig, err = GetTLSConfig(&m.Config.TLS, m.logger.Logger()) @@ -321,6 +307,15 @@ func (m *Command) SetupServer() error { advertiseURI.SetPort(uri.Port) } + // Get grpc advertise address as uri. + advertiseGRPCURI, err := pilosa.NewURIFromAddress(m.Config.AdvertiseGRPC) + if err != nil { + return errors.Wrap(err, "processing grpc advertise address") + } + if advertiseGRPCURI.Port == 0 { + advertiseGRPCURI.SetPort(grpcURI.Port) + } + // Primary store configuration is handled automatically now. if m.Config.Translation.PrimaryURL != "" { m.logger.Printf("DEPRECATED: The primary-url configuration option is no longer used.") @@ -349,7 +344,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerGCNotifier(gcnotify.NewActiveGCNotifier()), pilosa.OptServerStatsClient(statsClient), pilosa.OptServerURI(advertiseURI), - pilosa.OptServerGRPCURI(grpcURI), + pilosa.OptServerGRPCURI(advertiseGRPCURI), pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)), pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts), pilosa.OptServerSerializer(proto.Serializer{}),