Add support for publishing to an AWS SQS

.gitignore

@@ -1,2 +1,3 @@
 /vendor
 /dist
+/private

cmd/start.go

@@ -24,4 +24,6 @@ func init() {
 	viper.BindPFlag("api.port", startCmd.PersistentFlags().Lookup("port"))
 	viper.BindPFlag("api.address", startCmd.PersistentFlags().Lookup("address"))
 	viper.SetDefault("api.admin-password", "changeme")
+	viper.SetDefault("aws.access-key-id", "")
+	viper.SetDefault("aws.secret-access-key", "")
 }

collector/aws/aws.go

@@ -0,0 +1,26 @@
+package aws
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+)
+
+const (
+	// Region is the default aws region
+	Region = "us-east-2"
+	// MaxRetries is the number of retries when connecting to aws
+	MaxRetries = 5
+)
+
+// Session is a global session handle
+var Session *session.Session
+
+// NewSession creates a new aws session
+func NewSession(accessKeyID, secretAccessKey string) *session.Session {
+	return session.New(&aws.Config{
+		Region:      aws.String(Region),
+		Credentials: credentials.NewStaticCredentials(accessKeyID, secretAccessKey, ""),
+		MaxRetries:  aws.Int(MaxRetries),
+	})
+}

collector/aws/sqs.go

@@ -0,0 +1,35 @@
+package aws
+
+import (
+	"errors"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/sqs"
+)
+
+const (
+	// QueueURL is the URL of the message queue
+	queueURL = "https://sqs.us-east-2.amazonaws.com/082866812839/ooni-collector.fifo"
+)
+
+// SendMessage sends a message to the AWS SQS queue
+func SendMessage(sess *session.Session, body string, groupID string) (*string, error) {
+	if sess == nil {
+		return nil, errors.New("invalid aws Session")
+	}
+
+	svc := sqs.New(sess)
+	sendParams := &sqs.SendMessageInput{
+		MessageBody:    aws.String(body),
+		QueueUrl:       aws.String(queueURL),
+		MessageGroupId: aws.String(groupID),
+	}
+	sendResp, err := svc.SendMessage(sendParams)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return sendResp.MessageId, nil
+}

collector/collector.go

@@ -6,6 +6,7 @@ import (
 	"os"
 
 	"github.com/ooni/collector/collector/api/v1"
+	"github.com/ooni/collector/collector/aws"
 	"github.com/ooni/collector/collector/middleware"
 	"github.com/ooni/collector/collector/paths"
 	"github.com/ooni/collector/collector/report"
@@ -39,6 +40,16 @@ func initDataRoot() error {
 	return nil
 }
 
+func initAWS() error {
+	accessKeyID := viper.GetString("aws.access-key-id")
+	secretAccessKey := viper.GetString("aws.secret-access-key")
+	if accessKeyID == "" {
+		return nil
+	}
+	aws.Session = aws.NewSession(accessKeyID, secretAccessKey)
+	return nil
+}
+
 // Start the collector server
 func Start() {
 	var (
@@ -56,6 +67,10 @@ func Start() {
 	if err = initDataRoot(); err != nil {
 		log.WithError(err).Error("failed to init data root")
 	}
+	if err = initAWS(); err != nil {
+		log.WithError(err).Error("failed to init aws")
+	}
+
 	store := storage.New(paths.BadgerDir())
 	storageMw, err := middleware.InitStorageMiddleware(store)
 	if err != nil {

collector/report/report.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/apex/log"
+	"github.com/ooni/collector/collector/aws"
 	"github.com/ooni/collector/collector/info"
 	"github.com/ooni/collector/collector/paths"
 	"github.com/ooni/collector/collector/storage"
@@ -137,12 +138,23 @@ func CloseReport(store *storage.Storage, reportID string) error {
 		// There is no need to keep closed empty reports
 		os.Remove(meta.ReportFilePath)
 	}
+	meta.ReportFilePath = dstPath
 	meta.Closed = true
 	expiryTimers[reportID].Stop()
 
 	if err = store.SetReport(meta); err != nil {
 		return err
 	}
+	value, err := json.Marshal(meta)
+	if err != nil {
+		log.WithError(err).Error("failed to serialize meta")
+		return nil
+	}
+	_, err = aws.SendMessage(aws.Session, string(value), "report")
+	if err != nil {
+		log.WithError(err).Error("failed to publish to aws SQS")
+		return nil
+	}
 
 	return nil
 }