Support switching the sample sink on CLI

tracker/backend/ingestor/go.mod

@@ -18,12 +18,14 @@ require (
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.34 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.28 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.35 // indirect
+	github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.26.2 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.28 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.12.12 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.12 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.19.2 // indirect
 	github.com/aws/smithy-go v1.13.5 // indirect
 	github.com/bufbuild/connect-go v1.9.0 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	golang.org/x/sys v0.1.0 // indirect
 	google.golang.org/protobuf v1.31.0 // indirect
 )

tracker/backend/ingestor/go.sum

@@ -71,6 +71,8 @@ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.28 h1:srIVS45eQuewqz6fKK
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.28/go.mod h1:7VRpKQQedkfIEXb4k52I7swUnZP0wohVajJMRn3vsUw=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.3.35 h1:LWA+3kDM8ly001vJ1X1waCuLJdtTl48gwkPKWy9sosI=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.3.35/go.mod h1:0Eg1YjxE0Bhn56lx+SHJwCzhW+2JGtizsrx+lCqrfm0=
+github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.26.2 h1:PWGu2JhCb/XJlJ7SSFJq76pxk4xWsN76nZxh7TzMHx0=
+github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.26.2/go.mod h1:2KOZkkzMDZCo/aLzPhys06mHNkiU74u85aMJA3PLRvg=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.28 h1:bkRyG4a929RCnpVSTvLM2j/T4ls015ZhhYApbmYs15s=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.28/go.mod h1:jj7znCIg05jXlaGBlFMGP8+7UN3VtCkRBG2spnmRQkU=
 github.com/aws/aws-sdk-go-v2/service/sso v1.12.12 h1:nneMBM2p79PGWBQovYO/6Xnc2ryRMw3InnDJq1FHkSY=
@@ -265,6 +267,7 @@ github.com/iris-contrib/go.uuid v2.0.0+incompatible/go.mod h1:iz2lgM/1UnEf1kP0L/
 github.com/iris-contrib/httpexpect/v2 v2.3.1/go.mod h1:ICTf89VBKSD3KB0fsyyHviKF8G8hyepP0dOXJPWz3T0=
 github.com/iris-contrib/jade v1.1.4/go.mod h1:EDqR+ur9piDl6DUgs6qRrlfzmlx/D5UybogqrXvJTBE=
 github.com/iris-contrib/schema v0.0.6/go.mod h1:iYszG0IOsuIsfzjymw1kMzTL8YQcCWlm65f3wX8J5iA=
+github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=

tracker/backend/ingestor/handler/handler.go

@@ -11,33 +11,29 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-func NewBuildReportHandler(cfg aws.Config) *BuildReportHandler {
+func NewBuildReportHandler(store SampleStorageFunc) *BuildReportHandler {
 	hdlr := &BuildReportHandler{
-		Config: cfg,
-		storeSample: func(ctx context.Context, sample PackageSample) error {
-			logrus.WithField("sample", sample).Info("package sample")
-			return nil
-		},
+		storeSample: store,
 	}
 	return hdlr
 }
 
 type BuildReportHandler struct {
 	Config      aws.Config
-	storeSample func(ctx context.Context, sample PackageSample) error
+	storeSample SampleStorageFunc
 
 	v1connect.UnimplementedReporterServiceHandler
 }
 
 // BuildFinished implements v1connect.ReporterServiceHandler
 func (handler *BuildReportHandler) BuildFinished(ctx context.Context, req *connect_go.Request[v1.BuildFinishedRequest]) (*connect_go.Response[v1.EmptyResponse], error) {
-	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.Package.Name).Info("BuildFinished")
+	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.Package.Name).Debug("BuildFinished")
 	return &connect_go.Response[v1.EmptyResponse]{Msg: &v1.EmptyResponse{}}, nil
 }
 
 // BuildStarted implements v1connect.ReporterServiceHandler
 func (handler *BuildReportHandler) BuildStarted(ctx context.Context, req *connect_go.Request[v1.BuildStartedRequest]) (*connect_go.Response[v1.EmptyResponse], error) {
-	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.Package.Name).WithField("status", req.Msg.Status).Info("BuildStarted")
+	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.Package.Name).WithField("status", req.Msg.Status).Debug("BuildStarted")
 	return &connect_go.Response[v1.EmptyResponse]{Msg: &v1.EmptyResponse{}}, nil
 }
 
@@ -56,25 +52,26 @@ func (handler *BuildReportHandler) PackageBuildFinished(ctx context.Context, req
 		Time:             time.Now(),
 		Status:           status,
 		DirtyWorkingCopy: req.Msg.Package.DirtyWorkingCopy,
+		Type:             req.Msg.Package.Type,
 	}
 	err := handler.storeSample(ctx, sample)
 	if err != nil {
 		return nil, err
 	}
 
-	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.Package.Name).WithField("dur", req.Msg.DurationMs).WithField("success", req.Msg.Error == "").Info("PackageBuildFinished")
+	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.Package.Name).WithField("dur", req.Msg.DurationMs).WithField("success", req.Msg.Error == "").Debug("PackageBuildFinished")
 	return &connect_go.Response[v1.EmptyResponse]{Msg: &v1.EmptyResponse{}}, nil
 }
 
 // PackageBuildLog implements v1connect.ReporterServiceHandler
 func (handler *BuildReportHandler) PackageBuildLog(ctx context.Context, req *connect_go.Request[v1.PackageBuildLogRequest]) (*connect_go.Response[v1.EmptyResponse], error) {
-	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.PackageName).Info("PackageBuildLog")
+	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.PackageName).Debug("PackageBuildLog")
 	return &connect_go.Response[v1.EmptyResponse]{Msg: &v1.EmptyResponse{}}, nil
 }
 
 // PackageBuildStarted implements v1connect.ReporterServiceHandler
 func (handler *BuildReportHandler) PackageBuildStarted(ctx context.Context, req *connect_go.Request[v1.PackageBuildStartedRequest]) (*connect_go.Response[v1.EmptyResponse], error) {
-	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.Package.Name).WithField("dirtyWorkingCopy", req.Msg.Package.DirtyWorkingCopy).Info("PackageBuildStarted")
+	logrus.WithField("session", req.Msg.SessionId).WithField("pkg", req.Msg.Package.Name).WithField("dirtyWorkingCopy", req.Msg.Package.DirtyWorkingCopy).Debug("PackageBuildStarted")
 	return &connect_go.Response[v1.EmptyResponse]{Msg: &v1.EmptyResponse{}}, nil
 }
 

tracker/backend/ingestor/handler/sample.go

@@ -1,10 +1,21 @@
 package handler
 
-import "time"
+import (
+	context "context"
+	"strconv"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch/types"
+	v1 "github.com/gitpod-io/leeway/pkg/remotereporter/api/gen/v1"
+	"github.com/sirupsen/logrus"
+)
 
 type PackageSample struct {
 	FullName         string
 	DirtyWorkingCopy bool
+	Type             v1.PackageType
 	Status           PackageStatus
 	Time             time.Time
 	BuildDuration    time.Duration
@@ -17,3 +28,33 @@ const (
 	PackageStatusFailed      PackageStatus = "failed"
 	PackageStatusFailedTests PackageStatus = "failed_tests"
 )
+
+type SampleStorageFunc func(ctx context.Context, sample PackageSample) error
+
+func PrintSample(ctx context.Context, sample PackageSample) error {
+	logrus.WithField("sample", sample).Info("package sample")
+	return nil
+}
+
+func PutCloudwatchMetric(cw *cloudwatch.Client) SampleStorageFunc {
+	return func(ctx context.Context, sample PackageSample) error {
+		_, err := cw.PutMetricData(ctx, &cloudwatch.PutMetricDataInput{
+			Namespace: aws.String("leeway"),
+			MetricData: []types.MetricDatum{
+				{
+					Timestamp:  aws.Time(sample.Time),
+					Value:      aws.Float64(sample.BuildDuration.Seconds()),
+					Unit:       types.StandardUnitSeconds,
+					MetricName: aws.String("package_build_duration"),
+					Dimensions: []types.Dimension{
+						{Name: aws.String("name"), Value: aws.String(sample.FullName)},
+						{Name: aws.String("status"), Value: aws.String(string(sample.Status))},
+						{Name: aws.String("type"), Value: aws.String(sample.Type.String())},
+						{Name: aws.String("dirtyWorkingCopy"), Value: aws.String(strconv.FormatBool(sample.DirtyWorkingCopy))},
+					},
+				},
+			},
+		})
+		return err
+	}
+}

tracker/backend/ingestor/main.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/aws/aws-lambda-go/lambda"
 	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch"
 	"github.com/awslabs/aws-lambda-go-api-proxy/httpadapter"
 	grpcreflect "github.com/bufbuild/connect-grpcreflect-go"
 	"github.com/sirupsen/logrus"
@@ -18,20 +19,35 @@ import (
 )
 
 var (
-	listen = flag.String("listen", ":8080", "address to listen on when not running as lambda")
+	listen     = flag.String("listen", ":8080", "address to listen on when not running as lambda")
+	verbose    = flag.Bool("verbose", false, "enable verbose logging")
+	sampleSink = flag.String("sample-sink", "console", "where to write samples to. Valid values are: console, cloudwatch")
 )
 
 func main() {
 	flag.Parse()
 
+	if *verbose {
+		logrus.SetLevel(logrus.DebugLevel)
+	}
+
 	cfg, err := config.LoadDefaultConfig(context.TODO())
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	mux := http.NewServeMux()
 
-	mux.Handle(v1connect.NewReporterServiceHandler(handler.NewBuildReportHandler(cfg)))
+	var store handler.SampleStorageFunc
+	switch *sampleSink {
+	case "console":
+		store = handler.PrintSample
+	case "cloudwatch":
+		store = handler.PutCloudwatchMetric(cloudwatch.NewFromConfig(cfg))
+	default:
+		logrus.Fatalf("unsupported --sample-sink: %s", *sampleSink)
+	}
+	mux.Handle(v1connect.NewReporterServiceHandler(handler.NewBuildReportHandler(store)))
 
 	reflector := grpcreflect.NewStaticReflector(
 		v1connect.ReporterServiceName,