Revert to using remote subscription and service account

Author: gfr10598

cmd/etl_worker/app-ndt.yaml

@@ -42,6 +42,8 @@ env_variables:
   BIGQUERY_DATASET: 'mlab_sandbox'
   ANNOTATE_IP: 'true'
   # Location of subscription, and service account key.
-  SUBSCRIPTION_PROJECT: 'mlab-sandbox'
+  SUBSCRIPTION_PROJECT: 'mlab-oti'
   SUBSCRIPTION_NAME: 'sandbox-pipeline'
+  SUBSCRIPTION_KEY_BUCKET: 'gs://keys-mlab-sandbox/'
+  SUBSCRIPTION_KEY_FILE: 'pubsub-reader.mlab-oti.json'
   # TODO add custom service-account, instead of using default credentials.

cmd/etl_worker/etl_worker.go

@@ -8,11 +8,14 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"os/exec"
 	"runtime"
 	"strconv"
 	"sync/atomic"
 	"time"
 
+	"google.golang.org/api/option"
+
 	"cloud.google.com/go/pubsub"
 	"golang.org/x/net/context"
 
@@ -255,12 +258,22 @@ func setMaxInFlight() {
 }
 
 func runPubSubHandler() error {
+	keybucket := os.Getenv("SUBSCRIPTION_KEY_BUCKET")
+	keyfile := os.Getenv("SUBSCRIPTION_KEY_FILE")
+	cmd := exec.Command("gsutil", "cp", keybucket+keyfile, "/tmp/"+keyfile)
+	stdoutStderr, err := cmd.CombinedOutput()
+	if err != nil {
+		log.Println(stdoutStderr)
+		return err
+	}
+
 	ctx := context.Background()
+	opt := option.WithServiceAccountFile("/tmp/" + keyfile)
 	// Must use the project where the subscription resides, or else
 	// we can't find it.
 	proj := os.Getenv("SUBSCRIPTION_PROJECT")
 	subscription := os.Getenv("SUBSCRIPTION_NAME")
-	client, err := pubsub.NewClient(ctx, proj)
+	client, err := pubsub.NewClient(ctx, proj, opt)
 
 	if err != nil {
 		return err
@@ -305,6 +318,7 @@ func runPubSubHandler() error {
 			log.Println(outcome)
 			// TODO(gfr) Remove once we it looks ok.
 			msg.Nack()
+			log.Println("Warning - Nack and sleep 30 seconds")
 			time.Sleep(30 * time.Second)
 		} else {
 			msg.Ack()