From f7bdedc149b743049a57fb513ace0db52bdbc1a6 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Mon, 4 Jul 2016 13:25:45 +0100 Subject: [PATCH 1/6] Config struct for memcache client --- app/multitenant/memcache_client.go | 23 ++++++++++++++++------- prog/app.go | 9 +++++++-- 2 files changed, 23 insertions(+), 9 deletions(-) diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index 1a4738369..591de0db7 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -53,28 +53,37 @@ type MemcacheClient struct { wait sync.WaitGroup } +// MemcacheConfig defines how a MemcacheClient should be constructed. +type MemcacheConfig struct { + Host string + Timeout time.Duration + Service string + UpdateInterval time.Duration + Expiration int32 +} + // NewMemcacheClient creates a new MemcacheClient that gets its server list // from SRV and updates the server list on a regular basis. -func NewMemcacheClient(host string, timeout time.Duration, service string, updateInterval time.Duration, expiration int32) *MemcacheClient { +func NewMemcacheClient(config MemcacheConfig) *MemcacheClient { var servers memcache.ServerList client := memcache.NewFromSelector(&servers) - client.Timeout = timeout + client.Timeout = config.Timeout newClient := &MemcacheClient{ client: client, serverList: &servers, - expiration: expiration, - hostname: host, - service: service, + expiration: config.Expiration, + hostname: config.Host, + service: config.Service, quit: make(chan struct{}), } err := newClient.updateMemcacheServers() if err != nil { - log.Errorf("Error setting memcache servers to '%v': %v", host, err) + log.Errorf("Error setting memcache servers to '%v': %v", config.Host, err) } newClient.wait.Add(1) - go newClient.updateLoop(updateInterval) + go newClient.updateLoop(config.UpdateInterval) return newClient } diff --git a/prog/app.go b/prog/app.go index a3a02719c..1230c6b5f 100644 --- a/prog/app.go +++ b/prog/app.go @@ -110,8 +110,13 @@ func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHo var memcacheClient *multitenant.MemcacheClient if memcachedHostname != "" { memcacheClient = multitenant.NewMemcacheClient( - memcachedHostname, memcachedTimeout, memcachedService, - memcacheUpdateInterval, memcacheExpiration, + multitenant.MemcacheConfig{ + Host: memcachedHostname, + Timeout: memcachedTimeout, + Expiration: memcacheExpiration, + UpdateInterval: memcacheUpdateInterval, + Service: memcachedService, + }, ) } awsCollector, err := multitenant.NewAWSCollector( From c1dab17fb3a94d8e325017dd0c10663abceb3147 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Mon, 4 Jul 2016 13:30:23 +0100 Subject: [PATCH 2/6] Make expiration a Duration --- app/multitenant/memcache_client.go | 6 +++--- prog/app.go | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index 591de0db7..f82ec8966 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -56,10 +56,10 @@ type MemcacheClient struct { // MemcacheConfig defines how a MemcacheClient should be constructed. type MemcacheConfig struct { Host string - Timeout time.Duration Service string + Timeout time.Duration UpdateInterval time.Duration - Expiration int32 + Expiration time.Duration } // NewMemcacheClient creates a new MemcacheClient that gets its server list @@ -72,7 +72,7 @@ func NewMemcacheClient(config MemcacheConfig) *MemcacheClient { newClient := &MemcacheClient{ client: client, serverList: &servers, - expiration: config.Expiration, + expiration: int32(config.Expiration.Seconds()), hostname: config.Host, service: config.Service, quit: make(chan struct{}), diff --git a/prog/app.go b/prog/app.go index 1230c6b5f..3435cd235 100644 --- a/prog/app.go +++ b/prog/app.go @@ -28,7 +28,7 @@ import ( ) const ( - memcacheExpiration = 15 // seconds + memcacheExpiration = 15 * time.Second memcacheUpdateInterval = 1 * time.Minute ) From 7dd2c6371ec839bafbddc5e69c14cfbe2752148a Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Mon, 4 Jul 2016 13:40:50 +0100 Subject: [PATCH 3/6] Parametrize window rather than assuming default --- app/multitenant/aws_collector.go | 19 +++++++++++-------- prog/app.go | 4 ++-- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index 477e2261b..fb382db2e 100644 --- a/app/multitenant/aws_collector.go +++ b/app/multitenant/aws_collector.go @@ -23,12 +23,10 @@ import ( ) const ( - hourField = "hour" - tsField = "ts" - reportField = "report" - reportCacheSize = (15 / 3) * 10 * 5 // (window size * report rate) * number of hosts per user * number of users - reportCacheExpiration = 15 * time.Second - natsTimeout = 10 * time.Second + hourField = "hour" + tsField = "ts" + reportField = "report" + natsTimeout = 10 * time.Second ) var ( @@ -102,6 +100,7 @@ type AWSCollectorConfig struct { S3Store *S3Store NatsHost string MemcacheClient *MemcacheClient + Window time.Duration } type awsCollector struct { @@ -112,6 +111,7 @@ type awsCollector struct { merger app.Merger inProcess inProcessStore memcache *MemcacheClient + window time.Duration nats *nats.Conn waitersLock sync.Mutex @@ -144,14 +144,17 @@ func NewAWSCollector(config AWSCollectorConfig) (AWSCollector, error) { } } + // (window * report rate) * number of hosts per user * number of users + reportCacheSize := (int(config.Window.Seconds()) / 3) * 10 * 5 return &awsCollector{ db: dynamodb.New(session.New(config.DynamoDBConfig)), s3: config.S3Store, userIDer: config.UserIDer, tableName: config.DynamoTable, merger: app.NewSmartMerger(), - inProcess: newInProcessStore(reportCacheSize, reportCacheExpiration), + inProcess: newInProcessStore(reportCacheSize, config.Window), memcache: config.MemcacheClient, + window: config.Window, nats: nc, waiters: map[watchKey]*nats.Subscription{}, }, nil @@ -294,7 +297,7 @@ func (c *awsCollector) getReports(reportKeys []string) ([]report.Report, error) func (c *awsCollector) Report(ctx context.Context) (report.Report, error) { var ( now = time.Now() - start = now.Add(-15 * time.Second) + start = now.Add(-c.window) rowStart, rowEnd = start.UnixNano() / time.Hour.Nanoseconds(), now.UnixNano() / time.Hour.Nanoseconds() userid, err = c.userIDer(ctx) ) diff --git a/prog/app.go b/prog/app.go index 3435cd235..eaae140f6 100644 --- a/prog/app.go +++ b/prog/app.go @@ -28,7 +28,6 @@ import ( ) const ( - memcacheExpiration = 15 * time.Second memcacheUpdateInterval = 1 * time.Minute ) @@ -113,7 +112,7 @@ func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHo multitenant.MemcacheConfig{ Host: memcachedHostname, Timeout: memcachedTimeout, - Expiration: memcacheExpiration, + Expiration: window, UpdateInterval: memcacheUpdateInterval, Service: memcachedService, }, @@ -127,6 +126,7 @@ func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHo S3Store: &s3Store, NatsHost: natsHostname, MemcacheClient: memcacheClient, + Window: window, }, ) if err != nil { From 848574ecedb35874ae7a0a8c61313b9807d7c25b Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Mon, 4 Jul 2016 14:20:14 +0100 Subject: [PATCH 4/6] Make expiration a flag --- prog/app.go | 6 +++--- prog/main.go | 20 +++++++++++--------- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/prog/app.go b/prog/app.go index eaae140f6..f786a3883 100644 --- a/prog/app.go +++ b/prog/app.go @@ -80,7 +80,7 @@ func awsConfigFromURL(url *url.URL) (*aws.Config, error) { return config, nil } -func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHostname, memcachedHostname string, memcachedTimeout time.Duration, memcachedService string, window time.Duration, createTables bool) (app.Collector, error) { +func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHostname, memcachedHostname string, memcachedTimeout time.Duration, memcachedService string, memcachedExpiration, window time.Duration, createTables bool) (app.Collector, error) { if collectorURL == "local" { return app.NewCollector(window), nil } @@ -112,7 +112,7 @@ func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHo multitenant.MemcacheConfig{ Host: memcachedHostname, Timeout: memcachedTimeout, - Expiration: window, + Expiration: memcachedExpiration, UpdateInterval: memcacheUpdateInterval, Service: memcachedService, }, @@ -210,7 +210,7 @@ func appMain(flags appFlags) { collector, err := collectorFactory( userIDer, flags.collectorURL, flags.s3URL, flags.natsHostname, flags.memcachedHostname, - flags.memcachedTimeout, flags.memcachedService, flags.window, flags.awsCreateTables) + flags.memcachedTimeout, flags.memcachedService, flags.memcachedExpiration, flags.window, flags.awsCreateTables) if err != nil { log.Fatalf("Error creating collector: %v", err) return diff --git a/prog/main.go b/prog/main.go index f1c9da818..b989ad122 100644 --- a/prog/main.go +++ b/prog/main.go @@ -99,15 +99,16 @@ type appFlags struct { containerName string dockerEndpoint string - collectorURL string - s3URL string - controlRouterURL string - pipeRouterURL string - natsHostname string - memcachedHostname string - memcachedTimeout time.Duration - memcachedService string - userIDHeader string + collectorURL string + s3URL string + controlRouterURL string + pipeRouterURL string + natsHostname string + memcachedHostname string + memcachedTimeout time.Duration + memcachedService string + memcachedExpiration time.Duration + userIDHeader string awsCreateTables bool consulInf string @@ -190,6 +191,7 @@ func main() { flag.StringVar(&flags.app.natsHostname, "app.nats", "", "Hostname for NATS service to use for shortcut reports. If empty, shortcut reporting will be disabled.") flag.StringVar(&flags.app.memcachedHostname, "app.memcached.hostname", "", "Hostname for memcached service to use when caching reports. If empty, no memcached will be used.") flag.DurationVar(&flags.app.memcachedTimeout, "app.memcached.timeout", 100*time.Millisecond, "Maximum time to wait before giving up on memcached requests.") + flag.DurationVar(&flags.app.memcachedExpiration, "app.memcached.expiration", 15*time.Second, "How long reports stay in the memcache.") flag.StringVar(&flags.app.memcachedService, "app.memcached.service", "memcached", "SRV service used to discover memcache servers.") flag.StringVar(&flags.app.userIDHeader, "app.userid.header", "", "HTTP header to use as userid") From 31c88fd62b9f0336950e5fa597dd336556951e50 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Mon, 4 Jul 2016 16:03:50 +0100 Subject: [PATCH 5/6] Instrumentation that we might like to keep --- app/multitenant/aws_collector.go | 1 + app/multitenant/memcache_client.go | 5 +++++ 2 files changed, 6 insertions(+) diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index fb382db2e..b7ed86282 100644 --- a/app/multitenant/aws_collector.go +++ b/app/multitenant/aws_collector.go @@ -326,6 +326,7 @@ func (c *awsCollector) Report(ctx context.Context) (report.Report, error) { } } + log.Debugf("Fetching %d reports from %v to %v", len(reportKeys), start, now) reports, err := c.getReports(reportKeys) if err != nil { return report.MakeReport(), err diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index f82ec8966..bd3fb054c 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -188,6 +188,11 @@ func (c *MemcacheClient) FetchReports(keys []string) (map[string]report.Report, } } + if len(missing) > 0 { + sort.Strings(missing) + log.Warningf("Missing %d reports from memcache: %v", len(missing), missing) + } + memcacheHits.Add(float64(len(reports))) memcacheRequests.Add(float64(len(keys))) return reports, missing, nil From 45d0e5441aaefbbcd49bf631291bd319c4ff53e2 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Mon, 4 Jul 2016 16:04:04 +0100 Subject: [PATCH 6/6] Change default timeout for things in the memcache --- prog/main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/prog/main.go b/prog/main.go index b989ad122..6f9c22bde 100644 --- a/prog/main.go +++ b/prog/main.go @@ -191,7 +191,7 @@ func main() { flag.StringVar(&flags.app.natsHostname, "app.nats", "", "Hostname for NATS service to use for shortcut reports. If empty, shortcut reporting will be disabled.") flag.StringVar(&flags.app.memcachedHostname, "app.memcached.hostname", "", "Hostname for memcached service to use when caching reports. If empty, no memcached will be used.") flag.DurationVar(&flags.app.memcachedTimeout, "app.memcached.timeout", 100*time.Millisecond, "Maximum time to wait before giving up on memcached requests.") - flag.DurationVar(&flags.app.memcachedExpiration, "app.memcached.expiration", 15*time.Second, "How long reports stay in the memcache.") + flag.DurationVar(&flags.app.memcachedExpiration, "app.memcached.expiration", 2*15*time.Second, "How long reports stay in the memcache.") flag.StringVar(&flags.app.memcachedService, "app.memcached.service", "memcached", "SRV service used to discover memcache servers.") flag.StringVar(&flags.app.userIDHeader, "app.userid.header", "", "HTTP header to use as userid")