Allow user to specify table name and queue prefix. (#1538)

* Allow user to specify table name and queue prefix. * Trim leading slash, catch missed queue prefix * Comment out publish step until devwww is fixed.
2026-07-28 01:31:17 +00:00 · 2016-05-25 10:09:32 +01:00
parent d13784092b
commit c8828826ae
4 changed files with 34 additions and 29 deletions
--- 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),
--- 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
--- 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"
--- 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)