Merge branch 'master' into shift-op

This commit is contained in:
Travis Turner 2019-01-25 12:57:51 -06:00
commit 779f89815d
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
9 changed files with 456 additions and 19 deletions

View file

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

View file

@ -36,6 +36,17 @@ The config file is in the [toml format](https://github.com/toml-lang/toml) and h
### All Options
#### Advertise
* Description: Address advertised by the server to other nodes in the cluster and to clients via the `/status` endpoint. Host defaults to the IP address represented by `bind` and port to 10101. If `bind` is set to `0.0.0.0` and `advertise` is not specified, then Pilosa will try to determine a reasonable, external IP address to use for `advertise`.
* Flag: `--advertise="192.168.1.100:10101"`
* Env: `PILOSA_BIND="192.168.1.100:10101"`
* Config:
```toml
advertise = 192.168.1.100:10101
```
#### Anti Entropy Interval
* Description: Interval at which the cluster will run its anti-entropy routine which ensures that all replicas of each fragment are in sync.
@ -50,7 +61,7 @@ The config file is in the [toml format](https://github.com/toml-lang/toml) and h
#### Bind
* Description: host:port on which the Pilosa server will listen for requests. Host defaults to localhost and port to 10101.
* Description: host:port on which the Pilosa server will listen for requests. Host defaults to localhost and port to 10101. If `bind` is set to `0.0.0.0` then Pilosa will listen on all available interfaces.
* Flag: `--bind="localhost:10101"`
* Env: `PILOSA_BIND="localhost:10101"`
* Config:
@ -115,6 +126,30 @@ The config file is in the [toml format](https://github.com/toml-lang/toml) and h
max-writes-per-request = 5000
```
#### Gossip Advertise Host
* Description: Host on which memberlist should advertise. Defaults to `advertise` host.
* Flag: `--gossip.advertise-host=192.168.1.100`
* Env: `PILOSA_GOSSIP_ADVERTISE_HOST=192.168.1.100
* Config:
```toml
[gossip]
advertise-host = 192.168.1.100
```
#### Gossip Advertise Port
* Description: Port on which memberlist should advertise. Defaults to `advertise` port.
* Flag: `--gossip.advertise-port=15001`
* Env: `PILOSA_GOSSIP_ADVERTISE_PORT=15001`
* Config:
```toml
[gossip]
advertise-port = 15001
```
#### Gossip Port
* Description: Port to which Pilosa should bind for internal communication. If more than one Pilosa server is running on the same host, the gossip port for each server must be unique.

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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