veyron/runtimes/google/ipc: Refactor endpoint sorting.
The primary motivation for this change is to have
filterAndOrderEndpoints work on []naming.MountedServer instead
of []string. This is to ease a future change I plan to make where
naming.MountedServer will also include the BlessingPatterns that the
legitimate server should match and the IPC client will validate servers
it connects to against the pattern in the MountedServer.
Change-Id: I45833157be824542d23e79c39ce147c759140254
diff --git a/runtimes/google/ipc/client.go b/runtimes/google/ipc/client.go
index 7c59e5d..2462cf0 100644
--- a/runtimes/google/ipc/client.go
+++ b/runtimes/google/ipc/client.go
@@ -200,29 +200,6 @@
return flow, nil
}
-// connectFlow parses an endpoint and a suffix out of the server and establishes
-// a flow to the endpoint, returning the parsed suffix.
-// The server name passed in should be a rooted name, of the form "/ep/suffix" or
-// "/ep//suffix", or just "/ep".
-func (c *client) connectFlow(ctx *context.T, server string, vcOpts []stream.VCOpt) (stream.Flow, string, verror.E) {
- address, suffix := naming.SplitAddressName(server)
- if len(address) == 0 {
- return nil, "", verror.Make(errNonRootedName, ctx, server)
- }
- ep, err := inaming.NewEndpoint(address)
- if err != nil {
- return nil, "", verror.Make(errInvalidEndpoint, ctx, address)
- }
- if err = version.CheckCompatibility(ep); err != nil {
- return nil, "", verror.Make(errIncompatibleEndpoint, ctx, ep)
- }
- flow, verr := c.createFlow(ctx, ep, vcOpts)
- if verr != nil {
- return nil, "", verr
- }
- return flow, suffix, nil
-}
-
// A randomized exponential backoff. The randomness deters error convoys from forming.
// TODO(cnicolaou): rationalize this and the backoff in ipc.Server. Note
// that rand is not thread safe and may crash.
@@ -373,32 +350,49 @@
err verror.E
}
+// tryCreateFlow attempts to establish a Flow to "server" (which must be a
+// rooted name), over which a method invocation request could be sent.
// TODO(cnicolaou): implement real, configurable load balancing.
-func (c *client) tryServer(ctx *context.T, index int, server string, ch chan<- *serverStatus, vcOpts []stream.VCOpt) {
+func (c *client) tryCreateFlow(ctx *context.T, index int, server string, ch chan<- *serverStatus, vcOpts []stream.VCOpt) {
status := &serverStatus{index: index}
- var err verror.E
var span vtrace.Span
- ctx, span = vtrace.SetNewSpan(ctx, "<client>connectFlow")
+ ctx, span = vtrace.SetNewSpan(ctx, "<client>tryCreateFlow")
span.Annotatef("address:%v", server)
- defer span.Finish()
- if status.flow, status.suffix, err = c.connectFlow(ctx, server, vcOpts); err != nil {
- vlog.VI(2).Infof("ipc: connect to %s: %s", server, err)
- status.err = err
- status.flow = nil
+ defer func() {
+ ch <- status
+ span.Finish()
+ }()
+ address, suffix := naming.SplitAddressName(server)
+ if len(address) == 0 {
+ status.err = verror.Make(errNonRootedName, ctx, server)
+ return
}
- ch <- status
+ ep, err := inaming.NewEndpoint(address)
+ if err != nil {
+ status.err = verror.Make(errInvalidEndpoint, ctx, address)
+ return
+ }
+ if err = version.CheckCompatibility(ep); err != nil {
+ status.err = verror.Make(errIncompatibleEndpoint, ctx, ep)
+ return
+ }
+ if status.flow, status.err = c.createFlow(ctx, ep, vcOpts); status.err != nil {
+ vlog.VI(2).Infof("ipc: connect to %v: %v", server, status.err)
+ return
+ }
+ status.suffix = suffix
+ return
}
-// tryCall makes a single attempt at a call, against possibly multiple servers.
+// tryCall makes a single attempt at a call. It may connect to multiple servers
+// (all that serve "name"), but will invoke the method on at most one of them
+// (the server running on the most preferred protcol and network amongst all
+// the servers that were successfully connected to and authorized).
func (c *client) tryCall(ctx *context.T, name, method string, args []interface{}, opts []ipc.CallOpt) (ipc.Call, verror.ActionCode, verror.E) {
- // Get CallOpts that are also VCOpts.
- vcOpts := getVCOpts(opts)
- // Get CallOpts that are also ResolveOpts.
- resolveOpts := getResolveOpts(opts)
- var servers []string
+ var resolved *naming.MountEntry
var pattern security.BlessingPattern
-
- if resolved, err := c.ns.Resolve(ctx, name, resolveOpts...); err != nil {
+ var err error
+ if resolved, err = c.ns.Resolve(ctx, name, getResolveOpts(opts)...); err != nil {
vlog.Errorf("Resolve: %v", err)
if verror.Is(err, naming.ErrNoSuchName.ID) {
return nil, verror.RetryRefetch, verror.Make(verror.NoServers, ctx, name)
@@ -410,36 +404,27 @@
return nil, verror.RetryRefetch, verror.Make(verror.NoServers, ctx, name)
}
// An empty set of protocols means all protocols...
- ordered, err := filterAndOrderServers(resolved.Names(), c.preferredProtocols)
- if err != nil {
+ if resolved.Servers, err = filterAndOrderServers(resolved.Servers, c.preferredProtocols); err != nil {
return nil, verror.RetryRefetch, verror.Make(verror.NoServers, ctx, name, err)
- } else if len(ordered) == 0 {
- // sooo annoying....
- r := []interface{}{err}
- r = append(r, name)
- for _, s := range resolved.Servers {
- r = append(r, s)
- }
- return nil, verror.RetryRefetch, verror.Make(verror.NoServers, ctx, r)
}
- servers = ordered
}
// servers is now orderd by the priority heurestic implemented in
// filterAndOrderServers.
- attempts := len(servers)
-
- // Try to connect to all servers in parallel. Provide sufficient buffering
- // for all of the connections to finish instantaneously. This is important
- // because we want to process the responses in priority order; that order is
- // indicated by the order of entries in servers. So, if two respones come in
- // at the same 'instant', we prefer the first in the slice.
+ //
+ // Try to connect to all servers in parallel. Provide sufficient
+ // buffering for all of the connections to finish instantaneously. This
+ // is important because we want to process the responses in priority
+ // order; that order is indicated by the order of entries in servers.
+ // So, if two respones come in at the same 'instant', we prefer the
+ // first in the resolved.Servers)
+ attempts := len(resolved.Servers)
responses := make([]*serverStatus, attempts)
ch := make(chan *serverStatus, attempts)
- vcOpts = append(vcOpts, c.vcOpts...)
+ vcOpts := append(getVCOpts(opts), c.vcOpts...)
vcOpts = append(vcOpts, vc.DialContext{ctx})
- for i, server := range servers {
- go c.tryServer(ctx, i, server, ch, vcOpts)
+ for i, server := range resolved.Names() {
+ go c.tryCreateFlow(ctx, i, server, ch, vcOpts)
}
delay := time.Duration(ipc.NoTimeout)
@@ -465,7 +450,7 @@
}
case <-timeoutChan:
vlog.VI(2).Infof("ipc: timeout on connection to server %v ", name)
- _, _, err := c.failedTryCall(ctx, name, method, servers, responses, ch)
+ _, _, err := c.failedTryCall(ctx, name, method, responses, ch)
if !verror.Is(err, verror.Timeout.ID) {
return nil, verror.NoRetry, verror.Make(verror.Timeout, ctx, err)
}
@@ -495,8 +480,7 @@
// Validate caveats on the server's identity for the context associated with this call.
var err error
if serverB, grantedB, err = c.authorizeServer(ctx, r.flow, name, method, pattern, opts); err != nil {
- r.err = verror.Make(errNotTrusted, ctx,
- name, r.flow.RemoteBlessings(), err)
+ r.err = verror.Make(errNotTrusted, ctx, name, r.flow.RemoteBlessings(), err)
vlog.VI(2).Infof("ipc: err: %s", r.err)
r.flow.Close()
r.flow = nil
@@ -548,12 +532,12 @@
return fc, verror.NoRetry, nil
}
if numResponses == len(responses) {
- return c.failedTryCall(ctx, name, method, servers, responses, ch)
+ return c.failedTryCall(ctx, name, method, responses, ch)
}
}
}
-// cleanupTryCall ensures we've waited for every response from the tryServer
+// cleanupTryCall ensures we've waited for every response from the tryCreateFlow
// goroutines, and have closed the flow from each one except skip. This is a
// blocking function; it should be called in its own goroutine.
func cleanupTryCall(skip *serverStatus, responses []*serverStatus, ch chan *serverStatus) {
@@ -582,13 +566,12 @@
// failedTryCall performs asynchronous cleanup for tryCall, and returns an
// appropriate error from the responses we've already received. All parallel
// calls in tryCall failed or we timed out if we get here.
-func (c *client) failedTryCall(ctx *context.T, name, method string, servers []string, responses []*serverStatus, ch chan *serverStatus) (ipc.Call, verror.ActionCode, verror.E) {
+func (c *client) failedTryCall(ctx *context.T, name, method string, responses []*serverStatus, ch chan *serverStatus) (ipc.Call, verror.ActionCode, verror.E) {
go cleanupTryCall(nil, responses, ch)
c.ns.FlushCacheEntry(name)
noconn, untrusted := []string{}, []string{}
- for i, r := range responses {
+ for _, r := range responses {
if r != nil && r.err != nil {
- vlog.VI(2).Infof("Server: %s: %s", servers[i], r.err)
switch {
case verror.Is(r.err, errNotTrusted.ID) || verror.Is(r.err, errAuthError.ID):
untrusted = append(untrusted, "("+r.err.Error()+") ")
diff --git a/runtimes/google/ipc/server.go b/runtimes/google/ipc/server.go
index 5a473d8..20af110 100644
--- a/runtimes/google/ipc/server.go
+++ b/runtimes/google/ipc/server.go
@@ -151,29 +151,29 @@
// resolveToEndpoint resolves an object name or address to an endpoint.
func (s *server) resolveToEndpoint(address string) (string, error) {
- var names []string
+ var resolved *naming.MountEntry
+ var err error
if s.ns != nil {
- var entry *naming.MountEntry
- var err error
- if entry, err = s.ns.Resolve(s.ctx, address); err != nil {
+ if resolved, err = s.ns.Resolve(s.ctx, address); err != nil {
return "", err
}
- names = entry.Names()
} else {
- names = append(names, address)
+ // Fake a namespace resolution
+ resolved = &naming.MountEntry{Servers: []naming.MountedServer{
+ {Server: address},
+ }}
}
// An empty set of protocols means all protocols...
- ordered, err := filterAndOrderServers(names, s.preferredProtocols)
- if err != nil {
+ if resolved.Servers, err = filterAndOrderServers(resolved.Servers, s.preferredProtocols); err != nil {
return "", err
}
- for _, n := range ordered {
+ for _, n := range resolved.Names() {
address, suffix := naming.SplitAddressName(n)
if suffix != "" {
continue
}
- if _, err := inaming.NewEndpoint(address); err == nil {
- return address, nil
+ if ep, err := inaming.NewEndpoint(address); err == nil {
+ return ep.String(), nil
}
}
return "", fmt.Errorf("unable to resolve %q to an endpoint", address)
diff --git a/runtimes/google/ipc/sort_endpoints.go b/runtimes/google/ipc/sort_endpoints.go
index 58a6d57..93b8af0 100644
--- a/runtimes/google/ipc/sort_endpoints.go
+++ b/runtimes/google/ipc/sort_endpoints.go
@@ -3,7 +3,7 @@
import (
"fmt"
"net"
- "strings"
+ "sort"
"v.io/core/veyron2/naming"
"v.io/core/veyron2/vlog"
@@ -33,148 +33,51 @@
return r
}
-// TODO(cnicolaou): simplify this code, especially the use of maps+slices
-// and special cases.
-
func newErrorAccumulator() *errorAccumulator {
return &errorAccumulator{errs: make([]error, 0, 4)}
}
-type serverEndpoint struct {
- iep *inaming.Endpoint
- suffix string
+type serverLocality int
+
+const (
+ unknownNetwork serverLocality = iota
+ remoteNetwork
+ localNetwork
+)
+
+type sortableServer struct {
+ server naming.MountedServer
+ protocolRank int // larger values are preferred.
+ locality serverLocality // larger values are preferred.
}
-func (se *serverEndpoint) String() string {
- return fmt.Sprintf("(%s, %q)", se.iep, se.suffix)
+func (s *sortableServer) String() string {
+ return fmt.Sprintf("%v", s.server)
}
-func filterCompatibleEndpoints(errs *errorAccumulator, servers []string) []*serverEndpoint {
- se := make([]*serverEndpoint, 0, len(servers))
- for _, server := range servers {
- name := server
- address, suffix := naming.SplitAddressName(name)
- if len(address) == 0 {
- // Maybe it's not a rooted endpoint, just a bare one.
- address = name
- suffix = ""
- }
- iep, err := inaming.NewEndpoint(address)
- if err != nil {
- errs.add(fmt.Errorf("failed to parse %q: %s", name, err))
- continue
- }
- if err = version.CheckCompatibility(iep); err != nil {
- errs.add(fmt.Errorf("%q: %s", name, err))
- continue
- }
- sep := &serverEndpoint{iep, suffix}
- se = append(se, sep)
+type sortableServerList []sortableServer
+
+func (l sortableServerList) Len() int { return len(l) }
+func (l sortableServerList) Swap(i, j int) { l[i], l[j] = l[j], l[i] }
+func (l sortableServerList) Less(i, j int) bool {
+ if l[i].protocolRank == l[j].protocolRank {
+ return l[i].locality > l[j].locality
}
- return se
+ return l[i].protocolRank > l[j].protocolRank
}
-// sortByProtocols sorts the supplied slice of serverEndpoints into a hash
-// map keyed by the protocol of each of those endpoints where that protocol
-// is listed in the protocols parameter and keyed by '*' if not so listed.
-func sortByProtocol(eps []*serverEndpoint, protocols []string) (bool, map[string][]*serverEndpoint) {
- byProtocol := make(map[string][]*serverEndpoint)
- matched := false
- for _, ep := range eps {
- if ep.iep.Protocol == naming.UnknownProtocol {
- byProtocol[naming.UnknownProtocol] = append(byProtocol[naming.UnknownProtocol], ep)
- matched = true
- break
- }
- found := false
- for _, p := range protocols {
- if ep.iep.Protocol == p ||
- (p == "tcp" && strings.HasPrefix(ep.iep.Protocol, "tcp")) ||
- (p == "wsh" && strings.HasPrefix(ep.iep.Protocol, "wsh")) ||
- // Can't use strings.HasPrefix below because of "wsh" has the prefix "ws" but is a different protocol.
- (p == "ws" && (ep.iep.Protocol == "ws4" || ep.iep.Protocol == "ws6")) {
- byProtocol[p] = append(byProtocol[p], ep)
- found = true
- matched = true
- break
- }
- }
- if !found {
- byProtocol["*"] = append(byProtocol["*"], ep)
- }
+func mkProtocolRankMap(list []string) map[string]int {
+ if len(list) == 0 {
+ return nil
}
- return matched, byProtocol
+ m := make(map[string]int)
+ for idx, protocol := range list {
+ m[protocol] = len(list) - idx
+ }
+ return m
}
-func orderByLocality(ifcs netstate.AddrList, eps []*serverEndpoint) []*serverEndpoint {
- if len(ifcs) <= 1 {
- return append([]*serverEndpoint{}, eps...)
- }
- ipnets := make([]*net.IPNet, 0, len(ifcs))
- for _, a := range ifcs {
- // Try IP
- _, ipnet, err := net.ParseCIDR(a.Address().String())
- if err != nil {
- continue
- }
- ipnets = append(ipnets, ipnet)
- }
- if len(ipnets) == 0 {
- return eps
- }
- // TODO(cnicolaou): this can obviously be made more efficient...
- local := make([]*serverEndpoint, 0, len(eps))
- remote := make([]*serverEndpoint, 0, len(eps))
- notip := make([]*serverEndpoint, 0, len(eps))
- for _, ep := range eps {
- if strings.HasPrefix(ep.iep.Protocol, "tcp") || strings.HasPrefix(ep.iep.Protocol, "ws") {
- // Take care to use the Address directly, since the network
- // may be marked as a 'websocket'. This throws out any thought
- // of dealing with IPv6 etc and web sockets.
- host, _, err := net.SplitHostPort(ep.iep.Address)
- if err != nil {
- host = ep.iep.Address
- }
- ip := net.ParseIP(host)
- if ip == nil {
- notip = append(notip, ep)
- continue
- }
- found := false
- for _, ipnet := range ipnets {
- if ipnet.Contains(ip) {
- local = append(local, ep)
- found = true
- break
- }
- }
- if !found {
- remote = append(remote, ep)
- }
- } else {
- notip = append(notip, ep)
- }
- }
- return append(local, append(remote, notip...)...)
-}
-
-func slice(eps []*serverEndpoint) []string {
- r := make([]string, len(eps))
- for i, a := range eps {
- r[i] = naming.JoinAddressName(a.iep.String(), a.suffix)
- }
- return r
-}
-
-func sliceByProtocol(eps map[string][]*serverEndpoint, protocols []string) []string {
- r := make([]string, 0, 10)
- for _, p := range protocols {
- r = append(r, slice(eps[p])...)
- }
- return r
-}
-
-var defaultPreferredProtocolOrder = []string{"unixfd", "tcp4", "tcp", "*"}
+var defaultPreferredProtocolOrder = mkProtocolRankMap([]string{"unixfd", "wsh", "tcp4", "tcp", "*"})
// filterAndOrderServers returns a set of servers that are compatible with
// the current client in order of 'preference' specified by the supplied
@@ -189,49 +92,128 @@
// will be used, but unlike the previous case, any servers that don't support
// these protocols will be returned also, but following the default
// preferences.
-func filterAndOrderServers(servers []string, protocols []string) ([]string, error) {
- errs := newErrorAccumulator()
- vlog.VI(3).Infof("Candidates[%v]: %v", protocols, servers)
-
- // TODO(cnicolaou): ideally we should filter out unsupported protocols
- // here - e.g. no point dialing on a ws protocol if it's not registered
- // etc.
- compatible := filterCompatibleEndpoints(errs, servers)
- if len(compatible) == 0 {
- return nil, fmt.Errorf("failed to find any compatible servers: %s", errs)
+func filterAndOrderServers(servers []naming.MountedServer, protocols []string) ([]naming.MountedServer, error) {
+ vlog.VI(3).Infof("filterAndOrderServers%v: %v", protocols, servers)
+ var (
+ errs = newErrorAccumulator()
+ list = make(sortableServerList, 0, len(servers))
+ protoRanks = mkProtocolRankMap(protocols)
+ ipnets = ipNetworks()
+ )
+ if len(protoRanks) == 0 {
+ protoRanks = defaultPreferredProtocolOrder
}
- vlog.VI(3).Infof("Version Compatible: %v", compatible)
-
- defaultOrdering := len(protocols) == 0
- preferredProtocolOrder := defaultPreferredProtocolOrder
- if !defaultOrdering {
- preferredProtocolOrder = protocols
+ for _, server := range servers {
+ name := server.Server
+ ep, err := name2endpoint(name)
+ if err != nil {
+ errs.add(fmt.Errorf("malformed endpoint %q: %v", name, err))
+ continue
+ }
+ if err = version.CheckCompatibility(ep); err != nil {
+ errs.add(fmt.Errorf("%q: %v", name, err))
+ continue
+ }
+ rank, err := protocol2rank(ep.Addr().Network(), protoRanks)
+ if err != nil {
+ errs.add(fmt.Errorf("%q: %v", name, err))
+ continue
+ }
+ list = append(list, sortableServer{
+ server: server,
+ protocolRank: rank,
+ locality: locality(ep, ipnets),
+ })
}
-
- // Add unknown protocol to the order last.
- preferredProtocolOrder = append(preferredProtocolOrder, naming.UnknownProtocol)
-
- // put the server endpoints into per-protocol lists
- matched, byProtocol := sortByProtocol(compatible, preferredProtocolOrder)
- if !defaultOrdering && !matched {
- return nil, fmt.Errorf("failed to find any servers compatible with %v from %s", protocols, servers)
+ if len(list) == 0 {
+ return nil, fmt.Errorf("failed to find any compatible servers: %v", errs)
}
+ // TODO(ashankar): Don't have to use stable sorting, could
+ // just use sort.Sort. The only problem with that is the
+ // unittest.
+ sort.Stable(list)
+ // Convert to []naming.MountedServer
+ ret := make([]naming.MountedServer, len(list))
+ for idx, item := range list {
+ ret[idx] = item.server
+ }
+ return ret, nil
+}
- vlog.VI(3).Infof("Have Protocols(%v): %v", protocols, byProtocol)
+// name2endpoint returns the naming.Endpoint encoded in a name.
+func name2endpoint(name string) (naming.Endpoint, error) {
+ addr := name
+ if naming.Rooted(name) {
+ addr, _ = naming.SplitAddressName(name)
+ }
+ return inaming.NewEndpoint(addr)
+}
- networks, err := netstate.GetAll()
+// protocol2rank returns the "rank" of a protocol (given a map of ranks).
+// The higher the rank, the more preferable the protocol.
+func protocol2rank(protocol string, ranks map[string]int) (int, error) {
+ if r, ok := ranks[protocol]; ok {
+ return r, nil
+ }
+ // Special case: if "wsh" has a rank but "wsh4"/"wsh6" don't,
+ // then they get the same rank as "wsh". Similar for "tcp" and "ws".
+ if p := protocol; p == "wsh4" || p == "wsh6" || p == "tcp4" || p == "tcp6" || p == "ws4" || p == "ws6" {
+ if r, ok := ranks[p[:len(p)-1]]; ok {
+ return r, nil
+ }
+ }
+ // "*" means that any protocol is acceptable.
+ if r, ok := ranks["*"]; ok {
+ return r, nil
+ }
+ // UnknownProtocol should be rare, it typically happens when
+ // the endpoint is described in <host>:<port> format instead of
+ // the full fidelity description (@<version>@<protocol>@...).
+ if protocol == naming.UnknownProtocol {
+ return -1, nil
+ }
+ return 0, fmt.Errorf("undesired protocol %q", protocol)
+}
+
+// ipNetworks returns the IP networks on this machine.
+func ipNetworks() []*net.IPNet {
+ ifcs, err := netstate.GetAll()
if err != nil {
- // return whatever we have now, just not sorted by locality.
- return sliceByProtocol(byProtocol, preferredProtocolOrder), nil
+ vlog.VI(5).Infof("netstate.GetAll failed: %v", err)
+ return nil
}
-
- ordered := make([]*serverEndpoint, 0, len(byProtocol))
- for _, protocol := range preferredProtocolOrder {
- o := orderByLocality(networks, byProtocol[protocol])
- vlog.VI(3).Infof("Protocol: %q ordered by locality: %v", protocol, o)
- ordered = append(ordered, o...)
+ ret := make([]*net.IPNet, 0, len(ifcs))
+ for _, a := range ifcs {
+ _, ipnet, err := net.ParseCIDR(a.Address().String())
+ if err != nil {
+ vlog.VI(5).Infof("net.ParseCIDR(%q) failed: %v", a.Address(), err)
+ continue
+ }
+ ret = append(ret, ipnet)
}
+ return ret
+}
- vlog.VI(2).Infof("Ordered By Locality: %v", ordered)
- return slice(ordered), nil
+// locality returns the serverLocality to use given an endpoint and the
+// set of IP networks configured on this machine.
+func locality(ep naming.Endpoint, networks []*net.IPNet) serverLocality {
+ if len(networks) < 1 {
+ return unknownNetwork // 0 IP networks, locality doesn't matter.
+
+ }
+ host, _, err := net.SplitHostPort(ep.Addr().String())
+ if err != nil {
+ host = ep.Addr().String()
+ }
+ ip := net.ParseIP(host)
+ if ip == nil {
+ // Not an IP address (possibly not an IP network).
+ return unknownNetwork
+ }
+ for _, ipnet := range networks {
+ if ipnet.Contains(ip) {
+ return localNetwork
+ }
+ }
+ return remoteNetwork
}
diff --git a/runtimes/google/ipc/sort_internal_test.go b/runtimes/google/ipc/sort_internal_test.go
index 20b9663..041cacd 100644
--- a/runtimes/google/ipc/sort_internal_test.go
+++ b/runtimes/google/ipc/sort_internal_test.go
@@ -7,11 +7,15 @@
"v.io/core/veyron2/ipc/version"
"v.io/core/veyron2/naming"
- "v.io/core/veyron2/vlog"
)
+func servers2names(servers []naming.MountedServer) []string {
+ e := naming.MountEntry{Servers: servers}
+ return e.Names()
+}
+
func TestIncompatible(t *testing.T) {
- servers := []string{}
+ servers := []naming.MountedServer{}
_, err := filterAndOrderServers(servers, []string{"tcp"})
if err == nil || err.Error() != "failed to find any compatible servers: " {
@@ -21,64 +25,68 @@
for _, a := range []string{"127.0.0.1", "127.0.0.2"} {
addr := naming.FormatEndpoint("tcp", a, version.IPCVersionRange{100, 200})
name := naming.JoinAddressName(addr, "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
_, err = filterAndOrderServers(servers, []string{"tcp"})
if err == nil || (!strings.HasPrefix(err.Error(), "failed to find any compatible servers:") && !strings.Contains(err.Error(), "No compatible IPC versions available")) {
- vlog.Infof("A: %t . %t", strings.HasPrefix(err.Error(), "failed to find any compatible servers:"), !strings.Contains(err.Error(), "No compatible IPC versions available"))
t.Errorf("expected a different error to: %v", err)
}
for _, a := range []string{"127.0.0.3", "127.0.0.4"} {
name := naming.JoinAddressName(naming.FormatEndpoint("tcp", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
_, err = filterAndOrderServers(servers, []string{"foobar"})
- if err == nil || !strings.HasPrefix(err.Error(), "failed to find any servers compatible with [foobar] ") {
+ if err == nil || !strings.HasSuffix(err.Error(), "undesired protocol \"tcp\")") {
t.Errorf("expected a different error to: %v", err)
}
}
func TestOrderingByProtocol(t *testing.T) {
- servers := []string{}
+ servers := []naming.MountedServer{}
for _, a := range []string{"127.0.0.3", "127.0.0.4"} {
name := naming.JoinAddressName(naming.FormatEndpoint("tcp", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
for _, a := range []string{"127.0.0.1", "127.0.0.2"} {
name := naming.JoinAddressName(naming.FormatEndpoint("tcp4", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
for _, a := range []string{"127.0.0.10", "127.0.0.11"} {
name := naming.JoinAddressName(naming.FormatEndpoint("foobar", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
for _, a := range []string{"127.0.0.7", "127.0.0.8"} {
name := naming.JoinAddressName(naming.FormatEndpoint("tcp6", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
-
- got, err := filterAndOrderServers(servers, []string{"batman"})
- if err == nil {
+ if _, err := filterAndOrderServers(servers, []string{"batman"}); err == nil {
t.Fatalf("expected an error")
}
- got, err = filterAndOrderServers(servers, []string{"foobar", "tcp4"})
- if err != nil {
- t.Fatalf("unexpected error: %s", err)
- }
+ // Add a server with naming.UnknownProtocol. This typically happens
+ // when the endpoint is in <host>:<port> format. Currently, the sorting
+ // is setup to always allow UnknownProtocol, but put it in the end.
+ // May want to revisit this choice, but for now the test captures what
+ // the current state of the code intends.
+ servers = append(servers, naming.MountedServer{Server: "127.0.0.12:14141"})
// Just foobar and tcp4
want := []string{
- "/@3@foobar@127.0.0.10@00000000000000000000000000000000@@@m@@",
- "/@3@foobar@127.0.0.11@00000000000000000000000000000000@@@m@@",
- "/@3@tcp4@127.0.0.1@00000000000000000000000000000000@@@m@@",
- "/@3@tcp4@127.0.0.2@00000000000000000000000000000000@@@m@@",
+ "/@2@foobar@127.0.0.10@@@@@",
+ "/@2@foobar@127.0.0.11@@@@@",
+ "/@2@tcp4@127.0.0.1@@@@@",
+ "/@2@tcp4@127.0.0.2@@@@@",
+ "/127.0.0.12:14141",
}
- if !reflect.DeepEqual(got, want) {
+ result, err := filterAndOrderServers(servers, []string{"foobar", "tcp4"})
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ if got := servers2names(result); !reflect.DeepEqual(got, want) {
t.Errorf("got: %v, want %v", got, want)
}
@@ -87,117 +95,119 @@
// The order will be the default preferred order for protocols, the
// original ordering within each protocol, with protocols that
// are not in the default ordering list at the end.
- got, err = filterAndOrderServers(servers, nil)
- if err != nil {
+ want = []string{
+ "/@2@tcp4@127.0.0.1@@@@@",
+ "/@2@tcp4@127.0.0.2@@@@@",
+ "/@2@tcp@127.0.0.3@@@@@",
+ "/@2@tcp@127.0.0.4@@@@@",
+ "/@2@tcp6@127.0.0.7@@@@@",
+ "/@2@tcp6@127.0.0.8@@@@@",
+ "/@2@foobar@127.0.0.10@@@@@",
+ "/@2@foobar@127.0.0.11@@@@@",
+ "/127.0.0.12:14141",
+ }
+ if result, err = filterAndOrderServers(servers, nil); err != nil {
t.Fatalf("unexpected error: %s", err)
}
-
- want = []string{
- "/@3@tcp4@127.0.0.1@00000000000000000000000000000000@@@m@@",
- "/@3@tcp4@127.0.0.2@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@127.0.0.3@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@127.0.0.4@00000000000000000000000000000000@@@m@@",
- "/@3@tcp6@127.0.0.7@00000000000000000000000000000000@@@m@@",
- "/@3@tcp6@127.0.0.8@00000000000000000000000000000000@@@m@@",
- "/@3@foobar@127.0.0.10@00000000000000000000000000000000@@@m@@",
- "/@3@foobar@127.0.0.11@00000000000000000000000000000000@@@m@@",
- }
- if !reflect.DeepEqual(got, want) {
+ if got := servers2names(result); !reflect.DeepEqual(got, want) {
t.Errorf("got: %v, want %v", got, want)
}
- got, err = filterAndOrderServers(servers, []string{})
- if err != nil {
+ if result, err = filterAndOrderServers(servers, []string{}); err != nil {
t.Fatalf("unexpected error: %s", err)
}
-
- if !reflect.DeepEqual(got, want) {
+ if got := servers2names(result); !reflect.DeepEqual(got, want) {
t.Errorf("got: %v, want %v", got, want)
}
- got, err = filterAndOrderServers(servers, []string{"tcp"})
- if err != nil {
+ // Just "tcp" implies tcp4 and tcp6 as well.
+ want = []string{
+ "/@2@tcp@127.0.0.3@@@@@",
+ "/@2@tcp@127.0.0.4@@@@@",
+ "/@2@tcp4@127.0.0.1@@@@@",
+ "/@2@tcp4@127.0.0.2@@@@@",
+ "/@2@tcp6@127.0.0.7@@@@@",
+ "/@2@tcp6@127.0.0.8@@@@@",
+ "/127.0.0.12:14141",
+ }
+ if result, err = filterAndOrderServers(servers, []string{"tcp"}); err != nil {
t.Fatalf("unexpected error: %s", err)
}
- // tcp or tcp4
- want = []string{
- "/@3@tcp4@127.0.0.1@00000000000000000000000000000000@@@m@@",
- "/@3@tcp4@127.0.0.2@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@127.0.0.3@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@127.0.0.4@00000000000000000000000000000000@@@m@@",
+ if got := servers2names(result); !reflect.DeepEqual(got, want) {
+ t.Errorf("got: %v, want %v", got, want)
}
// Ask for all protocols, with no ordering, except for locality
- servers = []string{}
+ want = []string{
+ "/@2@tcp@127.0.0.3@@@@@",
+ "/@2@tcp@127.0.0.1@@@@@",
+ "/@2@tcp@74.125.69.139@@@@@",
+ "/@2@tcp@192.168.1.10@@@@@",
+ "/@2@tcp@74.125.142.83@@@@@",
+ "/127.0.0.12:14141",
+ "/@2@foobar@127.0.0.10@@@@@",
+ "/@2@foobar@127.0.0.11@@@@@",
+ }
+ servers = []naming.MountedServer{}
+ // naming.UnknownProtocol
+ servers = append(servers, naming.MountedServer{Server: "127.0.0.12:14141"})
for _, a := range []string{"74.125.69.139", "127.0.0.3", "127.0.0.1", "192.168.1.10", "74.125.142.83"} {
name := naming.JoinAddressName(naming.FormatEndpoint("tcp", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
for _, a := range []string{"127.0.0.10", "127.0.0.11"} {
name := naming.JoinAddressName(naming.FormatEndpoint("foobar", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
- // Everything, since we didn't specify a protocol
- got, err = filterAndOrderServers(servers, []string{})
- if err != nil {
+ if result, err = filterAndOrderServers(servers, []string{}); err != nil {
t.Fatalf("unexpected error: %s", err)
}
- want = []string{
- "/@3@tcp@127.0.0.3@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@127.0.0.1@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@74.125.69.139@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@192.168.1.10@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@74.125.142.83@00000000000000000000000000000000@@@m@@",
- "/@3@foobar@127.0.0.10@00000000000000000000000000000000@@@m@@",
- "/@3@foobar@127.0.0.11@00000000000000000000000000000000@@@m@@",
- }
- if !reflect.DeepEqual(got, want) {
+ if got := servers2names(result); !reflect.DeepEqual(got, want) {
t.Errorf("got: %v, want %v", got, want)
}
}
-func TestOrderByNetwork(t *testing.T) {
- servers := []string{}
+func TestOrderingByLocality(t *testing.T) {
+ servers := []naming.MountedServer{}
for _, a := range []string{"74.125.69.139", "127.0.0.3", "127.0.0.1", "192.168.1.10", "74.125.142.83"} {
name := naming.JoinAddressName(naming.FormatEndpoint("tcp", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
- got, err := filterAndOrderServers(servers, []string{"tcp"})
+ result, err := filterAndOrderServers(servers, []string{"tcp"})
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
want := []string{
- "/@3@tcp@127.0.0.3@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@127.0.0.1@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@74.125.69.139@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@192.168.1.10@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@74.125.142.83@00000000000000000000000000000000@@@m@@",
+ "/@2@tcp@127.0.0.3@@@@@",
+ "/@2@tcp@127.0.0.1@@@@@",
+ "/@2@tcp@74.125.69.139@@@@@",
+ "/@2@tcp@192.168.1.10@@@@@",
+ "/@2@tcp@74.125.142.83@@@@@",
}
- if !reflect.DeepEqual(got, want) {
+ if got := servers2names(result); !reflect.DeepEqual(got, want) {
t.Errorf("got: %v, want %v", got, want)
}
for _, a := range []string{"74.125.69.139", "127.0.0.3:123", "127.0.0.1", "192.168.1.10", "74.125.142.83"} {
name := naming.JoinAddressName(naming.FormatEndpoint("ws", a), "")
- servers = append(servers, name)
+ servers = append(servers, naming.MountedServer{Server: name})
}
- got, err = filterAndOrderServers(servers, []string{"ws", "tcp"})
- if err != nil {
+ if result, err = filterAndOrderServers(servers, []string{"ws", "tcp"}); err != nil {
t.Fatalf("unexpected error: %s", err)
}
want = []string{
- "/@3@ws@127.0.0.3:123@00000000000000000000000000000000@@@m@@",
- "/@3@ws@127.0.0.1@00000000000000000000000000000000@@@m@@",
- "/@3@ws@74.125.69.139@00000000000000000000000000000000@@@m@@",
- "/@3@ws@192.168.1.10@00000000000000000000000000000000@@@m@@",
- "/@3@ws@74.125.142.83@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@127.0.0.3@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@127.0.0.1@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@74.125.69.139@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@192.168.1.10@00000000000000000000000000000000@@@m@@",
- "/@3@tcp@74.125.142.83@00000000000000000000000000000000@@@m@@",
+ "/@2@ws@127.0.0.3:123@@@@@",
+ "/@2@ws@127.0.0.1@@@@@",
+ "/@2@ws@74.125.69.139@@@@@",
+ "/@2@ws@192.168.1.10@@@@@",
+ "/@2@ws@74.125.142.83@@@@@",
+ "/@2@tcp@127.0.0.3@@@@@",
+ "/@2@tcp@127.0.0.1@@@@@",
+ "/@2@tcp@74.125.69.139@@@@@",
+ "/@2@tcp@192.168.1.10@@@@@",
+ "/@2@tcp@74.125.142.83@@@@@",
}
- if !reflect.DeepEqual(got, want) {
+ if got := servers2names(result); !reflect.DeepEqual(got, want) {
t.Errorf("got: %v, want %v", got, want)
}
}
