diff --git a/probe/resolver.go b/probe/resolver.go index 1d99e41c0..58119497a 100644 --- a/probe/resolver.go +++ b/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{}{} } diff --git a/probe/resolver_test.go b/probe/resolver_test.go index 88e5807aa..4463e8f22 100644 --- a/probe/resolver_test.go +++ b/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 { diff --git a/xfer/http_publisher.go b/xfer/http_publisher.go index bb9aa3f57..8ff6e2f13 100644 --- a/xfer/http_publisher.go +++ b/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 } diff --git a/xfer/multi_publisher.go b/xfer/multi_publisher.go index 88d1455aa..532525600 100644 --- a/xfer/multi_publisher.go +++ b/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{}{} } diff --git a/xfer/multi_publisher_internal_test.go b/xfer/multi_publisher_internal_test.go new file mode 100644 index 000000000..4400b1b2e --- /dev/null +++ b/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) + } +}