Send metrics to cloudwatch

archives/archives.go

@@ -14,6 +14,8 @@ import (
 	"path/filepath"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch/types"
 	"github.com/jmoiron/sqlx"
 	"github.com/lib/pq"
 	"github.com/nyaruka/gocommon/analytics"
@@ -966,6 +968,91 @@ func ArchiveActiveOrgs(rt *runtime.Runtime) error {
 	timeTaken := dates.Now().Sub(start)
 	slog.Info("archiving of active orgs complete", "time_taken", timeTaken, "num_orgs", len(orgs))
 
+	dims := []types.Dimension{
+		{Name: aws.String("App"), Value: aws.String("archiver")},
+	}
+
+	metrics := []types.MetricDatum{
+		{
+			MetricName: aws.String("ArchiveElapsed"),
+			Dimensions: dims,
+			Value:      aws.Float64(timeTaken.Seconds()),
+			Unit:       types.StandardUnitSeconds,
+		},
+		{
+			MetricName: aws.String("OrgsArchived"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(len(orgs))),
+			Unit:       types.StandardUnitCount,
+		},
+		{
+			MetricName: aws.String("MsgsRecordsArchived"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalMsgsRecordsArchived)),
+			Unit:       types.StandardUnitCount,
+		},
+		{
+			MetricName: aws.String("MsgsArchivedsCreated"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalMsgsArchivesCreated)),
+			Unit:       types.StandardUnitCount,
+		},
+
+		{
+			MetricName: aws.String("MsgsArchivedsFailed"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalMsgsArchivesFailed)),
+			Unit:       types.StandardUnitCount,
+		},
+
+		{
+			MetricName: aws.String("MsgsRollupsCreated"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalMsgsRollupsCreated)),
+			Unit:       types.StandardUnitCount,
+		},
+		{
+			MetricName: aws.String("MsgsRollupsFailed"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalMsgsRollupsFailed)),
+			Unit:       types.StandardUnitCount,
+		},
+		{
+			MetricName: aws.String("RunsRecordsArchived"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalRunsRecordsArchived)),
+			Unit:       types.StandardUnitCount,
+		},
+		{
+			MetricName: aws.String("RunsArchivedsCreated"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalRunsArchivesCreated)),
+			Unit:       types.StandardUnitCount,
+		},
+		{
+			MetricName: aws.String("RunsArchivedsFailed"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalRunsArchivesFailed)),
+			Unit:       types.StandardUnitCount,
+		},
+		{
+			MetricName: aws.String("RunsRollupsCreated"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalRunsRollupsCreated)),
+			Unit:       types.StandardUnitCount,
+		},
+		{
+			MetricName: aws.String("RunsRollupsFailed"),
+			Dimensions: dims,
+			Value:      aws.Float64(float64(totalRunsRollupsFailed)),
+			Unit:       types.StandardUnitCount,
+		},
+	}
+
+	if err = rt.CW.Send(ctx, metrics...); err != nil {
+		slog.Error("error putting metrics", "error", err)
+	}
+
 	analytics.Gauge("archiver.archive_elapsed", timeTaken.Seconds())
 	analytics.Gauge("archiver.orgs_archived", float64(len(orgs)))
 	analytics.Gauge("archiver.msgs_records_archived", float64(totalMsgsRecordsArchived))

archives/archives_test.go

@@ -9,9 +9,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch/types"
 	"github.com/jmoiron/sqlx"
 	_ "github.com/lib/pq"
 	"github.com/nyaruka/gocommon/analytics"
+	"github.com/nyaruka/gocommon/aws/cwatch"
 	"github.com/nyaruka/gocommon/dates"
 	"github.com/nyaruka/rp-archiver/runtime"
 	"github.com/stretchr/testify/assert"
@@ -28,6 +31,7 @@ func setup(t *testing.T) (context.Context, *runtime.Runtime) {
 	config.AWSSecretAccessKey = "tembatemba"
 	config.S3Endpoint = "http://localhost:9000"
 	config.S3Minio = true
+	config.DeploymentID = "test"
 
 	testDB, err := os.ReadFile("../testdb.sql")
 	require.NoError(t, err)
@@ -43,7 +47,10 @@ func setup(t *testing.T) (context.Context, *runtime.Runtime) {
 
 	slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})))
 
-	return ctx, &runtime.Runtime{Config: config, DB: db, S3: s3Client}
+	CW, err := cwatch.NewService(config.AWSAccessKeyID, config.AWSSecretAccessKey, config.AWSRegion, config.CloudwatchNamespace, config.DeploymentID)
+	require.NoError(t, err)
+
+	return ctx, &runtime.Runtime{Config: config, DB: db, S3: s3Client, CW: CW}
 }
 
 func TestGetMissingDayArchives(t *testing.T) {
@@ -491,7 +498,14 @@ func TestArchiveOrgRuns(t *testing.T) {
 }
 
 func TestArchiveActiveOrgs(t *testing.T) {
-	_, rt := setup(t)
+	ctx, rt := setup(t)
+
+	err := rt.CW.Send(ctx,
+		types.MetricDatum{MetricName: aws.String("NumGoats"), Value: aws.Float64(10), Unit: types.StandardUnitCount},
+		types.MetricDatum{MetricName: aws.String("NumSheep"), Dimensions: []types.Dimension{{Name: aws.String("Host"), Value: aws.String("foo1")}}, Value: aws.Float64(20), Unit: types.StandardUnitCount},
+	)
+
+	assert.NoError(t, err)
 
 	mockAnalytics := analytics.NewMock()
 	analytics.RegisterBackend(mockAnalytics)
@@ -500,7 +514,7 @@ func TestArchiveActiveOrgs(t *testing.T) {
 	dates.SetNowFunc(dates.NewSequentialNow(time.Date(2018, 1, 8, 12, 30, 0, 0, time.UTC), time.Second))
 	defer dates.SetNowFunc(time.Now)
 
-	err := ArchiveActiveOrgs(rt)
+	err = ArchiveActiveOrgs(rt)
 	assert.NoError(t, err)
 
 	assert.Equal(t, map[string][]float64{
@@ -519,4 +533,5 @@ func TestArchiveActiveOrgs(t *testing.T) {
 	}, mockAnalytics.Gauges)
 
 	analytics.Stop()
+
 }

cmd/rp-archiver/main.go

@@ -13,6 +13,7 @@ import (
 	_ "github.com/lib/pq"
 	"github.com/nyaruka/ezconf"
 	"github.com/nyaruka/gocommon/analytics"
+	"github.com/nyaruka/gocommon/aws/cwatch"
 	"github.com/nyaruka/gocommon/dates"
 	"github.com/nyaruka/rp-archiver/archives"
 	"github.com/nyaruka/rp-archiver/runtime"
@@ -126,6 +127,13 @@ func main() {
 		analytics.RegisterBackend(analytics.NewLibrato(config.LibratoUsername, config.LibratoToken, config.InstanceName, time.Second, wg))
 	}
 
+	rt.CW, err = cwatch.NewService(config.AWSAccessKeyID, config.AWSSecretAccessKey, config.AWSRegion, config.CloudwatchNamespace, config.DeploymentID)
+	if err != nil {
+		logger.Error("unable to create cloudwatch service", "error", err)
+	} else {
+		logger.Info("cloudwatch service ok", "state", "starting")
+	}
+
 	analytics.Start()
 
 	if config.Once {

go.mod

@@ -1,51 +1,56 @@
 module github.com/nyaruka/rp-archiver
 
-go 1.22
+go 1.23
+
+toolchain go1.23.1
 
 require (
-	github.com/aws/aws-sdk-go-v2 v1.30.4
+	github.com/aws/aws-sdk-go-v2 v1.32.6
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.12
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.60.0
-	github.com/aws/smithy-go v1.20.4
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.69.0
+	github.com/aws/smithy-go v1.22.1
 	github.com/getsentry/sentry-go v0.28.1
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/lib/pq v1.10.9
 	github.com/nyaruka/ezconf v0.3.0
-	github.com/nyaruka/gocommon v1.58.0
+	github.com/nyaruka/gocommon v1.60.3
 	github.com/samber/slog-multi v1.2.0
 	github.com/samber/slog-sentry v1.2.2
-	github.com/stretchr/testify v1.9.0
+	github.com/stretchr/testify v1.10.0
 )
 
 require (
-	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.4 // indirect
-	github.com/aws/aws-sdk-go-v2/config v1.27.28 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.28 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.12 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.16 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.16 // indirect
+	github.com/aws/aws-sdk-go v1.55.5
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.7 // indirect
+	github.com/aws/aws-sdk-go-v2/config v1.28.5 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.46 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.20 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.25 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.25 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.16 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.4 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.3.18 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.18 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.17.16 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.22.5 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.5 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.30.4 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.24 // indirect
+	github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.43.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.24.6 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.33.1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fatih/structs v1.1.0 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/naoina/go-stringutil v0.1.0 // indirect
 	github.com/naoina/toml v0.1.1 // indirect
 	github.com/nyaruka/librato v1.1.1 // indirect
 	github.com/nyaruka/null/v2 v2.0.3 // indirect
-	github.com/nyaruka/phonenumbers v1.4.0 // indirect
+	github.com/nyaruka/phonenumbers v1.4.3 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/samber/lo v1.46.0 // indirect
-	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
-	google.golang.org/protobuf v1.34.2 // indirect
+	golang.org/x/exp v0.0.0-20241108190413-2d47ceb2692f // indirect
+	golang.org/x/sys v0.27.0 // indirect
+	golang.org/x/text v0.20.0 // indirect
+	google.golang.org/protobuf v1.35.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )