diff --git a/app/api_report.go b/app/api_report.go index 7ab162528..079625b54 100644 --- a/app/api_report.go +++ b/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)) } } diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index 03c7203d7..c18966125 100644 --- a/app/multitenant/aws_collector.go +++ b/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 } diff --git a/app/multitenant/collector.go b/app/multitenant/collector.go new file mode 100644 index 000000000..3c23373d6 --- /dev/null +++ b/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 +} diff --git a/app/server_helpers.go b/app/server_helpers.go index a9970c704..90f03d9d7 100644 --- a/app/server_helpers.go +++ b/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) + } +} diff --git a/go.mod b/go.mod index ae4052136..73322e315 100644 --- a/go.mod +++ b/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 diff --git a/prog/app.go b/prog/app.go index 5648b57d9..dff8a88da 100644 --- a/prog/app.go +++ b/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 diff --git a/prog/main.go b/prog/main.go index 30e9e1f86..66eb5c4c7 100644 --- a/prog/main.go +++ b/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)") diff --git a/report/marshal.go b/report/marshal.go index 38d98253d..c93b71071 100644 --- a/report/marshal.go +++ b/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 } diff --git a/vendor/modules.txt b/vendor/modules.txt index 1c1a3c808..2410cb25b 100644 --- a/vendor/modules.txt +++ b/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