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

@@ -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.")

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

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()

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)
 }

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
+}

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)
+		}
+	}
+}

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")

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))