From efb9f97e61e0f2ffd496add3e11d84bdfb62d5fe Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Fri, 18 Jan 2019 10:29:19 -0600 Subject: [PATCH] Advertise address and listen on 0.0.0.0 This commit adds support for advertise address by using a new config option `advertise`, or by defaulting its value to that specified in `bind`. Also adds support for listening on 0.0.0.0 by trying to determine the preferred outbound IP to use for the advertise address. --- ctl/server.go | 4 + gossip/gossip.go | 30 ++++- http/client.go | 2 +- pilosa.go | 3 +- server/config.go | 215 +++++++++++++++++++++++++++++++-- server/config_internal_test.go | 153 +++++++++++++++++++++++ server/server.go | 30 ++++- server/server_test.go | 1 - 8 files changed, 420 insertions(+), 18 deletions(-) create mode 100644 server/config_internal_test.go diff --git a/ctl/server.go b/ctl/server.go index 5ed40dbf4..441f2b1e7 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -26,6 +26,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags := cmd.Flags() flags.StringVarP(&srv.Config.DataDir, "data-dir", "d", srv.Config.DataDir, "Directory to store pilosa data files.") flags.StringVarP(&srv.Config.Bind, "bind", "b", srv.Config.Bind, "Default URI on which pilosa should listen.") + flags.StringVar(&srv.Config.Advertise, "advertise", srv.Config.Advertise, "Address to advertise externally.") 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") @@ -49,6 +50,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { // Gossip flags.StringVarP(&srv.Config.Gossip.Port, "gossip.port", "", srv.Config.Gossip.Port, "Port to which pilosa should bind for internal state sharing.") + flags.StringVarP(&srv.Config.Gossip.AdvertiseHost, "gossip.advertise-host", "", srv.Config.Gossip.AdvertiseHost, "Host on which memberlist should advertise.") + flags.StringVarP(&srv.Config.Gossip.AdvertisePort, "gossip.advertise-port", "", srv.Config.Gossip.AdvertisePort, "Port on which memberlist should advertise.") + flags.StringSliceVarP(&srv.Config.Gossip.Seeds, "gossip.seeds", "", srv.Config.Gossip.Seeds, "Host with which to seed the gossip membership.") flags.StringVarP(&srv.Config.Gossip.Key, "gossip.key", "", srv.Config.Gossip.Key, "The path to file of the encryption key for gossip. The contents of the file should be either 16, 24, or 32 bytes to select AES-128, AES-192, or AES-256.") flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.StreamTimeout), "gossip.stream-timeout", "", (time.Duration)(srv.Config.Gossip.StreamTimeout), "Timeout for establishing a stream connection with a remote node for a full state sync.") diff --git a/gossip/gossip.go b/gossip/gossip.go index e3c552f88..1eee9dc3d 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -241,14 +241,29 @@ func NewMemberSet(cfg Config, api *pilosa.API, options ...memberSetOption) (*mem } } + //////////////////// // memberlist config conf := memberlist.DefaultWANConfig() conf.Transport = g.transport.net conf.Name = api.Node().ID conf.BindAddr = api.Node().URI.Host conf.BindPort = port - conf.AdvertisePort = port - conf.AdvertiseAddr = hostToIP(api.Node().URI.Host) + // AdvertisePort + if cfg.AdvertisePort != "" { + if p, err := strconv.Atoi(cfg.Port); err != nil { + return nil, fmt.Errorf("convert advertise port: %s", err) + } else { + conf.AdvertisePort = p + } + } else { + conf.AdvertisePort = port + } + // AdvertiseHost + if cfg.AdvertiseHost != "" { + conf.AdvertiseAddr = cfg.AdvertiseHost + } else { + conf.AdvertiseAddr = hostToIP(api.Node().URI.Host) + } // conf.TCPTimeout = time.Duration(cfg.StreamTimeout) conf.SuspicionMult = cfg.SuspicionMult @@ -509,7 +524,16 @@ func newTransport(conf *memberlist.Config) (*memberlist.NetTransport, error) { // Config holds toml-friendly memberlist configuration. type Config struct { // Port indicates the port to which pilosa should bind for internal state sharing. - Port string `toml:"port"` + Port string `toml:"port"` + + // AdvertiseHost is the hostname or IP other nodes should use to connect to + // this host. If left blank, the value for Host will be used. This is useful + // in some proxy and NAT scenarios. + AdvertiseHost string `toml:"advertise-host"` + // AdvertisePort is the port other nodes will use to connect to this one. + // Behaves like AdvertiseHost. + AdvertisePort string `toml:"advertise-port"` + Seeds []string `toml:"seeds"` Key string `toml:"key"` // StreamTimeout is the timeout for establishing a stream connection with diff --git a/http/client.go b/http/client.go index 78382b81b..92f6269fe 100644 --- a/http/client.go +++ b/http/client.go @@ -1015,7 +1015,7 @@ func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, erro if resp != nil { resp.Body.Close() } - return nil, errors.Wrap(err, "executing request") + return nil, errors.Wrap(err, "getting response") } if resp.StatusCode < 200 || resp.StatusCode >= 300 { defer resp.Body.Close() diff --git a/pilosa.go b/pilosa.go index 550696a31..28d89e864 100644 --- a/pilosa.go +++ b/pilosa.go @@ -166,7 +166,6 @@ func stringSlicesAreEqual(a, b []string) bool { func AddressWithDefaults(addr string) (*URI, error) { if addr == "" { return defaultURI(), nil - } else { - return NewURIFromAddress(addr) } + return NewURIFromAddress(addr) } diff --git a/server/config.go b/server/config.go index 6c159721e..6223f07da 100644 --- a/server/config.go +++ b/server/config.go @@ -15,10 +15,17 @@ package server import ( + "context" + "fmt" + "log" + "net" + "strconv" + "strings" "time" "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/toml" + "github.com/pkg/errors" "github.com/uber/jaeger-client-go" ) @@ -37,9 +44,15 @@ type Config struct { // DataDir is the directory where Pilosa stores both indexed data and // running state such as cluster topology information. DataDir string `toml:"data-dir"` + // Bind is the host:port on which Pilosa will listen. Bind string `toml:"bind"` + // Advertise 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 Bind is listening on. + Advertise string `toml:"advertise"` + // MaxWritesPerRequest limits the number of mutating commands that can be in // a single request to the server. This includes Set, Clear, // SetRowAttrs & SetColumnAttrs. @@ -110,22 +123,17 @@ func NewConfig() *Config { DataDir: "~/.pilosa", Bind: ":10101", MaxWritesPerRequest: 5000, - // LogPath: "", - // Verbose: false, - TLS: TLSConfig{}, + TLS: TLSConfig{}, } // Cluster config. c.Cluster.Disabled = false - // c.Cluster.Coordinator = false c.Cluster.ReplicaN = 1 c.Cluster.Hosts = []string{} c.Cluster.LongQueryTime = toml.Duration(time.Minute) // Gossip config. c.Gossip.Port = "14000" - // c.Gossip.Seeds = []string{} - // c.Gossip.Key = "" c.Gossip.StreamTimeout = toml.Duration(10 * time.Second) c.Gossip.SuspicionMult = 4 c.Gossip.PushPullInterval = toml.Duration(30 * time.Second) @@ -140,7 +148,6 @@ func NewConfig() *Config { // Metric config. c.Metric.Service = "none" - // c.Metric.Host = "" c.Metric.PollInterval = toml.Duration(0 * time.Minute) c.Metric.Diagnostics = true @@ -150,3 +157,197 @@ func NewConfig() *Config { return c } + +// validateAddrs controls the address fields in the Config object +// and fills in any blanks. +// The addresses fields must be guaranteed by the caller to either be +// completely empty, or have both a host part and a port part +// separated by a colon. In the latter case either can be empty to +// 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) + 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) + if err != nil { + return errors.Wrap(err, "validating listen address") + } + cfg.Bind = schemeHostPortString(listenScheme, listenHost, listenPort) + + return nil +} + +// validateAdvertiseAddr validates and normalizes an address accessible +// Ensures that if the "host" part is empty, it gets filled in with +// 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) + if err != nil { + return "", "", "", errors.Wrap(err, "getting listen address") + } + + advScheme, advHostPort := splitScheme(advAddr) + advHost, advPort := "", "" + if advHostPort != "" { + var err error + advHost, advPort, err = net.SplitHostPort(advHostPort) + if err != nil { + return "", "", "", errors.Wrapf(err, "splitting host port: %s", advHostPort) + } + } + // If no advertise scheme was specified, use the one from + // the listen address. + if advScheme == "" { + advScheme = listenScheme + } + // If there was no port number, reuse the one from the listen + // address. + if advPort == "" || advPort == "0" { + advPort = listenPort + } + // Resolve non-numeric to numeric. + portNumber, err := net.DefaultResolver.LookupPort(ctx, "tcp", advPort) + if err != nil { + return "", "", "", errors.Wrapf(err, "looking up non-numeric port: %v", advPort) + } + advPort = strconv.Itoa(portNumber) + + // If the advertise host is empty, then we have two cases. + if advHost == "" { + if listenHost == "0.0.0.0" { + advHost = outboundIP().String() + } else { + advHost = listenHost + } + } + return advScheme, advHost, advPort, nil +} + +// outboundIP gets the preferred outbound ip of this machine. +func outboundIP() net.IP { + // This is not actually making a connection to 8.8.8.8. + // net.Dial() selects the IP address that would be used + // if an actual connection to 8.8.8.8 were made, so this + // choice of address is just meant to ensure that an + // external address is returned (as opposed to a local + // address like 127.0.0.1). + conn, err := net.Dial("udp", "8.8.8.8:80") + if err != nil { + log.Fatal(err) + } + defer conn.Close() + + localAddr := conn.LocalAddr().(*net.UDPAddr) + + return localAddr.IP +} + +// validateListenAddr validates and normalizes an address suitable for +// use with net.Listen(). This accepts an empty "host" part to signify +// 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) + if err != nil { + return "", "", "", errors.Wrap(err, "getting listen address") + } + rHost, rPort, err := resolveAddr(ctx, host, port) + if err != nil { + return "", "", "", errors.Wrap(err, "resolving address") + } + return scheme, rHost, rPort, nil +} + +// splitScheme returns two strings: the scheme and the hostPort. +func splitScheme(addr string) (string, string) { + parts := strings.SplitN(addr, "://", 2) + if len(parts) == 1 { + return "", addr + } + return parts[0], parts[1] +} + +func schemeHostPortString(scheme, host, port string) string { + var s string + if scheme != "" { + s += fmt.Sprintf("%s://", scheme) + } + return s + net.JoinHostPort(host, port) +} + +// splitAddr returns scheme, host, port as strings. +func splitAddr(addr string) (string, string, string, error) { + scheme, hostPort := splitScheme(addr) + host, port := "", "" + if hostPort != "" { + var err error + host, port, err = net.SplitHostPort(hostPort) + if err != nil { + return "", "", "", errors.Wrapf(err, "splitting host port: %s", hostPort) + } + } + // It's not ideal to have a default here, but the alterative + // results in a port of 0, which causes Pilosa to listen on + // a random port. + if port == "" { + port = "10101" + } + return scheme, host, port, nil +} + +// resolveAddr resolves non-numeric addresses to numeric (IP, port) addresses. +func resolveAddr(ctx context.Context, host, port string) (string, string, error) { + resolver := net.DefaultResolver + + // Resolve the port number. This may translate service names + // e.g. "postgresql" to a numeric value. + portNumber, err := resolver.LookupPort(ctx, "tcp", port) + if err != nil { + return "", "", errors.Wrapf(err, "resolving up port: %v", port) + } + port = strconv.Itoa(portNumber) + + // Resolve the address. + if host == "" || host == "localhost" { + return host, port, nil + } + + addr, err := lookupAddr(ctx, resolver, host) + if err != nil { + return "", "", errors.Wrap(err, "looking up address") + } + return addr, port, nil +} + +// lookupAddr resolves the given address/host to an IP address. If +// multiple addresses are resolved, it returns the first IPv4 address +// available if there is one, otherwise the first address. +func lookupAddr(ctx context.Context, resolver *net.Resolver, host string) (string, error) { + // Resolve the IP address or hostname to an IP address. + addrs, err := resolver.LookupIPAddr(ctx, host) + if err != nil { + return "", errors.Wrap(err, "looking up IP addresses") + } + if len(addrs) == 0 { + return "", fmt.Errorf("cannot resolve %q to an address", host) + } + + // LookupIPAddr() can return a mix of IPv6 and IPv4 + // addresses. Return the first IPv4 address if possible. + for _, addr := range addrs { + if ip := addr.IP.To4(); ip != nil { + return ip.String(), nil + } + } + + // No IPv4 address, return the first resolved address instead. + return addrs[0].String(), nil +} diff --git a/server/config_internal_test.go b/server/config_internal_test.go new file mode 100644 index 000000000..2b29ea51e --- /dev/null +++ b/server/config_internal_test.go @@ -0,0 +1,153 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "context" + "net" + "os" + "strings" + "testing" +) + +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() + 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{":10101", ":10101"}}, + {"", + addrs{":", ""}, + addrs{":10101", ":10101"}}, + {"", + addrs{"", ":"}, + addrs{":10101", ":10101"}}, + {"", + addrs{":", ":"}, + addrs{":10101", ":10101"}}, + // Listener :port. + {"", + addrs{":1234", ""}, + addrs{":1234", ":1234"}}, + // Listener with host:port. + {"", + addrs{hostAddr + ":10101", ""}, + addrs{hostAddr + ":10101", hostAddr + ":10101"}}, + // Listener with host:. + {"", + addrs{hostAddr + ":", ""}, + addrs{hostAddr + ":10101", hostAddr + ":10101"}}, + // Listener with scheme:. + {"", + addrs{"http://" + hostAddr + ":", ""}, + addrs{"http://" + hostAddr + ":10101", "http://" + hostAddr + ":10101"}}, + // Listener with localhost:port. + {"", + addrs{"localhost:1234", ""}, + addrs{"localhost:1234", "localhost:1234"}}, + // Listener with localhost:. + {"", + addrs{"localhost:", ""}, + addrs{"localhost:10101", "localhost:10101"}}, + // Listener and advertise addresses. + {"", + addrs{hostAddr + ":1234", hostAddr + ":"}, + addrs{hostAddr + ":1234", hostAddr + ":1234"}}, + // Explicit port number in advertise addr. + {"", + addrs{hostAddr + ":1234", hostAddr + ":7890"}, + addrs{hostAddr + ":1234", hostAddr + ":7890"}}, + // Use a non-numeric port number. + {"", + addrs{":postgresql", ""}, + addrs{":5432", ":5432"}}, + // Advertise port 0 means reuse listen port. + {"", + addrs{":1234", ":0"}, + addrs{":1234", ":1234"}}, + // Listen on all interfaces. Determine advertise address. + {"", + addrs{"0.0.0.0:1234", ""}, + addrs{"0.0.0.0:1234", 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 advertise address", + addrs{"localhost:foo", ""}, + addrs{}}, + {"no such host", + addrs{"333.333.333.333:1234", ""}, + addrs{}}, + } + + for i, test := range tests { + c := NewConfig() + + c.Bind = test.in.bind + c.Advertise = test.in.advertise + + err := c.validateAddrs(context.Background()) + + if err != nil && test.expErr == "" { + t.Fatal(err) + } else if err == nil && test.expErr != "" { + t.Fatalf("test %d: expected error string to contain %s, but got no error", i, test.expErr) + } else if err != nil && test.expErr != "" { + if strings.Contains(err.Error(), test.expErr) { + continue + } else { + t.Fatalf("test %d: expected error string to contain %s, but got %s", i, test.expErr, err.Error()) + } + } + + if c.Bind != test.exp.bind { + t.Fatalf("test %d: bind address: expected %s, but got %s", i, test.exp.bind, c.Bind) + } else if c.Advertise != test.exp.advertise { + t.Fatalf("test %d: advertise address: expected %s, but got %s", i, test.exp.advertise, c.Advertise) + } + } +} diff --git a/server/server.go b/server/server.go index 5f99e510e..836b7fec7 100644 --- a/server/server.go +++ b/server/server.go @@ -21,6 +21,7 @@ package server import ( "bytes" + "context" "crypto/tls" "io" "log" @@ -78,6 +79,7 @@ type Command struct { Handler pilosa.Handler API *pilosa.API ln net.Listener + listenURI *pilosa.URI closeTimeout time.Duration serverOptions []pilosa.ServerOption @@ -151,7 +153,7 @@ func (m *Command) Start() (err error) { return errors.Wrap(err, "opening server") } - m.logger.Printf("listening as %s\n", m.API.Node().URI) + m.logger.Printf("listening as %s\n", m.listenURI) return nil } @@ -187,6 +189,15 @@ func (m *Command) SetupServer() error { } m.logger.Printf("%s %s, build time %s\n", productName, pilosa.Version, pilosa.BuildTime) + // validateAddrs sets the appropriate values for Bind and Advertise + // based on the inputs. It is not responsible for applying defaults, although + // it does provide a non-zero port (10101) in the case where no port is specified. + // The alternative would be to use port 0, which would choose a random port, but + // currently that's not what we want. + if err := m.Config.validateAddrs(context.Background()); err != nil { + return errors.Wrap(err, "validating addresses") + } + uri, err := pilosa.AddressWithDefaults(m.Config.Bind) if err != nil { return errors.Wrap(err, "processing bind address") @@ -231,8 +242,20 @@ func (m *Command) SetupServer() error { uri.SetPort(uint16(m.ln.Addr().(*net.TCPAddr).Port)) } + // Save listenURI for later reference. + m.listenURI = uri + c := http.GetHTTPClient(TLSConfig) + // Get advertise address as uri. + advertiseURI, err := pilosa.AddressWithDefaults(m.Config.Advertise) + if err != nil { + return errors.Wrap(err, "processing advertise address") + } + if advertiseURI.Port == 0 { + advertiseURI.SetPort(uri.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.") @@ -258,7 +281,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerSystemInfo(gopsutil.NewSystemInfo()), pilosa.OptServerGCNotifier(gcnotify.NewActiveGCNotifier()), pilosa.OptServerStatsClient(statsClient), - pilosa.OptServerURI(uri), + pilosa.OptServerURI(advertiseURI), pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)), pilosa.OptServerPrimaryTranslateStoreFunc(http.NewTranslateStore), pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts), @@ -295,7 +318,6 @@ func (m *Command) SetupServer() error { http.OptHandlerCloseTimeout(m.closeTimeout), ) return errors.Wrap(err, "new handler") - } // setupNetworking sets up internode communication based on the configuration. @@ -310,7 +332,7 @@ func (m *Command) setupNetworking() error { } // get the host portion of addr to use for binding - gossipHost := m.API.Node().URI.Host + gossipHost := m.listenURI.Host m.gossipTransport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger.Logger()) if err != nil { return errors.Wrap(err, "getting transport") diff --git a/server/server_test.go b/server/server_test.go index 66e2cc112..3b6be7efa 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -468,7 +468,6 @@ func TestClusteringNodesReplica1(t *testing.T) { // Create new main with the same config. config := cluster[2].Command.Config config.Translation.MapSize = 100000 - // config.Bind = cluster[2].API.Node().URI.HostPort() // this isn't necessary, but makes the test run way faster config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port))