diff --git a/probe/main.go b/probe/main.go index 43b6dcea6..a2d91dcbe 100644 --- a/probe/main.go +++ b/probe/main.go @@ -80,16 +80,16 @@ func main() { log.Printf("warning: process reporting enabled, but that requires root to find everything") } - publisherFactory := func(target string) (xfer.Publisher, error) { - _, publisher, err := xfer.NewHTTPPublisher(target, *token, probeID) + factory := func(endpoint string) (string, xfer.Publisher, error) { + id, publisher, err := xfer.NewHTTPPublisher(endpoint, *token, probeID) if err != nil { - return nil, err + return "", nil, err } - return xfer.NewBackgroundPublisher(publisher), nil + return id, xfer.NewBackgroundPublisher(publisher), nil } - publishers := xfer.NewMultiPublisher(publisherFactory) + publishers := xfer.NewMultiPublisher(factory) defer publishers.Stop() - resolver := newStaticResolver(targets, publishers.Add) + resolver := newStaticResolver(targets, publishers.Set) defer resolver.Stop() addrs, err := net.InterfaceAddrs() diff --git a/probe/overlay/weave.go b/probe/overlay/weave.go index 1a278180f..ce1330523 100644 --- a/probe/overlay/weave.go +++ b/probe/overlay/weave.go @@ -10,9 +10,8 @@ import ( "strings" "sync" - "github.com/weaveworks/scope/common/sanitize" - "github.com/weaveworks/scope/common/exec" + "github.com/weaveworks/scope/common/sanitize" "github.com/weaveworks/scope/probe/docker" "github.com/weaveworks/scope/report" ) diff --git a/probe/resolver.go b/probe/resolver.go index 8b8c7e97e..6feae5d41 100644 --- a/probe/resolver.go +++ b/probe/resolver.go @@ -17,7 +17,7 @@ var ( type staticResolver struct { quit chan struct{} - add func(string) + set func(string, []string) peers []peer } @@ -31,10 +31,10 @@ type peer struct { // resolved IPs. It explictiy supports hostnames which // resolve to multiple IPs; it will repeatedly call // add with the same IP, expecting the target to dedupe. -func newStaticResolver(peers []string, add func(string)) staticResolver { +func newStaticResolver(peers []string, set func(target string, endpoints []string)) staticResolver { r := staticResolver{ quit: make(chan struct{}), - add: add, + set: set, peers: prepareNames(peers), } go r.loop() @@ -92,13 +92,15 @@ func (r staticResolver) resolveHosts() { } } + endpoints := make([]string, 0, len(addrs)) for _, addr := range addrs { // For now, ignore IPv6 if addr.To4() == nil { continue } - r.add(net.JoinHostPort(addr.String(), peer.port)) + endpoints = append(endpoints, net.JoinHostPort(addr.String(), peer.port)) } + r.set(peer.hostname, endpoints) } } diff --git a/probe/resolver_test.go b/probe/resolver_test.go index d1deada58..63826380c 100644 --- a/probe/resolver_test.go +++ b/probe/resolver_test.go @@ -39,15 +39,19 @@ func TestResolver(t *testing.T) { port := ":80" ip1 := "192.168.0.1" ip2 := "192.168.0.10" - adds := make(chan string) - add := func(s string) { adds <- s } + sets := make(chan string) + set := func(target string, endpoints []string) { + for _, endpoint := range endpoints { + sets <- endpoint + } + } - r := newStaticResolver([]string{"symbolic.name" + port, "namewithnoport", ip1 + port, ip2}, add) + r := newStaticResolver([]string{"symbolic.name" + port, "namewithnoport", ip1 + port, ip2}, set) assertAdd := func(want string) { _, _, line, _ := runtime.Caller(1) select { - case have := <-adds: + case have := <-sets: if want != have { t.Errorf("line %d: want %q, have %q", line, want, have) } diff --git a/xfer/background_publisher.go b/xfer/background_publisher.go new file mode 100644 index 000000000..ac511830e --- /dev/null +++ b/xfer/background_publisher.go @@ -0,0 +1,70 @@ +package xfer + +import ( + "bytes" + "log" + "time" +) + +const ( + initialBackoff = 1 * time.Second + maxBackoff = 60 * time.Second +) + +// BackgroundPublisher is a publisher which does the publish asynchronously. +// It will only do one publish at once; if there is an ongoing publish, +// concurrent publishes are dropped. +type BackgroundPublisher struct { + publisher Publisher + reports chan *bytes.Buffer + quit chan struct{} +} + +// NewBackgroundPublisher creates a new BackgroundPublisher with the given publisher +func NewBackgroundPublisher(p Publisher) *BackgroundPublisher { + result := &BackgroundPublisher{ + publisher: p, + reports: make(chan *bytes.Buffer), + quit: make(chan struct{}), + } + go result.loop() + return result +} + +func (b *BackgroundPublisher) loop() { + backoff := initialBackoff + + for r := range b.reports { + err := b.publisher.Publish(r) + if err == nil { + backoff = initialBackoff + continue + } + + log.Printf("Error publishing to %s, backing off %s: %v", b.publisher, backoff, err) + select { + case <-time.After(backoff): + case <-b.quit: + } + backoff = backoff * 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } +} + +// Publish implements Publisher +func (b *BackgroundPublisher) Publish(buf *bytes.Buffer) error { + select { + case b.reports <- buf: + default: + } + return nil +} + +// Stop implements Publisher +func (b *BackgroundPublisher) Stop() { + close(b.reports) + close(b.quit) + b.publisher.Stop() +} diff --git a/xfer/background_publisher_test.go b/xfer/background_publisher_test.go new file mode 100644 index 000000000..60a1a6283 --- /dev/null +++ b/xfer/background_publisher_test.go @@ -0,0 +1,7 @@ +package xfer_test + +import "testing" + +func TestBackgroundPublisher(t *testing.T) { + t.Skip("TODO") +} diff --git a/xfer/http_publisher.go b/xfer/http_publisher.go new file mode 100644 index 000000000..42eb3c70b --- /dev/null +++ b/xfer/http_publisher.go @@ -0,0 +1,80 @@ +package xfer + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + + "github.com/weaveworks/scope/common/sanitize" +) + +// HTTPPublisher publishes reports by POST to a fixed endpoint. +type HTTPPublisher struct { + url string + token string + probeID string +} + +// 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) + if err != nil { + return "", nil, err + } + defer resp.Body.Close() + var apiResponse struct { + ID string `json:"id"` + } + if err := json.NewDecoder(resp.Body).Decode(&apiResponse); err != nil { + return "", nil, err + } + return apiResponse.ID, &HTTPPublisher{ + url: sanitize.URL("http://", 0, "/api/report")(target), + token: token, + probeID: probeID, + }, nil +} + +func (p HTTPPublisher) String() string { + return p.url +} + +// Publish publishes the report to the URL. +func (p HTTPPublisher) Publish(buf *bytes.Buffer) error { + req, err := http.NewRequest("POST", p.url, buf) + if err != nil { + return err + } + req.Header.Set("Authorization", AuthorizationHeader(p.token)) + req.Header.Set(ScopeProbeIDHeader, p.probeID) + req.Header.Set("Content-Encoding", "gzip") + // req.Header.Set("Content-Type", "application/binary") // TODO: we should use http.DetectContentType(..) on the gob'ed + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf(resp.Status) + } + return nil +} + +// Stop implements Publisher +func (p HTTPPublisher) Stop() {} + +// AuthorizationHeader returns a value suitable for an HTTP Authorization +// header, based on the passed token string. +func AuthorizationHeader(token string) string { + return fmt.Sprintf("Scope-Probe token=%s", token) +} + +// ScopeProbeIDHeader is the header we use to carry the probe's unique ID. The +// ID is currently set to the probe's hostname. It's designed to deduplicate +// reports from the same probe to the same receiver, in case the probe is +// configured to publish to multiple receivers that resolve to the same app. +const ScopeProbeIDHeader = "X-Scope-Probe-ID" diff --git a/xfer/publisher_test.go b/xfer/http_publisher_test.go similarity index 67% rename from xfer/publisher_test.go rename to xfer/http_publisher_test.go index 787eee793..e0147a738 100644 --- a/xfer/publisher_test.go +++ b/xfer/http_publisher_test.go @@ -1,7 +1,6 @@ package xfer_test import ( - "bytes" "compress/gzip" "encoding/gob" "encoding/json" @@ -13,7 +12,6 @@ import ( "time" "github.com/gorilla/handlers" - "github.com/weaveworks/scope/report" "github.com/weaveworks/scope/test" "github.com/weaveworks/scope/xfer" @@ -82,32 +80,3 @@ func TestHTTPPublisher(t *testing.T) { t.Error("timeout") } } - -func TestMultiPublisher(t *testing.T) { - var ( - p = &mockPublisher{} - factory = func(string) (xfer.Publisher, error) { return p, nil } - multiPublisher = xfer.NewMultiPublisher(factory) - ) - - multiPublisher.Add("first") - if err := multiPublisher.Publish(&bytes.Buffer{}); err != nil { - t.Error(err) - } - if want, have := 1, p.count; want != have { - t.Errorf("want %d, have %d", want, have) - } - - multiPublisher.Add("second") // but factory returns same mockPublisher - if err := multiPublisher.Publish(&bytes.Buffer{}); err != nil { - t.Error(err) - } - if want, have := 3, p.count; want != have { - t.Errorf("want %d, have %d", want, have) - } -} - -type mockPublisher struct{ count int } - -func (p *mockPublisher) Publish(*bytes.Buffer) error { p.count++; return nil } -func (p *mockPublisher) Stop() {} diff --git a/xfer/multi_publisher.go b/xfer/multi_publisher.go new file mode 100644 index 000000000..eadf1ff90 --- /dev/null +++ b/xfer/multi_publisher.go @@ -0,0 +1,107 @@ +package xfer + +import ( + "bytes" + "errors" + "log" + "strings" + "sync" +) + +// MultiPublisher implements publisher over a collection of heterogeneous +// targets. See documentation of each method to understand the semantics. +type MultiPublisher struct { + mtx sync.Mutex + factory func(endpoint string) (string, Publisher, error) + list []tuple +} + +// NewMultiPublisher returns a new MultiPublisher ready for use. +func NewMultiPublisher(factory func(endpoint string) (string, Publisher, error)) *MultiPublisher { + return &MultiPublisher{ + factory: factory, + } +} + +type tuple struct { + publisher Publisher + target string // DNS name + endpoint string // IP addr + id string // unique ID from app +} + +// 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 +// to convert each endpoint to a publisher, and to get the remote receiver's +// unique ID. +func (p *MultiPublisher) Set(target string, endpoints []string) { + // Convert endpoints to publishers. + list := make([]tuple, 0, len(p.list)+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) + continue + } + list = append(list, tuple{publisher, target, endpoint, id}) + } + + // Copy all other tuples over to the new list. + p.mtx.Lock() + defer p.mtx.Unlock() + p.list = p.appendFilter(list, func(t tuple) bool { return t.target != target }) +} + +// Delete removes all endpoints that match the given target. +func (p *MultiPublisher) Delete(target string) { + p.mtx.Lock() + defer p.mtx.Unlock() + p.list = p.appendFilter([]tuple{}, func(t tuple) bool { return t.target != target }) +} + +// Publish implements Publisher by publishing the buffer to all of the +// underlying publishers sequentially. But, it will publish to one endpoint +// for each unique ID. Failed publishes don't count. +func (p *MultiPublisher) Publish(buf *bytes.Buffer) error { + var ( + ids = map[string]struct{}{} + errs = []string{} + ) + p.mtx.Lock() + defer p.mtx.Unlock() + for _, t := range p.list { + if _, ok := ids[t.id]; ok { + continue + } + if err := t.publisher.Publish(buf); err != nil { + errs = append(errs, err.Error()) + continue + } + ids[t.id] = struct{}{} // sent already + } + if len(errs) > 0 { + return errors.New(strings.Join(errs, "; ")) + } + return nil +} + +// Stop invokes stop on all underlying publishers and removes them. +func (p *MultiPublisher) Stop() { + p.mtx.Lock() + defer p.mtx.Unlock() + for _, t := range p.list { + t.publisher.Stop() + } + p.list = []tuple{} +} + +func (p *MultiPublisher) appendFilter(list []tuple, f func(tuple) bool) []tuple { + for _, t := range p.list { + if !f(t) { + continue + } + list = append(list, t) + } + return list +} diff --git a/xfer/multi_publisher_test.go b/xfer/multi_publisher_test.go new file mode 100644 index 000000000..0ec919bde --- /dev/null +++ b/xfer/multi_publisher_test.go @@ -0,0 +1,52 @@ +package xfer_test + +import ( + "bytes" + "fmt" + "testing" + + "github.com/weaveworks/scope/xfer" +) + +func TestMultiPublisher(t *testing.T) { + var ( + a1 = &mockPublisher{} // target a, endpoint 1 + a2 = &mockPublisher{} // target a, endpoint 2 (duplicate) + b2 = &mockPublisher{} // target b, endpoint 2 (duplicate) + b3 = &mockPublisher{} // target b, endpoint 3 + ) + + sum := func() int { return a1.count + a2.count + b2.count + b3.count } + + mp := xfer.NewMultiPublisher(func(endpoint string) (string, xfer.Publisher, error) { + switch endpoint { + case "a1": + return "1", a1, nil + case "a2": + return "2", a2, nil + case "b2": + return "2", b2, nil + case "b3": + return "3", b3, nil + default: + return "", nil, fmt.Errorf("invalid endpoint %s", endpoint) + } + }) + + mp.Set("a", []string{"a1", "a2"}) + mp.Set("b", []string{"b2", "b3"}) + + for i := 1; i < 10; i++ { + if err := mp.Publish(&bytes.Buffer{}); err != nil { + t.Error(err) + } + if want, have := 3*i, sum(); want != have { + t.Errorf("want %d, have %d", want, have) + } + } +} + +type mockPublisher struct{ count int } + +func (p *mockPublisher) Publish(*bytes.Buffer) error { p.count++; return nil } +func (p *mockPublisher) Stop() {} diff --git a/xfer/publisher.go b/xfer/publisher.go index 0a0df7504..10a7cd960 100644 --- a/xfer/publisher.go +++ b/xfer/publisher.go @@ -1,22 +1,6 @@ package xfer -import ( - "bytes" - "encoding/json" - "fmt" - "log" - "net/http" - "strings" - "sync" - "time" - - "github.com/weaveworks/scope/common/sanitize" -) - -const ( - initialBackoff = 1 * time.Second - maxBackoff = 60 * time.Second -) +import "bytes" // Publisher is something which can send a buffered set of data somewhere, // probably to a remote collector. @@ -24,195 +8,3 @@ type Publisher interface { Publish(*bytes.Buffer) error Stop() } - -// HTTPPublisher publishes reports by POST to a fixed endpoint. -type HTTPPublisher struct { - url string - token string - probeID string -} - -// ScopeProbeIDHeader is the header we use to carry the probe's unique ID. The -// ID is currently set to the probe's hostname. It's designed to deduplicate -// reports from the same probe to the same receiver, in case the probe is -// configured to publish to multiple receivers that resolve to the same app. -const ScopeProbeIDHeader = "X-Scope-Probe-ID" - -// 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) - if err != nil { - return "", nil, err - } - defer resp.Body.Close() - var apiResponse struct { - ID string `json:"id"` - } - if err := json.NewDecoder(resp.Body).Decode(&apiResponse); err != nil { - return "", nil, err - } - return apiResponse.ID, &HTTPPublisher{ - url: sanitize.URL("http://", 0, "/api/report")(target), - token: token, - probeID: probeID, - }, nil -} - -func (p HTTPPublisher) String() string { - return p.url -} - -// Publish publishes the report to the URL. -func (p HTTPPublisher) Publish(buf *bytes.Buffer) error { - req, err := http.NewRequest("POST", p.url, buf) - if err != nil { - return err - } - req.Header.Set("Authorization", AuthorizationHeader(p.token)) - req.Header.Set(ScopeProbeIDHeader, p.probeID) - req.Header.Set("Content-Encoding", "gzip") - // req.Header.Set("Content-Type", "application/binary") // TODO: we should use http.DetectContentType(..) on the gob'ed - - resp, err := http.DefaultClient.Do(req) - if err != nil { - return err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return fmt.Errorf(resp.Status) - } - return nil -} - -// Stop implements Publisher -func (p HTTPPublisher) Stop() {} - -// AuthorizationHeader returns a value suitable for an HTTP Authorization -// header, based on the passed token string. -func AuthorizationHeader(token string) string { - return fmt.Sprintf("Scope-Probe token=%s", token) -} - -// BackgroundPublisher is a publisher which does the publish asynchronously. -// It will only do one publish at once; if there is an ongoing publish, -// concurrent publishes are dropped. -type BackgroundPublisher struct { - publisher Publisher - reports chan *bytes.Buffer - quit chan struct{} -} - -// NewBackgroundPublisher creates a new BackgroundPublisher with the given publisher -func NewBackgroundPublisher(p Publisher) *BackgroundPublisher { - result := &BackgroundPublisher{ - publisher: p, - reports: make(chan *bytes.Buffer), - quit: make(chan struct{}), - } - go result.loop() - return result -} - -func (b *BackgroundPublisher) loop() { - backoff := initialBackoff - - for r := range b.reports { - err := b.publisher.Publish(r) - if err == nil { - backoff = initialBackoff - continue - } - - log.Printf("Error publishing to %s, backing off %s: %v", b.publisher, backoff, err) - select { - case <-time.After(backoff): - case <-b.quit: - } - backoff = backoff * 2 - if backoff > maxBackoff { - backoff = maxBackoff - } - } -} - -// Publish implements Publisher -func (b *BackgroundPublisher) Publish(buf *bytes.Buffer) error { - select { - case b.reports <- buf: - default: - } - return nil -} - -// Stop implements Publisher -func (b *BackgroundPublisher) Stop() { - close(b.reports) - close(b.quit) - b.publisher.Stop() -} - -// MultiPublisher implements Publisher over a set of publishers. -type MultiPublisher struct { - mtx sync.RWMutex - factory func(string) (Publisher, error) - m map[string]Publisher -} - -// NewMultiPublisher returns a new MultiPublisher ready for use. The factory -// should be e.g. NewHTTPPublisher, except you need to curry it over the -// probe token. -func NewMultiPublisher(factory func(string) (Publisher, error)) *MultiPublisher { - return &MultiPublisher{ - factory: factory, - m: map[string]Publisher{}, - } -} - -// Add allows additional targets to be added dynamically. It will dedupe -// identical targets. TODO we have no good mechanism to remove. -func (p *MultiPublisher) Add(target string) { - p.mtx.Lock() - defer p.mtx.Unlock() - - if _, ok := p.m[target]; ok { - return - } - - publisher, err := p.factory(target) - if err != nil { - log.Printf("multi-publisher: %v", err) - return - } - - p.m[target] = publisher -} - -// Publish implements Publisher by emitting the report to all publishers. -func (p *MultiPublisher) Publish(buf *bytes.Buffer) error { - p.mtx.RLock() - defer p.mtx.RUnlock() - - var errs []string - for _, publisher := range p.m { - if err := publisher.Publish(bytes.NewBuffer(buf.Bytes())); err != nil { - errs = append(errs, err.Error()) - } - } - - if len(errs) > 0 { - return fmt.Errorf(strings.Join(errs, "; ")) - } - return nil -} - -// Stop implements Publisher -func (p *MultiPublisher) Stop() { - p.mtx.RLock() - defer p.mtx.RUnlock() - - for _, publisher := range p.m { - publisher.Stop() - } -}