Merge pull request #3846 from weaveworks/query-collectors

Multitenant mode: fetch live data from collectors

app/api_report.go

@@ -19,7 +19,7 @@ func makeRawReportHandler(rep Reporter) CtxHandlerFunc {
 			return
 		}
 		censorCfg := report.GetCensorConfigFromRequest(r)
-		respondWith(ctx, w, http.StatusOK, report.CensorRawReport(rawReport, censorCfg))
+		respondWithReport(ctx, w, r, report.CensorRawReport(rawReport, censorCfg))
 	}
 }
 

app/multitenant/aws_collector.go

@@ -142,13 +142,7 @@ type AWSCollectorConfig struct {
 	MemcacheClient *MemcacheClient
 	Window         time.Duration
 	MaxTopNodes    int
-}
-
-// if StoreInterval is set, reports are merged into here and held until flushed to store
-type pendingEntry struct {
-	sync.Mutex
-	report report.Report
-	count  int
+	CollectorAddr  string
 }
 
 type awsCollector struct {
@@ -162,6 +156,9 @@ type awsCollector struct {
 	nats        *nats.Conn
 	waitersLock sync.Mutex
 	waiters     map[watchKey]*nats.Subscription
+
+	collectors   []string
+	lastResolved time.Time
 }
 
 // Shortcut reports:
@@ -205,7 +202,8 @@ func NewAWSCollector(config AWSCollectorConfig) (AWSCollector, error) {
 		waiters:   map[watchKey]*nats.Subscription{},
 	}
 
-	if config.StoreInterval != 0 {
+	// If given a StoreInterval we will be storing periodically; if not we only answer queries
+	if c.isCollector() {
 		c.ticker = time.NewTicker(config.StoreInterval)
 		go c.flushLoop()
 	}
@@ -247,11 +245,17 @@ func (c *awsCollector) flushPending(ctx context.Context) {
 			entry := value.(*pendingEntry)
 
 			entry.Lock()
-			rpt, count := entry.report, entry.count
-			entry.report, entry.count = report.MakeReport(), 0
+			rpt := entry.report
+			entry.report = nil
+			if entry.older == nil {
+				entry.older = make([]*report.Report, c.cfg.Window/c.cfg.StoreInterval)
+			} else {
+				copy(entry.older[1:], entry.older) // move everything down one
+			}
+			entry.older[0] = rpt
 			entry.Unlock()
 
-			if count > 0 {
+			if rpt != nil {
 				// serialise reports on one goroutine to limit CPU usage
 				buf, err := rpt.WriteBinary()
 				if err != nil {
@@ -464,15 +468,8 @@ func (c *awsCollector) massageReport(userid string, report report.Report) report
 	return report
 }
 
-/*
-S3 stores original reports from one probe at the timestamp they arrived at collector.
-Collector also sends every report to memcached.
-The in-memory cache stores:
- - individual reports deserialised, under S3 key for report
- - sets of reports in interval [t,t+3) merged, under key "instance:t"
-   - so to check the cache for reports from 14:31:00 to 14:31:15 you would request 5 keys 3 seconds apart
-*/
-
+// If we are running as a Query service, fetch data and merge into a report
+// If we are running as a Collector and the request is for live data, merge in-memory data and return
 func (c *awsCollector) Report(ctx context.Context, timestamp time.Time) (report.Report, error) {
 	span, ctx := opentracing.StartSpanFromContext(ctx, "awsCollector.Report")
 	defer span.Finish()
@@ -481,11 +478,36 @@ func (c *awsCollector) Report(ctx context.Context, timestamp time.Time) (report.
 		return report.MakeReport(), err
 	}
 	span.SetTag("userid", userid)
+	var reports []report.Report
+	if time.Since(timestamp) < c.cfg.Window {
+		reports, err = c.reportsFromLive(ctx, userid)
+	} else {
+		reports, err = c.reportsFromStore(ctx, userid, timestamp)
+	}
+	if err != nil {
+		return report.MakeReport(), err
+	}
+	span.LogFields(otlog.Int("merging", len(reports)))
+	return c.merger.Merge(reports), nil
+}
+
+/*
+Given a timestamp in the past, fetch reports within the window from store or cache
+
+S3 stores original reports from one probe at the timestamp they arrived at collector.
+Collector also sends every report to memcached.
+The in-memory cache stores:
+ - individual reports deserialised, under S3 key for report
+ - sets of reports in interval [t,t+3) merged, under key "instance:t"
+   - so to check the cache for reports from 14:31:00 to 14:31:15 you would request 5 keys 3 seconds apart
+*/
+func (c *awsCollector) reportsFromStore(ctx context.Context, userid string, timestamp time.Time) ([]report.Report, error) {
+	span := opentracing.SpanFromContext(ctx)
 	end := timestamp
 	start := end.Add(-c.cfg.Window)
 	reportKeys, err := c.getReportKeys(ctx, userid, start, end)
 	if err != nil {
-		return report.MakeReport(), err
+		return nil, err
 	}
 	span.LogFields(otlog.Int("keys", len(reportKeys)), otlog.String("timestamp", timestamp.String()))
 
@@ -496,18 +518,17 @@ func (c *awsCollector) Report(ctx context.Context, timestamp time.Time) (report.
 	for ; ts+(reportQuantisationInterval+gracePeriod).Nanoseconds() < endTS; ts += reportQuantisationInterval.Nanoseconds() {
 		quantumReport, err := c.reportForQuantum(ctx, userid, reportKeys, ts)
 		if err != nil {
-			return report.MakeReport(), err
+			return nil, err
 		}
 		reports = append(reports, quantumReport)
 	}
 	// Fetch individual reports for the period after the last quantum
 	last, err := c.reportsForKeysInRange(ctx, userid, reportKeys, ts, endTS)
 	if err != nil {
-		return report.MakeReport(), err
+		return nil, err
 	}
 	reports = append(reports, last...)
-	span.LogFields(otlog.Int("merging", len(reports)))
-	return c.merger.Merge(reports), nil
+	return reports, nil
 }
 
 // Fetch a merged report either from cache or from store which we put in cache
@@ -546,6 +567,10 @@ func (c *awsCollector) HasReports(ctx context.Context, timestamp time.Time) (boo
 	if err != nil {
 		return false, err
 	}
+	if time.Since(timestamp) < c.cfg.Window {
+		has, err := c.hasReportsFromLive(ctx, userid)
+		return has, err
+	}
 	start := timestamp.Add(-c.cfg.Window)
 	reportKeys, err := c.getReportKeys(ctx, userid, start, timestamp)
 	return len(reportKeys) > 0, err
@@ -681,6 +706,9 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report, buf []byte) e
 	if err != nil {
 		return err
 	}
+	if c.cfg.StoreInterval == 0 {
+		return fmt.Errorf("--app.collector.store-interval must be non-zero")
+	}
 
 	// Shortcut reports are published to nats but not persisted -
 	// we'll get a full report from the same probe in a few seconds
@@ -702,23 +730,8 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report, buf []byte) e
 		return nil
 	}
 
-	if c.cfg.StoreInterval == 0 {
-		rowKey, colKey, reportKey := calculateReportKeys(userid, time.Now())
-		err = c.persistReport(ctx, userid, rowKey, colKey, reportKey, buf)
-		if err != nil {
-			return err
-		}
-	} else {
-		rep = c.massageReport(userid, rep)
-		entry := &pendingEntry{report: report.MakeReport()}
-		if e, found := c.pending.LoadOrStore(userid, entry); found {
-			entry = e.(*pendingEntry)
-		}
-		entry.Lock()
-		entry.report.UnsafeMerge(rep)
-		entry.count++
-		entry.Unlock()
-	}
+	rep = c.massageReport(userid, rep)
+	c.addToLive(ctx, userid, rep)
 
 	return nil
 }

app/multitenant/collector.go

@@ -0,0 +1,198 @@
+package multitenant
+
+// Collect reports from probes per-tenant, and supply them to queriers on demand
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+	"sync"
+	"time"
+
+	"context"
+
+	"github.com/opentracing-contrib/go-stdlib/nethttp"
+	opentracing "github.com/opentracing/opentracing-go"
+	log "github.com/sirupsen/logrus"
+	"github.com/weaveworks/common/user"
+	"github.com/weaveworks/scope/report"
+	"golang.org/x/sync/errgroup"
+)
+
+// if StoreInterval is set, reports are merged into here and held until flushed to store
+type pendingEntry struct {
+	sync.Mutex
+	report *report.Report
+	older  []*report.Report
+}
+
+// We are building up a report in memory; merge into that and it will be saved shortly
+// NOTE: may retain a reference to rep; must not be used by caller after this.
+func (c *awsCollector) addToLive(ctx context.Context, userid string, rep report.Report) {
+	entry := &pendingEntry{}
+	if e, found := c.pending.LoadOrStore(userid, entry); found {
+		entry = e.(*pendingEntry)
+	}
+	entry.Lock()
+	if entry.report == nil {
+		entry.report = &rep
+	} else {
+		entry.report.UnsafeMerge(rep)
+	}
+	entry.Unlock()
+}
+
+func (c *awsCollector) isCollector() bool {
+	return c.cfg.StoreInterval != 0
+}
+
+func (c *awsCollector) hasReportsFromLive(ctx context.Context, userid string) (bool, error) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "hasReportsFromLive")
+	defer span.Finish()
+	if c.isCollector() {
+		e, found := c.pending.Load(userid)
+		if !found {
+			return false, nil
+		}
+		entry := e.(*pendingEntry)
+		entry.Lock()
+		defer entry.Unlock()
+		if entry.report != nil {
+			return true, nil
+		}
+		for _, v := range entry.older {
+			if v != nil {
+				return true, nil
+			}
+		}
+		return false, nil
+	}
+	// We are a querier: ask each collector if it has any
+	// (serially, since we will bail out on the first one that has reports)
+	addrs := resolve(c.cfg.CollectorAddr)
+	for _, addr := range addrs {
+		body, err := oneCall(ctx, addr, "/api/probes?sparse=true", userid)
+		if err != nil {
+			return false, err
+		}
+		var hasReports bool
+		decoder := json.NewDecoder(body)
+		if err := decoder.Decode(&hasReports); err != nil {
+			log.Errorf("Error encoding response: %v", err)
+		}
+		body.Close()
+		if hasReports {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func (c *awsCollector) reportsFromLive(ctx context.Context, userid string) ([]report.Report, error) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "reportsFromLive")
+	defer span.Finish()
+	if c.isCollector() {
+		e, found := c.pending.Load(userid)
+		if !found {
+			return nil, nil
+		}
+		entry := e.(*pendingEntry)
+		entry.Lock()
+		ret := make([]report.Report, 0, len(entry.older)+1)
+		if entry.report != nil {
+			ret = append(ret, entry.report.Copy()) // Copy contents because this report is being unsafe-merged to
+		}
+		for _, v := range entry.older {
+			if v != nil {
+				ret = append(ret, *v) // no copy because older reports are immutable
+			}
+		}
+		entry.Unlock()
+		return ret, nil
+	}
+
+	// We are a querier: fetch the most up-to-date reports from collectors
+	if time.Since(c.lastResolved) > time.Second*5 {
+		c.collectors = resolve(c.cfg.CollectorAddr)
+		c.lastResolved = time.Now()
+	}
+	reports := make([]*report.Report, len(c.collectors))
+	// make a call to each collector and fetch its data for this userid
+	g, ctx := errgroup.WithContext(ctx)
+	for i, addr := range c.collectors {
+		i, addr := i, addr // https://golang.org/doc/faq#closures_and_goroutines
+		g.Go(func() error {
+			body, err := oneCall(ctx, addr, "/api/report", userid)
+			if err != nil {
+				log.Warnf("error calling '%s': %v", addr, err)
+				return nil
+			}
+			reports[i], err = report.MakeFromBinary(ctx, body, false, true)
+			body.Close()
+			if err != nil {
+				log.Warnf("error decoding: %v", err)
+				return nil
+			}
+			return nil
+		})
+	}
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+
+	// dereference pointers into the expected return format
+	ret := make([]report.Report, 0, len(reports))
+	for _, rpt := range reports {
+		if rpt != nil {
+			ret = append(ret, *rpt)
+		}
+	}
+
+	return ret, nil
+}
+
+func resolve(name string) []string {
+	_, addrs, err := net.LookupSRV("", "", name)
+	if err != nil {
+		log.Warnf("Cannot resolve '%s': %v", name, err)
+		return []string{}
+	}
+	endpoints := make([]string, 0, len(addrs))
+	for _, addr := range addrs {
+		port := strconv.Itoa(int(addr.Port))
+		endpoints = append(endpoints, net.JoinHostPort(addr.Target, port))
+	}
+	return endpoints
+}
+
+func oneCall(ctx context.Context, endpoint, path, userid string) (io.ReadCloser, error) {
+	fullPath := "http://" + endpoint + path
+	req, err := http.NewRequest("GET", fullPath, nil)
+	if err != nil {
+		return nil, fmt.Errorf("error making request %s: %w", fullPath, err)
+	}
+	req = req.WithContext(ctx)
+	req.Header.Set(user.OrgIDHeaderName, userid)
+	req.Header.Set("Accept", "application/msgpack")
+	req.Header.Set("Accept-Encoding", "identity") // disable compression
+	if parentSpan := opentracing.SpanFromContext(ctx); parentSpan != nil {
+		var ht *nethttp.Tracer
+		req, ht = nethttp.TraceRequest(parentSpan.Tracer(), req, nethttp.OperationName("Collector Fetch"))
+		defer ht.Finish()
+	}
+	client := &http.Client{Transport: &nethttp.Transport{}}
+	res, err := client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("error getting %s: %w", fullPath, err)
+	}
+	if res.StatusCode != http.StatusOK {
+		content, _ := io.ReadAll(res.Body)
+		res.Body.Close()
+		return nil, fmt.Errorf("error from collector: %s (%s)", res.Status, string(content))
+	}
+
+	return res.Body, nil
+}

app/server_helpers.go

@@ -1,11 +1,14 @@
 package app
 
 import (
+	"bytes"
 	"context"
 	"net/http"
+	"strings"
 
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/ugorji/go/codec"
+	"github.com/weaveworks/scope/report"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -33,3 +36,31 @@ func respondWith(ctx context.Context, w http.ResponseWriter, code int, response
 		log.Errorf("Error encoding response: %v", err)
 	}
 }
+
+// Similar to the above function, but respect the request's Accept header.
+// Possibly we should do a complete parse of Accept, but for now just rudimentary check
+func respondWithReport(ctx context.Context, w http.ResponseWriter, req *http.Request, response report.Report) {
+	accept := req.Header.Get("Accept")
+	if strings.HasPrefix(accept, "application/msgpack") {
+		buf := bytes.Buffer{}
+		encoder := codec.NewEncoder(&buf, &codec.MsgpackHandle{})
+		if err := encoder.Encode(response); err != nil {
+			log.Errorf("Error encoding response: %v", err)
+		}
+		if span := opentracing.SpanFromContext(ctx); span != nil {
+			span.LogKV("encoded-size", len(buf.Bytes()))
+		}
+		w.Header().Set("Content-Type", "application/msgpack")
+		w.WriteHeader(http.StatusOK)
+		w.Write(buf.Bytes())
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Add("Cache-Control", "no-cache")
+	w.WriteHeader(http.StatusOK)
+	encoder := codec.NewEncoder(w, &codec.JsonHandle{})
+	if err := encoder.Encode(response); err != nil {
+		log.Errorf("Error encoding response: %v", err)
+	}
+}

go.mod

@@ -52,6 +52,7 @@ require (
 	github.com/nats-io/nuid v0.0.0-20160402145409-a5152d67cf63 // indirect
 	github.com/opencontainers/runc v1.0.0-rc5 // indirect
 	github.com/openebs/k8s-snapshot-client v0.0.0-20180831100134-a6506305fb16
+	github.com/opentracing-contrib/go-stdlib v0.0.0-20190519235532-cf7a6c988dc9
 	github.com/opentracing/opentracing-go v1.1.0
 	github.com/paypal/ionet v0.0.0-20130919195445-ed0aaebc5417
 	github.com/pborman/uuid v0.0.0-20150824212802-cccd189d45f7

prog/app.go

@@ -90,7 +90,7 @@ func router(collector app.Collector, controlRouter app.ControlRouter, pipeRouter
 }
 
 func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL string, storeInterval time.Duration, natsHostname string,
-	memcacheConfig multitenant.MemcacheConfig, window time.Duration, maxTopNodes int, createTables bool) (app.Collector, error) {
+	memcacheConfig multitenant.MemcacheConfig, window time.Duration, maxTopNodes int, createTables bool, collectorAddr string) (app.Collector, error) {
 	if collectorURL == "local" {
 		return app.NewCollector(window), nil
 	}
@@ -134,6 +134,7 @@ func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL string,
 				MemcacheClient: memcacheClient,
 				Window:         window,
 				MaxTopNodes:    maxTopNodes,
+				CollectorAddr:  collectorAddr,
 			},
 		)
 		if err != nil {
@@ -248,7 +249,7 @@ func appMain(flags appFlags) {
 			Service:          flags.memcachedService,
 			CompressionLevel: flags.memcachedCompressionLevel,
 		},
-		flags.window, flags.maxTopNodes, flags.awsCreateTables)
+		flags.window, flags.maxTopNodes, flags.awsCreateTables, flags.collectorAddr)
 	if err != nil {
 		log.Fatalf("Error creating collector: %v", err)
 		return

prog/main.go

@@ -162,7 +162,8 @@ type appFlags struct {
 	containerName  string
 	dockerEndpoint string
 
-	collectorURL              string
+	collectorURL              string // how collector talks to backing store (or "local" if none)
+	collectorAddr             string // how to find collectors if deployed as microservices
 	s3URL                     string
 	storeInterval             time.Duration
 	controlRouterURL          string
@@ -376,6 +377,7 @@ func setupFlags(flags *flags) {
 	flag.Var(&flags.containerLabelFilterFlagsExclude, "app.container-label-filter-exclude", "Add container label-based view filter that excludes containers with the given label, specified as title:label. Multiple flags are accepted. Example: --app.container-label-filter-exclude='Database Containers:role=db'")
 
 	flag.StringVar(&flags.app.collectorURL, "app.collector", "local", "Collector to use (local, dynamodb, or file/directory)")
+	flag.StringVar(&flags.app.collectorAddr, "app.collector-addr", "", "Address to look up collectors when deployed as microservices")
 	flag.StringVar(&flags.app.s3URL, "app.collector.s3", "local", "S3 URL to use (when collector is dynamodb)")
 	flag.DurationVar(&flags.app.storeInterval, "app.collector.store-interval", 0, "How often to store merged incoming reports. If 0, reports are stored unmerged as they arrive.")
 	flag.StringVar(&flags.app.controlRouterURL, "app.control.router", "local", "Control router to use (local or sqs)")

report/marshal.go

@@ -102,10 +102,6 @@ func MakeFromBinary(ctx context.Context, r io.Reader, gzipped bool, msgpack bool
 	if err != nil {
 		return nil, err
 	}
-	rep := MakeReport()
-	if err := codec.NewDecoderBytes(buf.Bytes(), codecHandle(msgpack)).Decode(&rep); err != nil {
-		return nil, err
-	}
 	log.Debugf(
 		"Received report sizes: compressed %d bytes, uncompressed %d bytes (%.2f%%)",
 		compressedSize,
@@ -113,6 +109,10 @@ func MakeFromBinary(ctx context.Context, r io.Reader, gzipped bool, msgpack bool
 		float32(compressedSize)/float32(uncompressedSize)*100,
 	)
 	span.LogFields(otlog.Uint64("compressedSize", compressedSize), otlog.Int64("uncompressedSize", uncompressedSize))
+	rep := MakeReport()
+	if err := codec.NewDecoderBytes(buf.Bytes(), codecHandle(msgpack)).Decode(&rep); err != nil {
+		return nil, err
+	}
 	return &rep, nil
 }
 

vendor/modules.txt

@@ -293,6 +293,7 @@ github.com/openebs/k8s-snapshot-client/snapshot/pkg/client/clientset/versioned
 github.com/openebs/k8s-snapshot-client/snapshot/pkg/client/clientset/versioned/scheme
 github.com/openebs/k8s-snapshot-client/snapshot/pkg/client/clientset/versioned/typed/volumesnapshot/v1
 # github.com/opentracing-contrib/go-stdlib v0.0.0-20190519235532-cf7a6c988dc9
+## explicit
 github.com/opentracing-contrib/go-stdlib/nethttp
 # github.com/opentracing/opentracing-go v1.1.0
 ## explicit