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