veyron/runtimes/google/ipc: add support for selecting an 'appropriate' IP address to publish on.
In many cases, the Listen call uses an address that does not
specify which network interface/address to use for communicating
with the service - e.g. ":0". In these cases, the address returned
by Listen is 'unspecified' - e.g. 0.0.0.0 or the [] equivalent for
IPv6. This change provides a mechanism for choosing an address to
use from the set of available ones. The default, built-in behaviour
is to prefer any IPv4 address over IPv6 and public/routable over
private/non-routable addresses. This means that a private IPv4
trumps a public IPv6; this is debatable at best.
This default can be overriden by provided an option to the Server
factory that specifies an alternative function to use for choosing
the preferred address. Such functions will be provided by the Profiles.
This is the first CL required to add support for DHCP roaming.
Change-Id: If95a91ea6d95c28f85edfb3312d1207e3b7e1ce1
diff --git a/runtimes/google/ipc/full_test.go b/runtimes/google/ipc/full_test.go
index 765c990..3135149 100644
--- a/runtimes/google/ipc/full_test.go
+++ b/runtimes/google/ipc/full_test.go
@@ -1055,70 +1055,6 @@
}
}
-// TestPublishOptions verifies that the options that are relevant to how
-// a server publishes its endpoints have the right effect.
-func TestPublishOptions(t *testing.T) {
- sm := imanager.InternalNew(naming.FixedRoutingID(0x555555555))
- ns := newNamespace()
- cases := []struct {
- opts []ipc.ServerOpt
- expect []string
- }{
- {[]ipc.ServerOpt{}, []string{"127.0.0.1", "127.0.0.1"}},
- {[]ipc.ServerOpt{veyron2.PublishAll}, []string{"127.0.0.1", "127.0.0.1"}},
- {[]ipc.ServerOpt{veyron2.PublishFirst}, []string{"127.0.0.1"}},
- {[]ipc.ServerOpt{veyron2.EndpointRewriteOpt("example1.com"), veyron2.EndpointRewriteOpt("example2.com")}, []string{"example2.com", "example2.com"}},
- {[]ipc.ServerOpt{veyron2.PublishFirst, veyron2.EndpointRewriteOpt("example.com")}, []string{"example.com"}},
- }
- for i, c := range cases {
- server, err := InternalNewServer(testContext(), sm, ns, append(c.opts, vc.FixedLocalID(serverID))...)
- if err != nil {
- t.Errorf("InternalNewServer failed: %v", err)
- continue
- }
- if _, err := server.Listen("tcp", "127.0.0.1:0"); err != nil {
- t.Errorf("server.Listen failed: %v", err)
- server.Stop()
- continue
- }
- if _, err := server.Listen("tcp", "127.0.0.1:0"); err != nil {
- t.Errorf("server.Listen failed: %v", err)
- server.Stop()
- continue
- }
- if err := server.Serve("mountpoint", &testServerDisp{}); err != nil {
- t.Errorf("server.Publish failed: %v", err)
- server.Stop()
- continue
- }
- servers, err := ns.Resolve(testContext(), "mountpoint")
- if err != nil {
- t.Errorf("mountpoint not found in mounttable")
- server.Stop()
- continue
- }
- var got []string
- for _, s := range servers {
- address, _ := naming.SplitAddressName(s)
- ep, err := inaming.NewEndpoint(address)
- if err != nil {
- t.Errorf("case #%d: server with invalid endpoint %q: %v", i, address, err)
- continue
- }
- host, _, err := net.SplitHostPort(ep.Addr().String())
- if err != nil {
- t.Errorf("case #%d: server endpoint with invalid address %q: %v", i, ep.Addr(), err)
- continue
- }
- got = append(got, host)
- }
- if want := c.expect; !reflect.DeepEqual(want, got) {
- t.Errorf("case #%d: expected mounted servers with addresses %q, got %q instead", i, want, got)
- }
- server.Stop()
- }
-}
-
// TestReconnect verifies that the client transparently re-establishes the
// connection to the server if the server dies and comes back (on the same
// endpoint).
@@ -1170,6 +1106,64 @@
}
}
+func TestPreferredAddress(t *testing.T) {
+ sm := imanager.InternalNew(naming.FixedRoutingID(0x555555555))
+ defer sm.Shutdown()
+ ns := newNamespace()
+ pa := func(string) (net.Addr, error) {
+ a := &net.IPAddr{}
+ a.IP = net.ParseIP("1.1.1.1")
+ return a, nil
+ }
+ server, err := InternalNewServer(testContext(), sm, ns, vc.FixedLocalID(serverID), veyron2.PreferredAddressOpt(pa))
+ if err != nil {
+ t.Errorf("InternalNewServer failed: %v", err)
+ }
+ defer server.Stop()
+ ep, err := server.Listen("tcp4", ":0")
+ iep := ep.(*inaming.Endpoint)
+ host, _, err := net.SplitHostPort(iep.Address)
+ if err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ if got, want := host, "1.1.1.1"; got != want {
+ t.Errorf("got %q, want %q", got, want)
+ }
+ // Won't override the specified address.
+ ep, err = server.Listen("tcp4", "127.0.0.1:0")
+ iep = ep.(*inaming.Endpoint)
+ host, _, err = net.SplitHostPort(iep.Address)
+ if err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ if got, want := host, "127.0.0.1"; got != want {
+ t.Errorf("got %q, want %q", got, want)
+ }
+}
+
+func TestPreferredAddressErrors(t *testing.T) {
+ sm := imanager.InternalNew(naming.FixedRoutingID(0x555555555))
+ defer sm.Shutdown()
+ ns := newNamespace()
+ paerr := func(string) (net.Addr, error) {
+ return nil, fmt.Errorf("oops")
+ }
+ server, err := InternalNewServer(testContext(), sm, ns, vc.FixedLocalID(serverID), veyron2.PreferredAddressOpt(paerr))
+ if err != nil {
+ t.Errorf("InternalNewServer failed: %v", err)
+ }
+ defer server.Stop()
+ ep, err := server.Listen("tcp4", ":0")
+ iep := ep.(*inaming.Endpoint)
+ host, _, err := net.SplitHostPort(iep.Address)
+ if err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ if got, want := host, "0.0.0.0"; got != want {
+ t.Errorf("got %q, want %q", got, want)
+ }
+}
+
type proxyHandle struct {
ns naming.Namespace
process *blackbox.Child
diff --git a/runtimes/google/ipc/server.go b/runtimes/google/ipc/server.go
index 4fa5f62..5b5b36d 100644
--- a/runtimes/google/ipc/server.go
+++ b/runtimes/google/ipc/server.go
@@ -3,11 +3,13 @@
import (
"fmt"
"io"
+ "net"
"reflect"
"strings"
"sync"
"time"
+ "veyron/runtimes/google/lib/netconfig"
"veyron/runtimes/google/lib/publisher"
inaming "veyron/runtimes/google/naming"
isecurity "veyron/runtimes/google/security"
@@ -44,27 +46,27 @@
stopped bool // whether the server has been stopped.
stoppedChan chan struct{} // closed when the server has been stopped.
ns naming.Namespace
- publishOpt veyron2.ServerPublishOpt // which endpoints to publish
- publishing bool // is some name being published?
+ preferredAddress func(network string) (net.Addr, error)
servesMountTable bool
}
func InternalNewServer(ctx context.T, streamMgr stream.Manager, ns naming.Namespace, opts ...ipc.ServerOpt) (ipc.Server, error) {
s := &server{
- ctx: ctx,
- streamMgr: streamMgr,
- publisher: publisher.New(ctx, ns, publishPeriod),
- listeners: make(map[stream.Listener]bool),
- stoppedChan: make(chan struct{}),
- ns: ns,
+ ctx: ctx,
+ streamMgr: streamMgr,
+ publisher: publisher.New(ctx, ns, publishPeriod),
+ listeners: make(map[stream.Listener]bool),
+ stoppedChan: make(chan struct{}),
+ preferredAddress: preferredIPAddress,
+ ns: ns,
}
for _, opt := range opts {
switch opt := opt.(type) {
+ case veyron2.PreferredAddressOpt:
+ s.preferredAddress = opt
case stream.ListenerOpt:
// Collect all ServerOpts that are also ListenerOpts.
s.listenerOpts = append(s.listenerOpts, opt)
- case veyron2.ServerPublishOpt:
- s.publishOpt = opt
case veyron2.ServesMountTableOpt:
s.servesMountTable = bool(opt)
}
@@ -108,6 +110,64 @@
return "", fmt.Errorf("unable to resolve %q to an endpoint", address)
}
+// preferredIPAddress returns the preferred IP address, which is,
+// a public IPv4 address, then any non-loopback IPv4, then a public
+// IPv6 address and finally any non-loopback/link-local IPv6
+func preferredIPAddress(network string) (net.Addr, error) {
+ interfaces, err := net.Interfaces()
+ if err != nil {
+ return nil, err
+ }
+ var any_ip4, any_ip6, pub_ip4, pub_ip6 net.Addr
+ for _, ifc := range interfaces {
+ addrs, err := ifc.Addrs()
+ if err != nil {
+ continue
+ }
+ for _, addr := range addrs {
+ ipn, ok := addr.(*net.IPNet)
+ if !ok {
+ continue
+ }
+ ip := ipn.IP
+ if ip == nil || ip.IsUnspecified() || ip.IsLoopback() || ip.IsMulticast() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() {
+ continue
+ }
+ if network == "tcp" || network == "tcp4" {
+ if t := ip.To4(); t != nil {
+ if any_ip4 == nil {
+ any_ip4 = addr
+ }
+ if pub_ip4 == nil && netconfig.IsGloballyRoutable(t) {
+ pub_ip4 = addr
+ }
+ }
+ }
+ if network == "tcp" || network == "tcp6" {
+ if t := ip.To16(); t != nil {
+ if any_ip6 == nil {
+ any_ip6 = addr
+ }
+ if pub_ip6 == nil && netconfig.IsGloballyRoutable(t) {
+ pub_ip6 = addr
+ }
+ }
+ }
+ }
+ }
+ switch {
+ case pub_ip4 != nil:
+ return pub_ip4, nil
+ case any_ip4 != nil:
+ return any_ip4, nil
+ case pub_ip6 != nil:
+ return pub_ip6, nil
+ case any_ip6 != nil:
+ return any_ip6, nil
+ }
+ return nil, fmt.Errorf("failed to find any usable address for %q", network)
+}
+
func (s *server) Listen(protocol, address string) (naming.Endpoint, error) {
s.Lock()
// Shortcut if the server is stopped, to avoid needlessly creating a
@@ -130,6 +190,30 @@
vlog.Errorf("ipc: Listen on %v %v failed: %v", protocol, address, err)
return nil, err
}
+ iep, ok := ep.(*inaming.Endpoint)
+ if !ok {
+ return nil, fmt.Errorf("ipc: Listen on %v %v failed translating internal endpoint data types", protocol, address)
+ }
+
+ // We know the endpoint format, so we crack it open...
+ switch iep.Protocol {
+ case "tcp", "tcp4", "tcp6":
+ host, port, err := net.SplitHostPort(iep.Address)
+ if err != nil {
+ return nil, err
+ }
+ ip := net.ParseIP(host)
+ if ip == nil {
+ return nil, fmt.Errorf("ipc: Listen(%q, %q) failed to parse IP address from address", protocol, address)
+ }
+ if ip.IsUnspecified() && s.preferredAddress != nil {
+ // Need to find a usable IP address.
+ if a, err := s.preferredAddress(protocol); err == nil {
+ iep.Address = net.JoinHostPort(a.String(), port)
+ }
+ }
+ }
+
s.Lock()
if s.stopped {
s.Unlock()
@@ -137,17 +221,15 @@
ln.Close()
return nil, errServerStopped
}
- publish := s.publishOpt == veyron2.PublishAll || !s.publishing
- s.publishing = true
s.listeners[ln] = true
- // We have a single goroutine per listener to accept new flows. Each flow is
- // served from its own goroutine.
+ // We have a single goroutine per listener to accept new flows.
+ // Each flow is served from its own goroutine.
s.active.Add(1)
if protocol == inaming.Network {
- go func(ln stream.Listener, ep naming.Endpoint, proxy string, publish bool) {
- s.proxyListenLoop(ln, ep, proxy, publish)
+ go func(ln stream.Listener, ep naming.Endpoint, proxy string) {
+ s.proxyListenLoop(ln, ep, proxy)
s.active.Done()
- }(ln, ep, proxyName, publish)
+ }(ln, ep, proxyName)
} else {
go func(ln stream.Listener, ep naming.Endpoint) {
s.listenLoop(ln, ep)
@@ -155,9 +237,7 @@
}(ln, ep)
}
s.Unlock()
- if publish {
- s.publisher.AddServer(s.publishEP(ep))
- }
+ s.publisher.AddServer(s.publishEP(ep))
return ep, nil
}
@@ -171,7 +251,7 @@
return naming.JoinAddressName(ep.String(), name)
}
-func (s *server) proxyListenLoop(ln stream.Listener, ep naming.Endpoint, proxy string, publish bool) {
+func (s *server) proxyListenLoop(ln stream.Listener, ep naming.Endpoint, proxy string) {
const (
min = 5 * time.Millisecond
max = 5 * time.Minute
@@ -180,9 +260,7 @@
s.listenLoop(ln, ep)
// The listener is done, so:
// (1) Unpublish its name
- if publish {
- s.publisher.RemoveServer(s.publishEP(ep))
- }
+ s.publisher.RemoveServer(s.publishEP(ep))
// (2) Reconnect to the proxy unless the server has been stopped
backoff := min
ln = nil
@@ -208,9 +286,7 @@
}
}
// (3) reconnected, publish new address
- if publish {
- s.publisher.AddServer(s.publishEP(ep))
- }
+ s.publisher.AddServer(s.publishEP(ep))
s.Lock()
s.listeners[ln] = true
s.Unlock()
diff --git a/runtimes/google/ipc/stream/manager/manager.go b/runtimes/google/ipc/stream/manager/manager.go
index 1921def..47646d5 100644
--- a/runtimes/google/ipc/stream/manager/manager.go
+++ b/runtimes/google/ipc/stream/manager/manager.go
@@ -13,7 +13,6 @@
"veyron/runtimes/google/ipc/version"
inaming "veyron/runtimes/google/naming"
- "veyron2"
"veyron2/ipc/stream"
"veyron2/naming"
"veyron2/verror"
@@ -122,17 +121,6 @@
}
func (m *manager) Listen(protocol, address string, opts ...stream.ListenerOpt) (stream.Listener, naming.Endpoint, error) {
- var rewriteEP string
- var filteredOpts []stream.ListenerOpt
- for _, o := range opts {
- if rewriteOpt, ok := o.(veyron2.EndpointRewriteOpt); ok {
- // Last one 'wins'.
- rewriteEP = string(rewriteOpt)
- } else {
- filteredOpts = append(filteredOpts, o)
- }
- }
- opts = filteredOpts
m.muListeners.Lock()
if m.shutdown {
m.muListeners.Unlock()
@@ -162,19 +150,7 @@
ln := newNetListener(m, netln, opts)
m.listeners[ln] = true
m.muListeners.Unlock()
-
- network, address := netln.Addr().Network(), netln.Addr().String()
- if network == "tcp" && len(rewriteEP) > 0 {
- if _, port, err := net.SplitHostPort(address); err != nil {
- return nil, nil, fmt.Errorf("%q not a valid address: %v", address, err)
- } else {
- address = net.JoinHostPort(rewriteEP, port)
- }
- }
- // We use protocol rather than network when creating the endpoint to
- // honour the original request to Listen even if tcp is used under the
- // covers.
- ep := version.Endpoint(protocol, address, m.rid)
+ ep := version.Endpoint(protocol, netln.Addr().String(), m.rid)
return ln, ep, nil
}
