Move concurrency from resolve to HTTP GET

- Process DNS resolution serially
- Process up to 10 HTTP GET (for app ID) concurrently

More than 10 concurrent GET requests will block on the semaphore. This
will cause the staticResolver.resolve method to block, which is probably
fine: it will just delay the next resolve loop, currently at 1m
intervals.

To make this a little bit more robust, I've also added a fastClient for
app ID resolution, with a timeout (total, including connect, request,
and response) of 5s.

probe/resolver.go

@@ -15,12 +15,9 @@ var (
 	lookupIP = net.LookupIP
 )
 
-const maxConcurrentLookup = 10
-
 type staticResolver struct {
 	set     func(string, []string)
 	targets []target
-	sema    semaphore
 	quit    chan struct{}
 }
 
@@ -35,7 +32,6 @@ func newStaticResolver(targets []string, set func(target string, endpoints []str
 	r := staticResolver{
 		targets: prepare(targets),
 		set:     set,
-		sema:    newSemaphore(maxConcurrentLookup),
 		quit:    make(chan struct{}),
 	}
 	go r.loop()
@@ -79,17 +75,15 @@ func prepare(strs []string) []target {
 }
 
 func (r staticResolver) resolve() {
-	for t, endpoints := range resolveMany(r.sema, r.targets) {
+	for t, endpoints := range resolveMany(r.targets) {
 		r.set(t.String(), endpoints)
 	}
 }
 
-func resolveMany(s semaphore, targets []target) map[target][]string {
+func resolveMany(targets []target) map[target][]string {
 	result := map[target][]string{}
 	for _, t := range targets {
-		c := make(chan []string)
-		go func(t target) { s.p(); defer s.v(); c <- resolveOne(t) }(t)
-		result[t] = <-c
+		result[t] = resolveOne(t)
 	}
 	return result
 }
@@ -115,15 +109,3 @@ func resolveOne(t target) []string {
 	}
 	return endpoints
 }
-
-type semaphore chan struct{}
-
-func newSemaphore(n int) semaphore {
-	c := make(chan struct{}, n)
-	for i := 0; i < n; i++ {
-		c <- struct{}{}
-	}
-	return semaphore(c)
-}
-func (s semaphore) p() { <-s }
-func (s semaphore) v() { s <- struct{}{} }

probe/resolver_test.go

@@ -96,40 +96,6 @@ func TestResolver(t *testing.T) {
 	}
 }
 
-func TestSemaphore(t *testing.T) {
-	n := 3
-	s := newSemaphore(n)
-
-	// First n should be fine
-	for i := 0; i < n; i++ {
-		ok := make(chan struct{})
-		go func() { s.p(); close(ok) }()
-		select {
-		case <-ok:
-		case <-time.After(10 * time.Millisecond):
-			t.Errorf("p (%d) failed", i+1)
-		}
-	}
-
-	// This should block
-	ok := make(chan struct{})
-	go func() { s.p(); close(ok) }()
-	select {
-	case <-ok:
-		t.Errorf("%dth p OK, but should block", n+1)
-	case <-time.After(10 * time.Millisecond):
-		//t.Logf("%dth p blocks, as expected", n+1)
-	}
-
-	s.v()
-
-	select {
-	case <-ok:
-	case <-time.After(10 * time.Millisecond):
-		t.Errorf("%dth p didn't resolve in time", n+1)
-	}
-}
-
 func makeIPs(addrs ...string) []net.IP {
 	var ips []net.IP
 	for _, addr := range addrs {

xfer/http_publisher.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"time"
 
 	"github.com/weaveworks/scope/common/sanitize"
 )
@@ -16,10 +17,14 @@ type HTTPPublisher struct {
 	probeID string
 }
 
+var fastClient = http.Client{
+	Timeout: 5 * time.Second,
+}
+
 // NewHTTPPublisher returns an HTTPPublisher ready for use.
 func NewHTTPPublisher(target, token, probeID string) (string, *HTTPPublisher, error) {
 	targetAPI := sanitize.URL("http://", 0, "/api")(target)
-	resp, err := http.Get(targetAPI)
+	resp, err := fastClient.Get(targetAPI)
 	if err != nil {
 		return "", nil, err
 	}

xfer/multi_publisher.go

@@ -15,6 +15,7 @@ import (
 type MultiPublisher struct {
 	mtx     sync.Mutex
 	factory func(endpoint string) (string, Publisher, error)
+	sema    semaphore
 	list    []tuple
 }
 
@@ -22,6 +23,7 @@ type MultiPublisher struct {
 func NewMultiPublisher(factory func(endpoint string) (string, Publisher, error)) *MultiPublisher {
 	return &MultiPublisher{
 		factory: factory,
+		sema:    newSemaphore(maxConcurrentGET),
 	}
 }
 
@@ -30,8 +32,11 @@ type tuple struct {
 	target    string // DNS name
 	endpoint  string // IP addr
 	id        string // unique ID from app
+	err       error  // if factory failed
 }
 
+const maxConcurrentGET = 10
+
 // Set declares that the target (DNS name) resolves to the provided endpoints
 // (IPs), and that we want to publish to each of those endpoints. Set replaces
 // any existing publishers to the given target. Set invokes the factory method
@@ -39,14 +44,23 @@ type tuple struct {
 // unique ID.
 func (p *MultiPublisher) Set(target string, endpoints []string) {
 	// Convert endpoints to publishers.
-	list := make([]tuple, 0, len(p.list)+len(endpoints))
+	c := make(chan tuple, len(endpoints))
 	for _, endpoint := range endpoints {
-		id, publisher, err := p.factory(endpoint)
-		if err != nil {
-			log.Printf("multi-publisher set: %s (%s): %v", target, endpoint, err)
+		go func(endpoint string) {
+			p.sema.p()
+			defer p.sema.v()
+			id, publisher, err := p.factory(endpoint)
+			c <- tuple{publisher, target, endpoint, id, err}
+		}(endpoint)
+	}
+	list := make([]tuple, 0, len(p.list)+len(endpoints))
+	for i := 0; i < cap(c); i++ {
+		t := <-c
+		if t.err != nil {
+			log.Printf("multi-publisher set: %s (%s): %v", t.target, t.endpoint, t.err)
 			continue
 		}
-		list = append(list, tuple{publisher, target, endpoint, id})
+		list = append(list, t)
 	}
 
 	// Copy all other tuples over to the new list.
@@ -115,3 +129,15 @@ func (p *MultiPublisher) appendFilter(list []tuple, f func(tuple) bool) []tuple
 	}
 	return list
 }
+
+type semaphore chan struct{}
+
+func newSemaphore(n int) semaphore {
+	c := make(chan struct{}, n)
+	for i := 0; i < n; i++ {
+		c <- struct{}{}
+	}
+	return semaphore(c)
+}
+func (s semaphore) p() { <-s }
+func (s semaphore) v() { s <- struct{}{} }

xfer/multi_publisher_internal_test.go

@@ -0,0 +1,40 @@
+package xfer
+
+import (
+	"testing"
+	"time"
+)
+
+func TestSemaphore(t *testing.T) {
+	n := 3
+	s := newSemaphore(n)
+
+	// First n should be fine
+	for i := 0; i < n; i++ {
+		ok := make(chan struct{})
+		go func() { s.p(); close(ok) }()
+		select {
+		case <-ok:
+		case <-time.After(10 * time.Millisecond):
+			t.Errorf("p (%d) failed", i+1)
+		}
+	}
+
+	// This should block
+	ok := make(chan struct{})
+	go func() { s.p(); close(ok) }()
+	select {
+	case <-ok:
+		t.Errorf("%dth p OK, but should block", n+1)
+	case <-time.After(10 * time.Millisecond):
+		//t.Logf("%dth p blocks, as expected", n+1)
+	}
+
+	s.v()
+
+	select {
+	case <-ok:
+	case <-time.After(10 * time.Millisecond):
+		t.Errorf("%dth p didn't resolve in time", n+1)
+	}
+}