veyron2/ipc: add RoamingListen support.
This CL adds support for 'roaming' whereby a server may dynamically
change its IP address in response to network changes. It adds a new
API method, RoamingListen, that listens on a config.Publisher channel
for a prescribed set of messages that encode the initial and changing
be changed with an option to NewServer and this usage is only expected
to be required for unit tests and other special cases.
The 'roaming' profile, veyron/profiles/roaming is required to enable
this support. This also brings in support for GCE's 1:1 NAT configuration.
This CL is preliminary and not fully tested, in particular more
unit tests are required for the core RoamingListen call and for
remove servers from the config.Publisher stream.
Change-Id: Ic09bfe0edc6391702e42468e88ae5f4d56bc3f4a
diff --git a/lib/netstate/netstate.go b/lib/netstate/netstate.go
index e35e595..f96b0e8 100644
--- a/lib/netstate/netstate.go
+++ b/lib/netstate/netstate.go
@@ -2,7 +2,7 @@
// of network addresess, for determining changes to those addresses and for
// selecting from amongst them according to some set of policies that are
// implemented by applying simple predicates (functions with names of the form
-// Is<condition> to filter or find the first matching address from a list
+// Is<condition>) to filter or find the first matching address from a list
// of addresses. The intent is to make it easy to create policies that do
// things like 'find the first IPv4 unicast address that is globally routable,
// failing that use a private IPv4 address, and failing that, an IPv6 address'.
@@ -10,7 +10,7 @@
// A typical usage would be:
//
// state, _ := netstate.GetAccessibleIPs()
-// first := netstate.First(netstate.IsPublicIPv4)
+// first := state.First(netstate.IsPublicIPv4)
// // first will contain the first public IPv4 address or be nil.
//
// The example policy described above would be implemented using a
@@ -25,7 +25,7 @@
// The term 'public' is used to refer to any globally routable IP address.
//
// All IPv6 addresses are intended to be 'public', but any starting with
-// fc00::/7 are (RFC4193) are reserved for private use, but the go
+// fc00::/7 (RFC4193) are reserved for private use, but the go
// net libraries do not appear to recognise this. Similarly fe80::/10
// (RFC 4291) are reserved for 'site-local' usage, but again this is not
// implemented in the go libraries. Any developer who needs to distinguish
@@ -57,6 +57,14 @@
return strings.TrimRight(r, " ")
}
+func FromIPAddr(a []*net.IPAddr) AddrList {
+ var al AddrList
+ for _, a := range a {
+ al = append(al, a)
+ }
+ return al
+}
+
// GetAll gets all of the available addresses on the device, including
// loopback addresses.
func GetAll() (AddrList, error) {
@@ -77,12 +85,13 @@
// GetAccessibleIPs returns all of the IP addresses on the device that are
// accessible to other devices - i.e. excluding loopback etc.
+// The IP addresses returned will be host addresses.
func GetAccessibleIPs() (AddrList, error) {
all, err := GetAll()
if err != nil {
return nil, err
}
- return all.Filter(IsAccessibleIP), nil
+ return all.Filter(IsAccessibleIP).Map(ConvertToIPHost), nil
}
type Predicate func(a net.Addr) bool
@@ -109,7 +118,30 @@
return nil
}
-func IsIPNetwork(n string) bool {
+type Mapper func(a net.Addr) net.Addr
+
+// Map will apply the Mapper function to all of the items in its receiver
+// and return a new AddrList containing all of the non-nil results from
+// said calls.
+func (al AddrList) Map(mapper Mapper) AddrList {
+ var ral AddrList
+ for _, a := range al {
+ if na := mapper(a); na != nil {
+ ral = append(ral, na)
+ }
+ }
+ return ral
+}
+
+// Convert the net.Addr argument into an instance of net.Addr that contains
+// an IP host address (as opposed to a network CIDR for example).
+func ConvertToIPHost(a net.Addr) net.Addr {
+ return AsIPAddr(a)
+}
+
+// IsIPProtocol returns true if its parameter is one of the allowed
+// network/protocol values for IP.
+func IsIPProtocol(n string) bool {
switch n {
case "ip+net", "tcp", "tcp4", "tcp6", "udp":
return true
@@ -119,22 +151,30 @@
}
}
-// AsIP returns its argument as a net.IP if that's possible.
-func AsIP(a net.Addr) net.IP {
- ipn, ok := a.(*net.IPNet)
- if ok {
- return ipn.IP
+// AsIPAddr returns its argument as a net.IPAddr if that's possible.
+func AsIPAddr(a net.Addr) *net.IPAddr {
+ if v, ok := a.(*net.IPAddr); ok {
+ return v
}
- ipa, ok := a.(*net.IPAddr)
- if ok {
- return ipa.IP
+ if ipn, ok := a.(*net.IPNet); ok {
+ return &net.IPAddr{IP: ipn.IP}
}
switch a.Network() {
- default:
- return nil
case "ip+net", "tcp", "tcp4", "tcp6", "udp":
+ if r := net.ParseIP(a.String()); r != nil {
+ return &net.IPAddr{IP: r}
+ }
}
- return net.ParseIP(a.String())
+ return nil
+}
+
+// AsIP returns its argument as a net.IP if that's possible.
+func AsIP(a net.Addr) net.IP {
+ ipAddr := AsIPAddr(a)
+ if ipAddr == nil {
+ return nil
+ }
+ return ipAddr.IP
}
// IsUnspecified returns true if its argument is an unspecified IP address
diff --git a/lib/netstate/netstate_test.go b/lib/netstate/netstate_test.go
index 262d5e5..746341c 100644
--- a/lib/netstate/netstate_test.go
+++ b/lib/netstate/netstate_test.go
@@ -1,9 +1,7 @@
package netstate_test
import (
- "fmt"
"net"
- "os"
"reflect"
"testing"
@@ -15,6 +13,7 @@
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
+ all = all.Map(netstate.ConvertToIPHost)
accessible, err := netstate.GetAccessibleIPs()
if err != nil {
t.Fatalf("unexpected error: %s", err)
@@ -33,7 +32,6 @@
func mkAddr(n, a string) net.Addr {
ip := net.ParseIP(a)
- fmt.Fprintf(os.Stderr, "%s -> %d\n", a, len(ip))
return netstate.AsAddr(n, ip)
}
diff --git a/profiles/doc.go b/profiles/doc.go
index bfd4bfc..f5fc4dc 100644
--- a/profiles/doc.go
+++ b/profiles/doc.go
@@ -31,37 +31,3 @@
// networking.
// TODO(cnicolaou,ashankar): add this
package profiles
-
-import (
- "veyron2"
- "veyron2/config"
-)
-
-type generic struct{}
-
-// New returns a new instance of a very generic Profile. It can be used
-// as a default by Runtime implementations, in unit tests etc.
-func New() veyron2.Profile {
- return &generic{}
-}
-
-func (g *generic) Name() string {
- return "generic"
-}
-
-func (g *generic) Runtime() string {
- return ""
-}
-
-func (g *generic) Platform() *veyron2.Platform {
- p, _ := Platform()
- return p
-}
-
-func (g *generic) Init(rt veyron2.Runtime, _ *config.Publisher) {
- rt.Logger().VI(1).Infof("%s", g)
-}
-
-func (g *generic) String() string {
- return "generic profile on " + g.Platform().String()
-}
diff --git a/profiles/gce/init.go b/profiles/gce/init.go
index 58a148d..915a9f6 100644
--- a/profiles/gce/init.go
+++ b/profiles/gce/init.go
@@ -5,6 +5,7 @@
package gce
import (
+ "net"
"veyron/profiles"
"veyron2"
@@ -18,7 +19,9 @@
rt.RegisterProfile(&profile{})
}
-type profile struct{}
+type profile struct {
+ publicAddress net.Addr
+}
func (p *profile) Name() string {
return "GCE"
@@ -37,17 +40,23 @@
return "net " + p.Platform().String()
}
-func (p *profile) Init(rt veyron2.Runtime, publisher *config.Publisher) {
- if !gce.RunningOnGCE() {
- panic("GCE profile used on a non-GCE system")
+func (p *profile) AddressChooser() veyron2.AddressChooser {
+ return func(network string, addrs []net.Addr) (net.Addr, error) {
+ return p.publicAddress, nil
}
- go addressPublisher(rt, publisher)
}
-func addressPublisher(rt veyron2.Runtime, p *config.Publisher) {
- ch := make(chan config.Setting)
- p.CreateStream("net", "network configuration", ch)
- for {
- // TODO(cnicolaou): publish local and external address..
+func (p *profile) Init(rt veyron2.Runtime, publisher *config.Publisher) {
+ if !gce.RunningOnGCE() {
+ return
+ // TODO(cnicolaou): add error return to init
+ //return fmt.Errorf("GCE profile used on a non-GCE system")
+ }
+ if ip, err := gce.ExternalIPAddress(); err != nil {
+ return
+ // TODO(cnicolaou): add error return to init
+ // return err
+ } else {
+ p.publicAddress = &net.IPAddr{IP: ip}
}
}
diff --git a/profiles/generic.go b/profiles/generic.go
new file mode 100644
index 0000000..7b493e0
--- /dev/null
+++ b/profiles/generic.go
@@ -0,0 +1,41 @@
+package profiles
+
+import (
+ "veyron2"
+ "veyron2/config"
+
+ "veyron/profiles/internal"
+)
+
+type generic struct{}
+
+// New returns a new instance of a very generic Profile. It can be used
+// as a default by Runtime implementations, in unit tests etc.
+func New() veyron2.Profile {
+ return &generic{}
+}
+
+func (*generic) Name() string {
+ return "generic"
+}
+
+func (*generic) Runtime() string {
+ return ""
+}
+
+func (*generic) Platform() *veyron2.Platform {
+ p, _ := Platform()
+ return p
+}
+
+func (*generic) AddressChooser() veyron2.AddressChooser {
+ return internal.IPAddressChooser
+}
+
+func (g *generic) Init(rt veyron2.Runtime, _ *config.Publisher) {
+ rt.Logger().VI(1).Infof("%s", g)
+}
+
+func (g *generic) String() string {
+ return "generic profile on " + g.Platform().String()
+}
diff --git a/profiles/internal/util.go b/profiles/internal/util.go
new file mode 100644
index 0000000..156a973
--- /dev/null
+++ b/profiles/internal/util.go
@@ -0,0 +1,25 @@
+package internal
+
+import (
+ "fmt"
+ "net"
+
+ "veyron/lib/netstate"
+)
+
+// IPAddressChooser 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 IPAddressChooser(network string, addrs []net.Addr) (net.Addr, error) {
+ if !netstate.IsIPProtocol(network) {
+ return nil, fmt.Errorf("can't support network protocol %q", network)
+ }
+ al := netstate.AddrList(addrs).Map(netstate.ConvertToIPHost)
+ for _, predicate := range []netstate.Predicate{netstate.IsPublicUnicastIPv4,
+ netstate.IsUnicastIPv4, netstate.IsPublicUnicastIPv6} {
+ if a := al.First(predicate); a != nil {
+ return a, nil
+ }
+ }
+ return nil, fmt.Errorf("failed to find any usable address for %q", network)
+}
diff --git a/profiles/net/gce_other.go b/profiles/net/gce_other.go
deleted file mode 100644
index 1abfc5c..0000000
--- a/profiles/net/gce_other.go
+++ /dev/null
@@ -1,12 +0,0 @@
-// +build !linux
-
-package net
-
-import (
- "veyron2"
- "veyron2/config"
-)
-
-func handleGCE(veyron2.Runtime, *config.Publisher) bool {
- return false
-}
diff --git a/profiles/net/init.go b/profiles/net/init.go
deleted file mode 100644
index 6f61d3e..0000000
--- a/profiles/net/init.go
+++ /dev/null
@@ -1,193 +0,0 @@
-// +build linux darwin
-
-// Package net provides a network-aware Profile that provides appropriate
-// options and configuration for a variety of network configurations, including
-// being behind 1-1 NATs, using dhcp and auto-configuration for being on
-// Google Compute Engine.
-//
-// The config.Publisher mechanism is used for communicating networking
-// settings to other components of the application. The Settings stream
-// is called "net" (net.StreamName) and the Settings sent over it are
-// those defined by the ipc package.
-// TODO(cnicolaou): define the settings in the ipc package, not here.
-package net
-
-import (
- "flag"
- "net"
-
- "veyron2"
- "veyron2/config"
- "veyron2/rt"
-
- "veyron/lib/netconfig"
- "veyron/profiles"
-)
-
-const (
- StreamName = "net"
-
- // TODO(cnicolaou): these will eventually be defined in the veyron2/ipc
- // package.
- ProtocolSetting = "Protocol"
- ListenSpecSetting = "ListenSpec"
- AddPublishAddressSetting = "AddPublishAddr"
- RmPublishAddressSetting = "RmPublishAddr"
-)
-
-var (
- listenProtocolFlag = config.TCPProtocolFlag{"tcp4"}
- listenSpecFlag = config.IPHostPortFlag{Port: ":0"}
-)
-
-func init() {
- flag.Var(&listenProtocolFlag, "veyron.tcpprotocol", "protocol to listen with")
- flag.Var(&listenSpecFlag, "veyron.tcpaddress", "address to listen on")
- rt.RegisterProfile(New())
-}
-
-type profile struct{}
-
-func New() veyron2.Profile {
- return &profile{}
-}
-
-func (p *profile) Platform() *veyron2.Platform {
- platform, _ := profiles.Platform()
- return platform
-}
-
-func (p *profile) Name() string {
- return "net"
-}
-
-func (p *profile) Runtime() string {
- return ""
-}
-
-func (p *profile) String() string {
- return "net " + p.Platform().String()
-}
-
-func (p *profile) Init(rt veyron2.Runtime, publisher *config.Publisher) {
- log := rt.Logger()
-
- // TODO(cnicolaou): figure out the correct heuristics for using IPv6.
- _, _, first := firstUsableIPv4()
- if first == nil {
- log.Infof("failed to find any usable IP addresses at startup")
- }
- public := publicIP(first)
-
- // We now know that there is an IP address to listen on, and whether
- // it's public or private.
-
- // Our address is private, so we test for running on GCE
- // and for its 1:1 NAT configuration. handleGCE returns true
- // if we are indeed running on GCE.
- if !public && handleGCE(rt, publisher) {
- return
- }
-
- // Create stream in Init function to avoid a race between any
- // goroutines started here and consumers started after Init returns.
- ch := make(chan config.Setting)
- stop, err := publisher.CreateStream(StreamName, "network configuration", ch)
- if err != nil {
- log.Errorf("failed to create publisher: %s", err)
- return
- }
- go monitorAndPublishNetworkSettings(rt, stop, ch, listenProtocolFlag.Protocol, listenSpecFlag)
-}
-
-func publishInitialSettings(ch chan<- config.Setting, protocol, listenSpec string, addr net.IP) {
- for _, setting := range []config.Setting{
- config.NewString(ProtocolSetting, "protocol to listen with", protocol),
- config.NewString(ListenSpecSetting, "address spec to listen on", listenSpec),
- config.NewIP(AddPublishAddressSetting, "address to publish", addr),
- } {
- ch <- setting
- }
-}
-
-// monitorNetworkSettings will publish initial Settings and then
-// monitor network configuration changes and publish subsequent
-// Settings to reflect any changes detected. It will never publish an
-// RmPublishAddressSetting without first sending an AddPublishAddressSetting.
-func monitorAndPublishNetworkSettings(rt veyron2.Runtime, stop <-chan struct{},
- ch chan<- config.Setting,
- listenProtocol string, listenSpec config.IPHostPortFlag) {
- defer close(ch)
-
- log := rt.Logger()
-
- prev4, _, prevAddr := firstUsableIPv4()
- // TODO(cnicolaou): check that prev4 is one of the IPs that a hostname
- // resolved to, if we used a hostname in the listenspec.
-
- // prevAddr may be nil if we are currently offline.
-
- log.Infof("Initial Settings: %s %s publish %s", listenProtocol, listenSpec, prevAddr)
- publishInitialSettings(ch, listenProtocol, listenSpec.String(), prevAddr)
-
- // Start the dhcp watcher.
- watcher, err := netconfig.NewNetConfigWatcher()
- if err != nil {
- log.VI(2).Infof("Failed to get new config watcher: %s", err)
- <-stop
- return
- }
-
- for {
- select {
- case <-watcher.Channel():
- cur4, _, _ := ipState()
- removed := findRemoved(prev4, cur4)
- added := findAdded(prev4, cur4)
- log.VI(2).Infof("Previous: %d: %s", len(prev4), prev4)
- log.VI(2).Infof("Current : %d: %s", len(cur4), cur4)
- log.VI(2).Infof("Added : %d: %s", len(added), added)
- log.VI(2).Infof("Removed : %d: %s", len(removed), removed)
- if len(removed) == 0 && len(added) == 0 {
- log.VI(2).Infof("Network event that lead to no address changes since our last 'baseline'")
- continue
- }
- if len(added) == 0 {
- // Nothing has been added, but something has been removed.
- if !removed.has(prevAddr) {
- log.VI(1).Infof("An address we were not using was removed")
- continue
- }
- log.Infof("Publish address is no longer available: %s", prevAddr)
- // Our currently published address has been removed and
- // a new one has not been added.
- // TODO(cnicolaou): look for a new address to use right now.
- prevAddr = nil
- continue
- }
- log.Infof("At least one address has been added and zero or more removed")
- // something has been added, and maybe something has been removed
- ifc, newAddr := added.first()
- if newAddr != nil && (prevAddr == nil || removed.has(prevAddr)) {
- log.Infof("Address being changed from %s to %s:%s",
- prevAddr, ifc, newAddr)
- ch <- config.NewIP(AddPublishAddressSetting, "new dhcp address to publish", newAddr)
- ch <- config.NewIP(RmPublishAddressSetting, "remove address", prevAddr)
- prevAddr = newAddr
- prev4 = cur4
- log.VI(2).Infof("Network baseline set to %s", cur4)
- }
- case <-stop:
- return
- }
- }
-}
-
-func firstUsableIPv4() (ipAndIf, string, net.IP) {
- v4, _, _ := publicIPState()
- if v4.empty() {
- v4, _, _ = ipState()
- }
- ifc, first := v4.first()
- return v4, ifc, first
-}
diff --git a/profiles/net/net_test.go b/profiles/net/net_test.go
deleted file mode 100644
index 31b88a8..0000000
--- a/profiles/net/net_test.go
+++ /dev/null
@@ -1,129 +0,0 @@
-package net
-
-import (
- "fmt"
- "net"
- "reflect"
- "testing"
-)
-
-// TODO(cnicolaou): more unit tests for:
-// Monitor/Publish network settings
-// PublicIPState
-// Empty
-// Has
-// First
-// etc.
-
-func exampleIPState() {
- ip4, ip6, err := ipState()
- if err != nil {
- panic("failed to find any network interfaces!")
- }
- for ifc, addrs := range ip4 {
- for _, addr := range addrs {
- fmt.Printf("ipv4: %s: %s", ifc, addr)
- }
- }
- for ifc, addrs := range ip6 {
- for _, addr := range addrs {
- fmt.Printf("ipv6: %s: %s", ifc, addr)
- }
- }
-}
-
-var (
- a = net.ParseIP("1.2.3.4")
- b = net.ParseIP("1.2.3.5")
- c = net.ParseIP("1.2.3.6")
- d = net.ParseIP("1.2.3.7")
- a6 = net.ParseIP("2001:4860:0:2001::68")
- b6 = net.ParseIP("2001:4860:0:2001::69")
- c6 = net.ParseIP("2001:4860:0:2001::70")
- d6 = net.ParseIP("2001:4860:0:2001::71")
-)
-
-func TestRemoved(t *testing.T) {
- aif := make(ipAndIf)
- bif := make(ipAndIf)
- aif["eth0"] = []net.IP{a, b, c, a6, b6, c6}
- aif["eth1"] = []net.IP{a, b, c, a6, b6, c6}
-
- // no changes.
- got, want := findRemoved(aif, aif), ipAndIf{}
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-
- // missing interfaces
- got, want = findRemoved(aif, bif), aif
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-
- // missing some addresses on a single interface
- bif["eth0"] = []net.IP{a, b, a6, b6}
- got, want = findRemoved(aif, bif), ipAndIf{
- "eth0": []net.IP{c, c6},
- "eth1": aif["eth1"],
- }
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-
- // change an address on a single interface
- bif["eth0"] = []net.IP{a, b, c, a6, b6, c6}
- bif["eth1"] = []net.IP{a, b, c, a6, b6, c6}
- bif["eth1"][2] = d
- got, want = findRemoved(aif, bif), ipAndIf{
- "eth1": []net.IP{c},
- }
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-
- // additions
- bif["eth0"] = []net.IP{a, b, c, a6, b6, c6, d6}
- bif["eth1"] = []net.IP{a, b, c, a6, b6, c6, d6}
- got, want = findRemoved(aif, bif), ipAndIf{}
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-}
-
-func TestAdded(t *testing.T) {
- aif := make(ipAndIf)
- bif := make(ipAndIf)
- aif["eth0"] = []net.IP{a, b, c, a6, b6, c6}
-
- // no changes.
- got, want := findAdded(aif, aif), ipAndIf{}
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-
- // added interface
- bif["eth0"] = []net.IP{a, b, c, a6, b6, c6}
- bif["eth1"] = []net.IP{a, b, c, a6, b6, c6}
- got, want = findAdded(aif, bif), ipAndIf{
- "eth1": []net.IP{a, b, c, a6, b6, c6}}
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-
- // added some address on a single interface
- bif["eth0"] = []net.IP{a, b, c, d, a6, b6, c6, d6}
- delete(bif, "eth1")
- got, want = findAdded(aif, bif), ipAndIf{
- "eth0": []net.IP{d, d6}}
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-
- // removals
- bif["eth0"] = []net.IP{a, b, c, b6}
- got, want = findAdded(aif, bif), ipAndIf{}
- if !reflect.DeepEqual(got, want) {
- t.Errorf("got %s, want %s", got, want)
- }
-}
diff --git a/profiles/net/net_watcher.go b/profiles/net/net_watcher.go
deleted file mode 100644
index cf99fc1..0000000
--- a/profiles/net/net_watcher.go
+++ /dev/null
@@ -1,30 +0,0 @@
-// +build ignore
-
-package main
-
-import (
- "fmt"
-
- "veyron2/config"
- "veyron2/rt"
-
- "veyron/profiles/net"
-)
-
-func main() {
- r := rt.Init()
- defer r.Cleanup()
-
- ch := make(chan config.Setting, 10)
- p := r.Publisher()
- settings, err := p.ForkStream(net.StreamName, ch)
- if err != nil {
- r.Logger().Infof("failed to fork stream: %s", err)
- }
- for _, setting := range settings.Latest {
- fmt.Println("Setting: ", setting)
- }
- for setting := range ch {
- fmt.Println("Setting: ", setting)
- }
-}
diff --git a/profiles/net/util.go b/profiles/net/util.go
deleted file mode 100644
index 57d42a5..0000000
--- a/profiles/net/util.go
+++ /dev/null
@@ -1,153 +0,0 @@
-package net
-
-import (
- "fmt"
- "net"
-
- "veyron/lib/netstate"
-)
-
-// TODO(cnicolaou): this will be removed in a subsequent CL by using the
-// new netstate library.
-
-// ipAndIf is a map of interface name to the set of IP addresses available on
-// that interface.
-type ipAndIf map[string][]net.IP
-
-func (ifcs ipAndIf) String() string {
- r := ""
- for k, v := range ifcs {
- r += fmt.Sprintf("(%v: %v)", k, v)
- }
- return r
-}
-
-// empty returns true if there are no addresses on any interfaces.
-func (ifcs ipAndIf) empty() bool {
- for _, addrs := range ifcs {
- if len(addrs) > 0 {
- return false
- }
- }
- return true
-}
-
-func (ifcs ipAndIf) first() (string, net.IP) {
- for ifc, addrs := range ifcs {
- if len(addrs) > 0 {
- return ifc, addrs[0]
- }
- }
- return "", nil
-}
-
-func (ifcs ipAndIf) has(ip net.IP) bool {
- for _, addrs := range ifcs {
- for _, a := range addrs {
- if a.Equal(ip) {
- return true
- }
- }
- }
- return false
-}
-
-// ipState returns the set of IPv4 and IPv6 addresses available to the running
-// process.
-func ipState() (ip4, ip6 ipAndIf, err error) {
- interfaces, err := net.Interfaces()
- if err != nil {
- return nil, nil, err
- }
- ip4 = make(ipAndIf)
- ip6 = make(ipAndIf)
- 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.IsLoopback() {
- continue
- }
- if t := ip.To4(); t != nil {
- ip4[ifc.Name] = append(ip4[ifc.Name], ip)
- }
- if t := ip.To16(); t != nil {
- ip6[ifc.Name] = append(ip6[ifc.Name], ip)
- }
- }
- }
- return ip4, ip6, nil
-}
-
-func filterPublic(ips ipAndIf) ipAndIf {
- public := make(ipAndIf)
- for ifc, addrs := range ips {
- for _, a := range addrs {
- if publicIP(a) {
- public[ifc] = append(public[ifc], a)
- }
- }
- }
- return public
-}
-
-// publicIPState returns the set of IPv4 and IPv6 addresses available
-// to the running process that are publicly/globally routable.
-func publicIPState() (ipAndIf, ipAndIf, error) {
- v4, v6, err := ipState()
- if err != nil {
- return nil, nil, err
- }
- return filterPublic(v4), filterPublic(v6), nil
-}
-
-// publicIP returns true if the supplied IP address is publicly/gobally
-// routable.
-func publicIP(ip net.IP) bool {
- if ip == nil {
- return false
- }
- return netstate.IsGloballyRoutableIP(ip)
-}
-
-func diffAB(a, b ipAndIf) ipAndIf {
- diff := make(ipAndIf)
- for ak, av := range a {
- if b[ak] == nil {
- diff[ak] = av
- continue
- }
- for _, v := range av {
- found := false
- for _, bv := range b[ak] {
- if v.Equal(bv) {
- found = true
- break
- }
- }
- if !found {
- diff[ak] = append(diff[ak], v)
- }
- }
- }
- return diff
-}
-
-// findAdded returns the set of interfaces and/or addresses that are
-// present in b, but not in a - i.e. have been added.
-func findAdded(a, b ipAndIf) ipAndIf {
- return diffAB(b, a)
-}
-
-// findRemoved returns the set of interfaces and/or addresses that
-// are present in a, but not in b - i.e. have been removed.
-func findRemoved(a, b ipAndIf) ipAndIf {
- return diffAB(a, b)
-}
diff --git a/profiles/net/gce_linux.go b/profiles/roaming/gce_linux.go
similarity index 64%
rename from profiles/net/gce_linux.go
rename to profiles/roaming/gce_linux.go
index d350880..93f5510 100644
--- a/profiles/net/gce_linux.go
+++ b/profiles/roaming/gce_linux.go
@@ -1,6 +1,6 @@
// +build linux
-package net
+package roaming
import (
"net"
@@ -11,28 +11,35 @@
"veyron/profiles/internal/gce"
)
-func handleGCE(rt veyron2.Runtime, publisher *config.Publisher) bool {
+func handleGCE(rt veyron2.Runtime, publisher *config.Publisher) *net.IPAddr {
log := rt.Logger()
if gce.RunningOnGCE() {
- var pub net.IP
+ var pub *net.IPAddr
// Determine the IP address from GCE's metadata
if ip, err := gce.ExternalIPAddress(); err != nil {
log.Infof("failed to query GCE metadata: %s", err)
} else {
// 1:1 NAT case, our network config will not change.
- pub = ip
+ pub = &net.IPAddr{IP: ip}
}
if pub == nil {
log.Infof("failed to determine public IP address to publish with")
+ return nil
}
ch := make(chan config.Setting)
defer close(ch)
- if _, err := publisher.CreateStream(StreamName, "network configuration", ch); err != nil {
- return false
+ stop, err := publisher.CreateStream(SettingsStreamName, "dhcp", ch)
+ if err != nil {
+ return nil
}
- publishInitialSettings(ch, listenProtocolFlag.Protocol, listenSpecFlag.String(), pub)
- return true
+
+ _ = stop
+ // TODO(cnicolaou): stop should be used by the soon to be added
+ // Cleanup method.
+
+ publishInitialSettings(ch, listenProtocolFlag.Protocol, listenSpecFlag.String(), []net.Addr{pub})
+ return pub
}
- return false
+ return nil
}
diff --git a/profiles/roaming/gce_other.go b/profiles/roaming/gce_other.go
new file mode 100644
index 0000000..b528aa2
--- /dev/null
+++ b/profiles/roaming/gce_other.go
@@ -0,0 +1,14 @@
+// +build !linux
+
+package roaming
+
+import (
+ "net"
+
+ "veyron2"
+ "veyron2/config"
+)
+
+func handleGCE(veyron2.Runtime, *config.Publisher) *net.IPAddr {
+ return nil
+}
diff --git a/profiles/roaming/init.go b/profiles/roaming/init.go
new file mode 100644
index 0000000..9196dd6
--- /dev/null
+++ b/profiles/roaming/init.go
@@ -0,0 +1,195 @@
+// +build linux darwin
+
+// Package roaming provides a network-aware Profile that provides appropriate
+// options and configuration for a variety of network configurations, including
+// being behind 1-1 NATs, using dhcp and auto-configuration for being on
+// Google Compute Engine.
+//
+// The config.Publisher mechanism is used for communicating networking
+// settings to the ipc.Server implementation of the runtime and publishes
+// the Settings it expects.
+package roaming
+
+import (
+ "flag"
+ "fmt"
+ "net"
+
+ "veyron2"
+ "veyron2/config"
+ "veyron2/ipc"
+ "veyron2/rt"
+
+ "veyron/lib/netconfig"
+ "veyron/lib/netstate"
+ "veyron/profiles"
+)
+
+const (
+ SettingsStreamName = "dhcp"
+)
+
+var (
+ listenProtocolFlag = config.TCPProtocolFlag{"tcp"}
+ listenSpecFlag = config.IPHostPortFlag{Port: "0"}
+)
+
+func init() {
+ flag.Var(&listenProtocolFlag, "veyron.tcp.protocol", "protocol to listen with")
+ flag.Var(&listenSpecFlag, "veyron.tcp.address", "address to listen on")
+ rt.RegisterProfile(New())
+}
+
+type profile struct {
+ addrChooser veyron2.AddressChooser
+ gce string
+}
+
+func preferredIPAddress(network string, addrs []net.Addr) (net.Addr, error) {
+ if !netstate.IsIPProtocol(network) {
+ return nil, fmt.Errorf("can't support network protocol %q", network)
+ }
+ al := netstate.AddrList(addrs).Map(netstate.ConvertToIPHost)
+ for _, predicate := range []netstate.Predicate{netstate.IsPublicUnicastIPv4,
+ netstate.IsUnicastIPv4, netstate.IsPublicUnicastIPv6} {
+ if a := al.First(predicate); a != nil {
+ return a, nil
+ }
+ }
+ return nil, fmt.Errorf("failed to find any usable address for %q", network)
+}
+
+func New() veyron2.Profile {
+ return &profile{addrChooser: preferredIPAddress}
+}
+
+func (p *profile) Platform() *veyron2.Platform {
+ platform, _ := profiles.Platform()
+ return platform
+}
+
+func (p *profile) Name() string {
+ return "dhcp" + p.gce
+}
+
+func (p *profile) Runtime() string {
+ return ""
+}
+
+func (p *profile) String() string {
+ return p.Name() + " " + p.Platform().String()
+}
+
+func (p *profile) Init(rt veyron2.Runtime, publisher *config.Publisher) {
+ log := rt.Logger()
+
+ state, err := netstate.GetAccessibleIPs()
+ if err != nil {
+ log.Infof("failed to determine network state")
+ // TODO(cnicolaou): in a subsequent CL, change Init to return an error.
+ return
+ //return err
+ }
+ first := state.First(netstate.IsUnicastIP)
+ if first == nil {
+ log.Infof("failed to find any usable IP addresses at startup")
+ }
+ public := netstate.IsPublicUnicastIPv4(first)
+
+ // We now know that there is an IP address to listen on, and whether
+ // it's public or private.
+
+ // Our address is private, so we test for running on GCE and for its
+ // 1:1 NAT configuration. handleGCE returns a non-nil addr
+ // if we are indeed running on GCE.
+ if !public {
+ if addr := handleGCE(rt, publisher); addr != nil {
+ p.addrChooser = func(string, []net.Addr) (net.Addr, error) {
+ return addr, nil
+ }
+ p.gce = "+gce"
+ return
+ }
+ }
+
+ // Create stream in Init function to avoid a race between any
+ // goroutines started here and consumers started after Init returns.
+ ch := make(chan config.Setting)
+ stop, err := publisher.CreateStream(SettingsStreamName, "dhcp", ch)
+ if err != nil {
+ log.Errorf("failed to create publisher: %s", err)
+ return
+ }
+
+ protocol := listenProtocolFlag.Protocol
+ log.VI(2).Infof("Initial Network Settings: %s %s available: %s", protocol, listenSpecFlag, state)
+ publishInitialSettings(ch, protocol, listenSpecFlag.String(), state)
+ go monitorNetworkSettings(rt, stop, ch, state, protocol, p.addrChooser)
+}
+
+func (p *profile) AddressChooser() veyron2.AddressChooser {
+ return p.addrChooser
+}
+
+func publishInitialSettings(ch chan<- config.Setting, protocol, listenSpec string, addrs []net.Addr) {
+ // TODO(cnicolaou): consider applying the address chooser here and not in
+ // the server, or better yet not sending the InitialAddrsSetting.
+ for _, setting := range []config.Setting{
+ ipc.NewProtocolSetting(protocol),
+ ipc.NewListenSpecSetting(listenSpecFlag),
+ ipc.NewInitialAddrsSetting(addrs),
+ } {
+ ch <- setting
+ }
+}
+
+// monitorNetworkSettings will monitor network configuration changes and
+// publish subsequent Settings to reflect any changes detected.
+func monitorNetworkSettings(rt veyron2.Runtime, stop <-chan struct{},
+ ch chan<- config.Setting, prev netstate.AddrList, protocol string, chooser veyron2.AddressChooser) {
+ defer close(ch)
+
+ log := rt.Logger()
+
+ // Start the dhcp watcher.
+ watcher, err := netconfig.NewNetConfigWatcher()
+ if err != nil {
+ log.VI(2).Infof("Failed to get new config watcher: %s", err)
+ // TODO(cnicolaou): add support for shutting down profiles
+ //<-stop
+ return
+ }
+
+ for {
+ select {
+ case <-watcher.Channel():
+ cur, err := netstate.GetAccessibleIPs()
+ if err != nil {
+ log.Errorf("failed to read network state: %s", err)
+ continue
+ }
+ removed := netstate.FindRemoved(prev, cur)
+ added := netstate.FindAdded(prev, cur)
+ log.VI(2).Infof("Previous: %d: %s", len(prev), prev)
+ log.VI(2).Infof("Current : %d: %s", len(cur), cur)
+ log.VI(2).Infof("Added : %d: %s", len(added), added)
+ log.VI(2).Infof("Removed : %d: %s", len(removed), removed)
+ if len(removed) == 0 && len(added) == 0 {
+ log.VI(2).Infof("Network event that lead to no address changes since our last 'baseline'")
+ continue
+ }
+ if len(removed) > 0 {
+ log.VI(2).Infof("Sending removed: %s", removed)
+ ch <- ipc.NewRmAddrsSetting(removed)
+ }
+ // We will always send the best currently available address
+ if chosen, err := chooser(protocol, cur); err == nil && chosen != nil {
+ ch <- ipc.NewAddAddrsSetting([]net.Addr{chosen})
+ }
+ prev = cur
+ // TODO(cnicolaou): add support for shutting down profiles.
+ //case <-stop:
+ // return
+ }
+ }
+}
diff --git a/profiles/roaming/net_watcher.go b/profiles/roaming/net_watcher.go
new file mode 100644
index 0000000..66f8d35
--- /dev/null
+++ b/profiles/roaming/net_watcher.go
@@ -0,0 +1,37 @@
+// +build ignore
+
+package main
+
+import (
+ "fmt"
+
+ "veyron2/config"
+ "veyron2/rt"
+
+ "veyron/profiles/roaming"
+)
+
+func main() {
+ r := rt.Init()
+ defer r.Cleanup()
+
+ fmt.Println("Profile: ", r.Profile().Name())
+
+ if addrOpt := r.Profile().PreferredAddressOpt(); addrOpt != nil {
+ if gce, err := addrOpt("", nil); err == nil {
+ fmt.Printf("%s: 1:1 NAT address is %s\n", r.Profile().Name(), gce)
+ }
+ }
+
+ ch := make(chan config.Setting, 10)
+ settings, err := r.Publisher().ForkStream(roaming.SettingsStreamName, ch)
+ if err != nil {
+ r.Logger().Infof("failed to fork stream: %s", err)
+ }
+ for _, setting := range settings.Latest {
+ fmt.Println("Setting: ", setting)
+ }
+ for setting := range ch {
+ fmt.Println("Setting: ", setting)
+ }
+}
diff --git a/profiles/roaming/print_addrs.go b/profiles/roaming/print_addrs.go
new file mode 100644
index 0000000..a6260bd
--- /dev/null
+++ b/profiles/roaming/print_addrs.go
@@ -0,0 +1,19 @@
+// +build ignore
+
+package main
+
+import (
+ "fmt"
+ "veyron/lib/netstate"
+)
+
+func main() {
+ al, err := netstate.GetAll()
+ if err != nil {
+ fmt.Printf("error getting networking state: %s", err)
+ return
+ }
+ for _, a := range al {
+ fmt.Println(a)
+ }
+}
diff --git a/profiles/roaming/roaming_server.go b/profiles/roaming/roaming_server.go
new file mode 100644
index 0000000..bc29b17
--- /dev/null
+++ b/profiles/roaming/roaming_server.go
@@ -0,0 +1,44 @@
+// +build ignore
+
+package main
+
+import (
+ "fmt"
+
+ "veyron2/ipc"
+ "veyron2/rt"
+
+ "veyron/profiles/roaming"
+)
+
+func main() {
+ r := rt.Init()
+ defer r.Cleanup()
+ log := r.Logger()
+
+ server, err := r.NewServer()
+ defer server.Stop()
+ if err != nil {
+ log.Fatalf("unexpected error: %q", err)
+ }
+
+ ep, err := server.RoamingListen(r.Publisher(), roaming.SettingsStreamName)
+ if err != nil {
+ log.Fatalf("unexpected error: %q", err)
+ }
+ if ep != nil {
+ fmt.Println(ep)
+ }
+ if err := server.Serve("roamer", ipc.LeafDispatcher(&dispatcher{}, nil)); err != nil {
+ log.Fatalf("unexpected error: %q", err)
+ }
+
+ done := make(chan struct{})
+ <-done
+}
+
+type dispatcher struct{}
+
+func (d *dispatcher) Echo(call ipc.ServerCall, arg string) (string, error) {
+ return arg, nil
+}
diff --git a/runtimes/google/ipc/full_test.go b/runtimes/google/ipc/full_test.go
index 8f3f127..2ac1be0 100644
--- a/runtimes/google/ipc/full_test.go
+++ b/runtimes/google/ipc/full_test.go
@@ -455,18 +455,18 @@
nameErr = "does not match the provided pattern"
)
var (
- // TODO(ataly, ashankar): Uncomment the following once server authorization
- // is enabled.
- // now = time.Now()
- // cavOnlyV1 = caveat.UniversalCaveat(caveat.PeerIdentity{"client/v1"})
- // cavExpired = security.ServiceCaveat{
- // Service: security.AllPrincipals,
- // Caveat: &caveat.Expiry{IssueTime: now, ExpiryTime: now},
- // }
- // clientV1ID = derive(clientID, "v1")
- // clientV2ID = derive(clientID, "v2")
- // serverV1ID = derive(serverID, "v1", cavOnlyV1)
- // serverExpiredID = derive(serverID, "expired", cavExpired)
+ // TODO(ataly, ashankar): Uncomment the following once server authorization
+ // is enabled.
+ // now = time.Now()
+ // cavOnlyV1 = caveat.UniversalCaveat(caveat.PeerIdentity{"client/v1"})
+ // cavExpired = security.ServiceCaveat{
+ // Service: security.AllPrincipals,
+ // Caveat: &caveat.Expiry{IssueTime: now, ExpiryTime: now},
+ // }
+ // clientV1ID = derive(clientID, "v1")
+ // clientV2ID = derive(clientID, "v2")
+ // serverV1ID = derive(serverID, "v1", cavOnlyV1)
+ // serverExpiredID = derive(serverID, "expired", cavExpired)
)
tests := []struct {
@@ -1106,7 +1106,7 @@
a.IP = net.ParseIP("1.1.1.1")
return a, nil
}
- server, err := InternalNewServer(testContext(), sm, ns, vc.FixedLocalID(serverID), veyron2.PreferredAddressOpt(pa))
+ server, err := InternalNewServer(testContext(), sm, ns, vc.FixedLocalID(serverID), &veyron2.AddressChooserOpt{pa})
if err != nil {
t.Errorf("InternalNewServer failed: %v", err)
}
@@ -1139,7 +1139,7 @@
paerr := func(string, []net.Addr) (net.Addr, error) {
return nil, fmt.Errorf("oops")
}
- server, err := InternalNewServer(testContext(), sm, ns, vc.FixedLocalID(serverID), veyron2.PreferredAddressOpt(paerr))
+ server, err := InternalNewServer(testContext(), sm, ns, vc.FixedLocalID(serverID), &veyron2.AddressChooserOpt{paerr})
if err != nil {
t.Errorf("InternalNewServer failed: %v", err)
}
diff --git a/runtimes/google/ipc/roaming_test.go b/runtimes/google/ipc/roaming_test.go
new file mode 100644
index 0000000..57437bd
--- /dev/null
+++ b/runtimes/google/ipc/roaming_test.go
@@ -0,0 +1,25 @@
+package ipc_test
+
+/*
+func startRoamingServer() {
+ mgr := imanager.InternalNew(naming.FixedRoutingID(0x1111111))
+ ns := newNamespace()
+}
+
+func TestRoamingListen(t *testing.T) {
+ r, err := rt.New()
+ if err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ publisher := config.CreateStream("test")
+
+ server, err := r.NewServer()
+ defer server.Stop()
+
+ if err != nil {
+ t.Errorf("unexpected error: %s", err)
+ }
+ server.RoamingListen(r.Publisher())
+ server.Serve(nil)
+}
+*/
diff --git a/runtimes/google/ipc/server.go b/runtimes/google/ipc/server.go
index 3457328..cb2cece 100644
--- a/runtimes/google/ipc/server.go
+++ b/runtimes/google/ipc/server.go
@@ -10,13 +10,13 @@
"time"
"veyron/lib/netstate"
-
"veyron/runtimes/google/lib/publisher"
inaming "veyron/runtimes/google/naming"
isecurity "veyron/runtimes/google/security"
vsecurity "veyron/security"
"veyron2"
+ "veyron2/config"
"veyron2/context"
"veyron2/ipc"
"veyron2/ipc/stream"
@@ -37,36 +37,48 @@
type server struct {
sync.Mutex
- ctx context.T // context used by the server to make internal RPCs.
- streamMgr stream.Manager // stream manager to listen for new flows.
- publisher publisher.Publisher // publisher to publish mounttable mounts.
- listenerOpts []stream.ListenerOpt // listener opts passed to Listen.
- listeners map[stream.Listener]bool // listeners created by Listen.
- disp ipc.Dispatcher // dispatcher to serve RPCs
- active sync.WaitGroup // active goroutines we've spawned.
- stopped bool // whether the server has been stopped.
- stoppedChan chan struct{} // closed when the server has been stopped.
+ ctx context.T // context used by the server to make internal RPCs.
+ streamMgr stream.Manager // stream manager to listen for new flows.
+ publisher publisher.Publisher // publisher to publish mounttable mounts.
+ listenerOpts []stream.ListenerOpt // listener opts passed to Listen.
+ listeners map[stream.Listener]*dhcpListener // listeners created by Listen.
+ disp ipc.Dispatcher // dispatcher to serve RPCs
+ active sync.WaitGroup // active goroutines we've spawned.
+ stopped bool // whether the server has been stopped.
+ stoppedChan chan struct{} // closed when the server has been stopped.
ns naming.Namespace
- preferredAddress func(network string, addrs []net.Addr) (net.Addr, error)
+ addressChooser veyron2.AddressChooser
servesMountTable bool
- stats *ipcStats // stats for this server.
+ roamingOpt veyron2.RoamingPublisherOpt
+ // TODO(cnicolaou): add roaming stats to ipcStats
+ stats *ipcStats // stats for this server.
+}
+
+type dhcpListener struct {
+ sync.Mutex
+ publisher *config.Publisher // publisher used to fork the stream
+ name string // name of the publisher stream
+ ep *inaming.Endpoint // endpoint returned after listening and choosing an address to be published
+ port string
+ ch chan config.Setting // channel to receive settings over
}
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{}),
- preferredAddress: preferredIPAddress,
- ns: ns,
- stats: newIPCStats(naming.Join("ipc", "server", streamMgr.RoutingID().String())),
+ ctx: ctx,
+ streamMgr: streamMgr,
+ publisher: publisher.New(ctx, ns, publishPeriod),
+ listeners: make(map[stream.Listener]*dhcpListener),
+ stoppedChan: make(chan struct{}),
+ ns: ns,
+ stats: newIPCStats(naming.Join("ipc", "server", streamMgr.RoutingID().String())),
}
for _, opt := range opts {
switch opt := opt.(type) {
- case veyron2.PreferredAddressOpt:
- s.preferredAddress = opt
+ case *veyron2.AddressChooserOpt:
+ s.addressChooser = opt.AddressChooser
+ case *veyron2.RoamingPublisherOpt:
+ s.roamingOpt = *opt
case stream.ListenerOpt:
// Collect all ServerOpts that are also ListenerOpts.
s.listenerOpts = append(s.listenerOpts, opt)
@@ -113,23 +125,6 @@
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, addrs []net.Addr) (net.Addr, error) {
- if !netstate.IsIPNetwork(network) {
- return nil, fmt.Errorf("can't support network %q", network)
- }
- al := netstate.AddrList(addrs)
- for _, predicate := range []netstate.Predicate{netstate.IsPublicUnicastIPv4,
- netstate.IsUnicastIPv4, netstate.IsPublicUnicastIPv6} {
- if a := al.First(predicate); a != nil {
- return a, 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
@@ -169,18 +164,21 @@
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.
- addrs, err := netstate.GetAccessibleIPs()
- if err == nil {
- if a, err := s.preferredAddress(protocol, addrs); err == nil {
- if ip := netstate.AsIP(a); ip != nil {
- // a may be an IPNet or an IPAddr under the covers,
- // but we really want the IP portion without any
- // netmask so we use AsIP to ensure that.
- iep.Address = net.JoinHostPort(ip.String(), port)
+ if ip.IsUnspecified() {
+ if s.addressChooser != nil {
+ // Need to find a usable IP address.
+ if addrs, err := netstate.GetAccessibleIPs(); err == nil {
+ if a, err := s.addressChooser(protocol, addrs); err == nil {
+ if ip := netstate.AsIP(a); ip != nil {
+ // a may be an IPNet or an IPAddr under the covers,
+ // but we really want the IP portion without any
+ // netmask so we use AsIP to ensure that.
+ iep.Address = net.JoinHostPort(ip.String(), port)
+ }
}
}
+ } else {
+ vlog.Errorf("no address chooser specified")
}
}
}
@@ -193,7 +191,7 @@
ln.Close()
return nil, errServerStopped
}
- s.listeners[ln] = true
+ s.listeners[ln] = nil
// We have a single goroutine per listener to accept new flows.
// Each flow is served from its own goroutine.
s.active.Add(1)
@@ -213,6 +211,143 @@
return ep, nil
}
+// externalEndpoint examines the endpoint returned by the stream listen call
+// and fills in the address to publish to the mount table. It also returns the
+// IP host address that it selected for publishing to the mount table.
+func (s *server) externalEndpoint(lep naming.Endpoint) (*inaming.Endpoint, *net.IPAddr, error) {
+ // We know the endpoint format, so we crack it open...
+ iep, ok := lep.(*inaming.Endpoint)
+ if !ok {
+ return nil, nil, fmt.Errorf("failed translating internal endpoint data types")
+ }
+
+ switch iep.Protocol {
+ case "tcp", "tcp4", "tcp6":
+ host, port, err := net.SplitHostPort(iep.Address)
+ if err != nil {
+ return nil, nil, err
+ }
+ ip := net.ParseIP(host)
+ if ip == nil {
+ return nil, nil, fmt.Errorf("failed to parse %q as an IP host", host)
+ }
+ if ip.IsUnspecified() && s.addressChooser != nil {
+ // Need to find a usable IP address since the call to listen
+ // didn't specify one.
+ addrs, err := netstate.GetAccessibleIPs()
+ if err == nil {
+ if a, err := s.addressChooser(iep.Protocol, addrs); err == nil {
+ iep.Address = net.JoinHostPort(a.String(), port)
+ return iep, a.(*net.IPAddr), nil
+ }
+ }
+ } else {
+ // Listen used a fixed IP address, which essentially disables
+ // roaming.
+ return iep, nil, nil
+ }
+ }
+ return iep, nil, nil
+}
+
+func (s *server) RoamingListen() (naming.Endpoint, error) {
+ s.Lock()
+ // Shortcut if the server is stopped, to avoid needlessly creating a
+ // listener.
+ if s.stopped {
+ s.Unlock()
+ return nil, errServerStopped
+ }
+ s.Unlock()
+
+ publisher := s.roamingOpt.Publisher
+ streamName := s.roamingOpt.StreamName
+
+ ch := make(chan config.Setting)
+ configStream, err := publisher.ForkStream(streamName, ch)
+ if err != nil {
+ return nil, fmt.Errorf("failed to fork stream %q: %s", streamName, err)
+ }
+ setting := configStream.Latest[ipc.ProtocolSetting]
+ if setting == nil {
+ return nil, fmt.Errorf("protocol setting has not be sent")
+ }
+ protocol, ok := setting.Value().(string)
+ if !ok {
+ return nil, fmt.Errorf("protocol setting is of the wrong type %T", setting.Value())
+ }
+
+ setting = configStream.Latest[ipc.ListenSpecSetting]
+ if setting == nil {
+ return nil, fmt.Errorf("listen spec setting has not be sent")
+ }
+ listenSpec, ok := setting.Value().(config.IPHostPortFlag)
+ if !ok {
+ return nil, fmt.Errorf("listen spec setting is of the wrong type %T", setting.Value())
+ }
+
+ address := listenSpec.String()
+
+ isNotLoopback := func(a net.Addr) bool {
+ if ip := netstate.AsIP(a); ip != nil {
+ return !ip.IsLoopback()
+ }
+ return true
+ }
+
+ al := netstate.FromIPAddr(listenSpec.IP)
+ if len(al) > 0 && al.First(isNotLoopback) == nil {
+ // All our addresses are loopback addresses
+ // TODO(cnicolaou): use Listen for now, but should refactor more completely.
+ return s.Listen(protocol, listenSpec.String())
+ }
+
+ ln, lep, err := s.streamMgr.Listen(protocol, address, s.listenerOpts...)
+ if err != nil {
+ vlog.Errorf("ipc: Listen on %v %v failed: %v", protocol, address, err)
+ return nil, err
+ }
+ ep, ipaddr, err := s.externalEndpoint(lep)
+ if ipaddr == nil || err != nil {
+ ln.Close()
+ if ipaddr == nil {
+ return nil, fmt.Errorf("the address %q requested for listening contained a fixed IP address which disables roaming, use :0 instead", address)
+ }
+ return nil, err
+ }
+
+ s.Lock()
+ if s.stopped {
+ s.Unlock()
+ // Ignore error return since we can't really do much about it.
+ ln.Close()
+ return nil, errServerStopped
+ }
+ _, port, _ := net.SplitHostPort(ep.Address)
+ dhcpl := &dhcpListener{ep: ep, port: port, ch: ch, name: streamName, publisher: publisher}
+ s.listeners[ln] = dhcpl
+ // We have a goroutine per listener to accept new flows and
+ // a goroutine to listen for address changes.
+ // Each flow is served from its own goroutine.
+ s.active.Add(2)
+
+ // goroutine to listen for connections
+ go func(ln stream.Listener, ep naming.Endpoint) {
+ s.listenLoop(ln, ep)
+ s.active.Done()
+ }(ln, ep)
+
+ // goroutine to listen for address changes.
+ go func(dl *dhcpListener) {
+ s.dhcpLoop(dl)
+ s.active.Done()
+ }(dhcpl)
+
+ s.Unlock()
+ s.publisher.AddServer(s.publishEP(ep))
+ return ep, nil
+}
+
func (s *server) publishEP(ep naming.Endpoint) string {
var name string
if !s.servesMountTable {
@@ -260,7 +395,7 @@
// (3) reconnected, publish new address
s.publisher.AddServer(s.publishEP(ep))
s.Lock()
- s.listeners[ln] = true
+ s.listeners[ln] = nil
s.Unlock()
}
}
@@ -292,6 +427,48 @@
}
}
+func (s *server) applyChange(dhcpl *dhcpListener, addrs []net.Addr, fn func(string)) {
+ dhcpl.Lock()
+ defer dhcpl.Unlock()
+ for _, a := range addrs {
+ if ip := netstate.AsIP(a); ip != nil {
+ dhcpl.ep.Address = net.JoinHostPort(ip.String(), dhcpl.port)
+ fn(s.publishEP(dhcpl.ep))
+ }
+ }
+}
+
+func (s *server) dhcpLoop(dhcpl *dhcpListener) {
+ defer vlog.VI(1).Infof("ipc: Stopped listen for dhcp changes on %v", dhcpl.ep)
+ vlog.VI(2).Infof("ipc: dhcp loop")
+ for setting := range dhcpl.ch {
+ if setting == nil {
+ return
+ }
+ switch v := setting.Value().(type) {
+ case bool:
+ return
+ case []net.Addr:
+ s.Lock()
+ if s.stopped {
+ s.Unlock()
+ return
+ }
+ publisher := s.publisher
+ s.Unlock()
+ switch setting.Name() {
+ case ipc.NewAddrsSetting:
+ vlog.Infof("Added some addresses: %q", v)
+ s.applyChange(dhcpl, v, publisher.AddServer)
+ case ipc.RmAddrsSetting:
+ vlog.Infof("Removed some addresses: %q", v)
+ s.applyChange(dhcpl, v, publisher.RemoveServer)
+ }
+
+ }
+ }
+}
+
func (s *server) Serve(name string, disp ipc.Dispatcher) error {
s.Lock()
defer s.Unlock()
@@ -339,10 +516,16 @@
// flows will continue until they terminate naturally.
nListeners := len(s.listeners)
errCh := make(chan error, nListeners)
- for ln, _ := range s.listeners {
+ for ln, dhcpl := range s.listeners {
go func(ln stream.Listener) {
errCh <- ln.Close()
}(ln)
+ if dhcpl != nil {
+ dhcpl.Lock()
+ dhcpl.publisher.CloseFork(dhcpl.name, dhcpl.ch)
+ dhcpl.ch <- config.NewBool("EOF", "stop", true)
+ dhcpl.Unlock()
+ }
}
s.Unlock()
var firstErr error
diff --git a/runtimes/google/lib/publisher/publisher.go b/runtimes/google/lib/publisher/publisher.go
index 77eab43..97e2176 100644
--- a/runtimes/google/lib/publisher/publisher.go
+++ b/runtimes/google/lib/publisher/publisher.go
@@ -137,7 +137,7 @@
}
func (p *publisher) runLoop(ctx context.T, ns naming.Namespace, period time.Duration) {
- vlog.VI(1).Info("ipc pub: start runLoop")
+ vlog.VI(2).Info("ipc pub: start runLoop")
state := newPubState(ctx, ns, period)
for {
select {
@@ -146,7 +146,7 @@
// Closing the cmdchan signals us to break out of the loop. Unmount
// everything and signal that we're done by closing the donechan.
state.unmountAll()
- vlog.VI(1).Info("ipc pub: exit runLoop")
+ vlog.VI(2).Info("ipc pub: exit runLoop")
close(p.donechan)
return
}
diff --git a/runtimes/google/rt/ipc.go b/runtimes/google/rt/ipc.go
index ccd1319..f3884f0 100644
--- a/runtimes/google/rt/ipc.go
+++ b/runtimes/google/rt/ipc.go
@@ -127,20 +127,30 @@
ns := rt.ns
var id security.PublicID
var otherOpts []ipc.ServerOpt
+ addressChooserOpt := &veyron2.AddressChooserOpt{rt.profile.AddressChooser()}
+ roamingOpt := &veyron2.RoamingPublisherOpt{rt.publisher, "roaming"}
for _, opt := range opts {
switch topt := opt.(type) {
case veyron2.NamespaceOpt:
ns = topt
case veyron2.LocalIDOpt:
id = topt.PublicID
+ case *veyron2.AddressChooserOpt:
+ addressChooserOpt = topt
+ case *veyron2.RoamingPublisherOpt:
+ roamingOpt = topt
default:
otherOpts = append(otherOpts, opt)
}
}
// Add the option that provides the local identity to the server.
otherOpts = append(otherOpts, rt.newLocalID(id))
+ // Add the preferredAddr and roaming opts
+ otherOpts = append(otherOpts, addressChooserOpt)
+ otherOpts = append(otherOpts, roamingOpt)
ctx := rt.NewContext()
+
return iipc.InternalNewServer(ctx, sm, ns, otherOpts...)
}
