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 {