move cron cloudwatch log name watcher up near cron

cmd/cron/cw_id_watcher.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"context"
+	"time"
+
+	"github.com/ReconfigureIO/platform/models"
+	awsaws "github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/batch"
+	"github.com/aws/aws-sdk-go/service/batch/batchiface"
+	log "github.com/sirupsen/logrus"
+)
+
+type LogWatcher struct {
+	batchRepo    models.BatchRepo
+	batchSession batchiface.BatchAPI
+}
+
+func NewLogWatcher(batchRepo models.BatchRepo, batchSession batchiface.BatchAPI) *LogWatcher {
+	w := LogWatcher{
+		batchRepo:    batchRepo,
+		batchSession: batchSession,
+	}
+	return &w
+}
+
+func (watcher *LogWatcher) FindLogNames(ctx context.Context, limit int, sinceTime time.Time) error {
+	batchJobs, err := watcher.batchRepo.ActiveJobsWithoutLogs(sinceTime)
+
+	if len(batchJobs) == 0 {
+		return nil
+	}
+
+	batchJobIDs := []string{}
+	for _, returnedBatchJob := range batchJobs {
+		batchJobIDs = append(batchJobIDs, returnedBatchJob.BatchID)
+	}
+
+	LogNames, err := watcher.getLogNames(ctx, batchJobIDs)
+	if err != nil {
+		log.WithError(err).
+			WithFields(log.Fields{"batch_job_ids": batchJobIDs}).
+			Error("Couldn't get cw log names for batch jobs")
+		return err
+	}
+
+	for _, jobID := range batchJobIDs {
+		logName, found := LogNames[jobID]
+		if found {
+			err := watcher.batchRepo.SetLogName(jobID, logName)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (watcher *LogWatcher) getLogNames(ctx context.Context, batchJobIDs []string) (map[string]string, error) {
+	ret := make(map[string]string)
+
+	cfg := batch.DescribeJobsInput{
+		Jobs: awsaws.StringSlice(batchJobIDs),
+	}
+
+	results, err := watcher.batchSession.DescribeJobsWithContext(ctx, &cfg)
+	if err != nil {
+		return ret, err
+	}
+
+	for _, job := range results.Jobs {
+		ret[*job.JobId] = *job.Container.LogStreamName
+	}
+
+	return ret, nil
+}

cmd/cron/cw_id_watcher_test.go

@@ -0,0 +1,83 @@
+// +build integration
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/ReconfigureIO/platform/models"
+	awsaws "github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/request"
+	awsbatch "github.com/aws/aws-sdk-go/service/batch"
+	"github.com/aws/aws-sdk-go/service/batch/batchiface"
+	"github.com/golang/mock/gomock"
+)
+
+func TestFindLogNames(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	repo := models.NewMockBatchRepo(mockCtrl)
+
+	watcher := NewLogWatcher(repo, fakeBatchClient{})
+
+	batchJobs := []models.BatchJob{
+		models.BatchJob{
+			ID:      123456789,
+			BatchID: "foobar",
+			Events: []models.BatchJobEvent{
+				models.BatchJobEvent{
+					Timestamp: time.Unix(20, 0),
+					Status:    "STARTED",
+				},
+				models.BatchJobEvent{
+					Timestamp: time.Unix(0, 0),
+					Status:    "QUEUED",
+				},
+			},
+		},
+	}
+	LogNames := map[string]string{batchJobs[0].BatchID: "LogName"}
+
+	ctx := context.Background()
+	limit := 100
+	sinceTime := time.Unix(0, 0)
+
+	repo.EXPECT().ActiveJobsWithoutLogs(sinceTime).Return(batchJobs, nil)
+	repo.EXPECT().SetLogName(batchJobs[0].BatchID, LogNames[batchJobs[0].BatchID]).Return(nil)
+
+	err := watcher.FindLogNames(ctx, limit, sinceTime)
+
+	if err != nil {
+		t.Error(err)
+	}
+}
+
+type fakeBatchClient struct {
+	batchiface.BatchAPI
+}
+
+func (
+	batch fakeBatchClient,
+) DescribeJobsWithContext(
+	ctx awsaws.Context,
+	req *awsbatch.DescribeJobsInput,
+	opts ...request.Option,
+) (
+	*awsbatch.DescribeJobsOutput,
+	error,
+) {
+	return &awsbatch.DescribeJobsOutput{
+		Jobs: []*awsbatch.JobDetail{
+			&awsbatch.JobDetail{
+				JobId: awsaws.String("foobar"),
+				Container: &awsbatch.ContainerDetail{
+					LogStreamName: awsaws.String("LogName"),
+				},
+			},
+		},
+	}, nil
+}

cmd/cron/main.go

@@ -8,12 +8,15 @@ import (
 	"github.com/ReconfigureIO/platform/config"
 	"github.com/ReconfigureIO/platform/handlers/api"
 	"github.com/ReconfigureIO/platform/models"
-	"github.com/ReconfigureIO/platform/service/aws"
+	"github.com/ReconfigureIO/platform/service/batch/aws"
+	"github.com/ReconfigureIO/platform/service/batch/aws/logs/cloudwatch"
 	"github.com/ReconfigureIO/platform/service/billing_hours"
-	"github.com/ReconfigureIO/platform/service/cw_id_watcher"
 	"github.com/ReconfigureIO/platform/service/deployment"
 	"github.com/ReconfigureIO/platform/service/fpgaimage/afi"
 	"github.com/ReconfigureIO/platform/service/fpgaimage/afi/afiwatcher"
+	awsaws "github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/batch"
 	"github.com/jinzhu/gorm"
 	_ "github.com/jinzhu/gorm/dialects/postgres"
 	"github.com/robfig/cron"
@@ -23,8 +26,9 @@ import (
 )
 
 var (
-	deploy     deployment.Service
-	awsService aws.Service
+	deploy      deployment.Service
+	awsService  aws.Service
+	batchClient *batch.Batch
 
 	db *gorm.DB
 
@@ -50,7 +54,11 @@ func setup(*cobra.Command, []string) {
 	}
 
 	deploy = deployment.New(conf.Reco.Deploy)
-	awsService = aws.New(conf.Reco.AWS)
+	awsService = *aws.New(conf.Reco.AWS, &cloudwatch.Service{
+		LogGroup: conf.Reco.AWS.LogGroup,
+	})
+
+	batchClient = batch.New(session.Must(session.NewSession(awsaws.NewConfig().WithRegion("us-east-1"))))
 
 	db = config.SetupDB(conf)
 	api.DB(db)
@@ -149,7 +157,7 @@ func generatedAFIs() {
 
 func getBatchJobLogNames() {
 	log.Printf("Getting log names")
-	watcher := cw_id_watcher.NewLogWatcher(awsService, models.BatchDataSource(db))
+	watcher := NewLogWatcher(models.BatchDataSource(db), batchClient)
 
 	// find batch jobs that've become active in the last hour
 	sinceTime := time.Now().Add(-1 * time.Hour)