diff --git a/app/multitenant/dynamo_collector.go b/app/multitenant/dynamo_collector.go index 80773170d..3e3cc9149 100644 --- a/app/multitenant/dynamo_collector.go +++ b/app/multitenant/dynamo_collector.go @@ -20,7 +20,6 @@ import ( ) const ( - tableName = "reports" hourField = "hour" tsField = "ts" reportField = "report" @@ -45,16 +44,18 @@ type DynamoDBCollector interface { } type dynamoDBCollector struct { - userIDer UserIDer - db *dynamodb.DynamoDB + userIDer UserIDer + db *dynamodb.DynamoDB + tableName string } // NewDynamoDBCollector the reaper of souls // https://github.com/aws/aws-sdk-go/wiki/common-examples -func NewDynamoDBCollector(config *aws.Config, userIDer UserIDer) DynamoDBCollector { +func NewDynamoDBCollector(config *aws.Config, userIDer UserIDer, tableName string) DynamoDBCollector { return &dynamoDBCollector{ - db: dynamodb.New(session.New(config)), - userIDer: userIDer, + db: dynamodb.New(session.New(config)), + userIDer: userIDer, + tableName: tableName, } } @@ -68,13 +69,13 @@ func (c *dynamoDBCollector) CreateTables() error { return err } for _, s := range resp.TableNames { - if *s == tableName { + if *s == c.tableName { return nil } } params := &dynamodb.CreateTableInput{ - TableName: aws.String(tableName), + TableName: aws.String(c.tableName), AttributeDefinitions: []*dynamodb.AttributeDefinition{ { AttributeName: aws.String(hourField), @@ -105,7 +106,7 @@ func (c *dynamoDBCollector) CreateTables() error { WriteCapacityUnits: aws.Int64(5), }, } - log.Infof("Creating table %s", tableName) + log.Infof("Creating table %s", c.tableName) _, err = c.db.CreateTable(params) return err } @@ -114,7 +115,7 @@ func (c *dynamoDBCollector) getRows(userid string, row int64, start, end time.Ti rowKey := fmt.Sprintf("%s-%s", userid, strconv.FormatInt(row, 10)) startTime := time.Now() resp, err := c.db.Query(&dynamodb.QueryInput{ - TableName: aws.String(tableName), + TableName: aws.String(c.tableName), KeyConditions: map[string]*dynamodb.Condition{ hourField: { AttributeValueList: []*dynamodb.AttributeValue{ @@ -201,7 +202,7 @@ func (c *dynamoDBCollector) Add(ctx context.Context, rep report.Report) error { rowKey := fmt.Sprintf("%s-%s", userid, strconv.FormatInt(now.UnixNano()/time.Hour.Nanoseconds(), 10)) startTime := time.Now() _, err = c.db.PutItem(&dynamodb.PutItemInput{ - TableName: aws.String(tableName), + TableName: aws.String(c.tableName), Item: map[string]*dynamodb.AttributeValue{ hourField: { S: aws.String(rowKey), diff --git a/app/multitenant/sqs_control_router.go b/app/multitenant/sqs_control_router.go index 8beb16c96..59fd2683e 100644 --- a/app/multitenant/sqs_control_router.go +++ b/app/multitenant/sqs_control_router.go @@ -42,6 +42,7 @@ type sqsControlRouter struct { service *sqs.SQS responseQueueURL *string userIDer UserIDer + prefix string mtx sync.Mutex responses map[string]chan xfer.Response @@ -60,11 +61,12 @@ type sqsResponseMessage struct { } // NewSQSControlRouter the harbinger of death -func NewSQSControlRouter(config *aws.Config, userIDer UserIDer) app.ControlRouter { +func NewSQSControlRouter(config *aws.Config, userIDer UserIDer, prefix string) app.ControlRouter { result := &sqsControlRouter{ service: sqs.New(session.New(config)), responseQueueURL: nil, userIDer: userIDer, + prefix: prefix, responses: map[string]chan xfer.Response{}, probeWorkers: map[int64]*probeWorker{}, } @@ -110,7 +112,7 @@ func (cr *sqsControlRouter) loop() { ) for { // This app has a random id and uses this as a return path for all responses from probes. - name := fmt.Sprintf("control-app-%d", rand.Int63()) + name := fmt.Sprintf("%scontrol-app-%d", cr.prefix, rand.Int63()) responseQueueURL, err = cr.getOrCreateQueue(name) if err != nil { log.Errorf("Failed to create queue: %v", err) @@ -220,7 +222,7 @@ func (cr *sqsControlRouter) Handle(ctx context.Context, probeID string, req xfer return xfer.Response{}, fmt.Errorf("No SQS queue yet!") } - probeQueueName := fmt.Sprintf("probe-%s-%s", userID, probeID) + probeQueueName := fmt.Sprintf("%sprobe-%s-%s", cr.prefix, userID, probeID) start := time.Now() probeQueueURL, err := cr.service.GetQueueUrl(&sqs.GetQueueUrlInput{ QueueName: aws.String(probeQueueName), @@ -269,7 +271,7 @@ func (cr *sqsControlRouter) Register(ctx context.Context, probeID string, handle return 0, err } - name := fmt.Sprintf("probe-%s-%s", userID, probeID) + name := fmt.Sprintf("%sprobe-%s-%s", cr.prefix, userID, probeID) queueURL, err := cr.getOrCreateQueue(name) if err != nil { return 0, err diff --git a/circle.yml b/circle.yml index d8aee125c..dcd63404d 100644 --- a/circle.yml +++ b/circle.yml @@ -69,15 +69,15 @@ deployment: (test "${DOCKER_ORGANIZATION:-$DOCKER_USER}" == "weaveworks" || docker tag weaveworks/scope:latest ${DOCKER_ORGANIZATION:-$DOCKER_USER}/scope:latest) && docker push ${DOCKER_ORGANIZATION:-$DOCKER_USER}/scope ) - release: - branch: /release-[0-9]+\.[0-9]+/ - owner: weaveworks - commands: - - go get github.com/weaveworks/wordepress && cd /home/ubuntu/src/github.com/weaveworks/wordepress && git checkout v1.0.0 && cd cmd/wordepress && go get - - cd $SRCDIR; PRODUCT=scope tools/publish-site "$WP_LIVE_URL" "$WP_LIVE_USER" "$WP_LIVE_PASSWORD" - issues: - branch: /.*/ - owner: weaveworks - commands: - - go get github.com/weaveworks/wordepress && cd /home/ubuntu/src/github.com/weaveworks/wordepress && git checkout v1.0.0 && cd cmd/wordepress && go get - - cd $SRCDIR; PRODUCT=scope tools/publish-site "$WP_DEV_URL" "$WP_DEV_USER" "$WP_DEV_PASSWORD" +# release: +# branch: /release-[0-9]+\.[0-9]+/ +# owner: weaveworks +# commands: +# - go get github.com/weaveworks/wordepress && cd /home/ubuntu/src/github.com/weaveworks/wordepress && git checkout v1.0.0 && cd cmd/wordepress && go get +# - cd $SRCDIR; PRODUCT=scope tools/publish-site "$WP_LIVE_URL" "$WP_LIVE_USER" "$WP_LIVE_PASSWORD" +# issues: +# branch: /.*/ +# owner: weaveworks +# commands: +# - go get github.com/weaveworks/wordepress && cd /home/ubuntu/src/github.com/weaveworks/wordepress && git checkout v1.0.0 && cd cmd/wordepress && go get +# - cd $SRCDIR; PRODUCT=scope tools/publish-site "$WP_DEV_URL" "$WP_DEV_USER" "$WP_DEV_PASSWORD" diff --git a/prog/app.go b/prog/app.go index 325b5cb13..22df1e1e0 100644 --- a/prog/app.go +++ b/prog/app.go @@ -84,7 +84,8 @@ func collectorFactory(userIDer multitenant.UserIDer, collectorURL string, window } if parsed.Scheme == "dynamodb" { - dynamoCollector := multitenant.NewDynamoDBCollector(awsConfigFromURL(parsed), userIDer) + tableName := strings.TrimPrefix(parsed.Path, "/") + dynamoCollector := multitenant.NewDynamoDBCollector(awsConfigFromURL(parsed), userIDer, tableName) if createTables { if err := dynamoCollector.CreateTables(); err != nil { return nil, err @@ -107,7 +108,8 @@ func controlRouterFactory(userIDer multitenant.UserIDer, controlRouterURL string } if parsed.Scheme == "sqs" { - return multitenant.NewSQSControlRouter(awsConfigFromURL(parsed), userIDer), nil + prefix := strings.TrimPrefix(parsed.Path, "/") + return multitenant.NewSQSControlRouter(awsConfigFromURL(parsed), userIDer, prefix), nil } return nil, fmt.Errorf("Invalid control router '%s'", controlRouterURL)