chore(pgo): regenerate default.pgo with broad service load and benchmark profiles

cmd/pgoload/clients.go

@@ -0,0 +1,163 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awscfg "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/aws/aws-sdk-go-v2/service/ec2"
+	"github.com/aws/aws-sdk-go-v2/service/eventbridge"
+	"github.com/aws/aws-sdk-go-v2/service/iam"
+	"github.com/aws/aws-sdk-go-v2/service/kinesis"
+	"github.com/aws/aws-sdk-go-v2/service/kms"
+	"github.com/aws/aws-sdk-go-v2/service/lambda"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/secretsmanager"
+	"github.com/aws/aws-sdk-go-v2/service/sfn"
+	"github.com/aws/aws-sdk-go-v2/service/sns"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/service/ssm"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
+)
+
+// clients bundles one SDK client per service exercised by pgoload, all
+// pointed at the same Gopherstack endpoint.
+type clients struct {
+	ddb            *dynamodb.Client
+	s3             *s3.Client
+	sqs            *sqs.Client
+	sns            *sns.Client
+	kinesis        *kinesis.Client
+	iam            *iam.Client
+	sts            *sts.Client
+	ssm            *ssm.Client
+	secretsmanager *secretsmanager.Client
+	cloudwatch     *cloudwatch.Client
+	logs           *cloudwatchlogs.Client
+	ec2            *ec2.Client
+	lambda         *lambda.Client
+	kms            *kms.Client
+	eventbridge    *eventbridge.Client
+	stepfunctions  *sfn.Client
+}
+
+// buildClients constructs every service client used by pgoload, using
+// static test credentials against cfg.endpoint.
+func buildClients(ctx context.Context, cfg config) (*clients, error) {
+	awsCfg, err := awscfg.LoadDefaultConfig(ctx,
+		awscfg.WithRegion(awsRegion),
+		awscfg.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("load aws config: %w", err)
+	}
+
+	ep := aws.String(cfg.endpoint)
+
+	return &clients{
+		ddb: dynamodb.NewFromConfig(awsCfg, func(o *dynamodb.Options) { o.BaseEndpoint = ep }),
+		s3: s3.NewFromConfig(awsCfg, func(o *s3.Options) {
+			o.BaseEndpoint = ep
+			o.UsePathStyle = true
+		}),
+		sqs:            sqs.NewFromConfig(awsCfg, func(o *sqs.Options) { o.BaseEndpoint = ep }),
+		sns:            sns.NewFromConfig(awsCfg, func(o *sns.Options) { o.BaseEndpoint = ep }),
+		kinesis:        kinesis.NewFromConfig(awsCfg, func(o *kinesis.Options) { o.BaseEndpoint = ep }),
+		iam:            iam.NewFromConfig(awsCfg, func(o *iam.Options) { o.BaseEndpoint = ep }),
+		sts:            sts.NewFromConfig(awsCfg, func(o *sts.Options) { o.BaseEndpoint = ep }),
+		ssm:            ssm.NewFromConfig(awsCfg, func(o *ssm.Options) { o.BaseEndpoint = ep }),
+		secretsmanager: secretsmanager.NewFromConfig(awsCfg, func(o *secretsmanager.Options) { o.BaseEndpoint = ep }),
+		cloudwatch:     cloudwatch.NewFromConfig(awsCfg, func(o *cloudwatch.Options) { o.BaseEndpoint = ep }),
+		logs:           cloudwatchlogs.NewFromConfig(awsCfg, func(o *cloudwatchlogs.Options) { o.BaseEndpoint = ep }),
+		ec2:            ec2.NewFromConfig(awsCfg, func(o *ec2.Options) { o.BaseEndpoint = ep }),
+		lambda:         lambda.NewFromConfig(awsCfg, func(o *lambda.Options) { o.BaseEndpoint = ep }),
+		kms:            kms.NewFromConfig(awsCfg, func(o *kms.Options) { o.BaseEndpoint = ep }),
+		eventbridge:    eventbridge.NewFromConfig(awsCfg, func(o *eventbridge.Options) { o.BaseEndpoint = ep }),
+		stepfunctions:  sfn.NewFromConfig(awsCfg, func(o *sfn.Options) { o.BaseEndpoint = ep }),
+	}, nil
+}
+
+// resources holds the identifiers of everything provisioned once during
+// setup and then reused by every breadth-scenario worker.
+type resources struct {
+	roleArn         string
+	queueURL        string
+	queueArn        string
+	topicArn        string
+	kinesisShardID  string
+	kmsKeyID        string
+	functionName    string
+	stateMachineArn string
+	s3Buckets       []string
+}
+
+// setupResources provisions the DynamoDB table, S3 buckets, and every
+// breadth-scenario's resources, bounded by setupTimeout so a short
+// -duration still leaves time for the load phase itself. Each service's
+// setup is independent; a failure in one is fatal (unlike op-level errors
+// during the load phase, which are only counted) because a missing
+// resource would make that whole scenario error-spin instead of exercising
+// its real hot paths.
+func setupResources(ctx context.Context, cls *clients, log *slog.Logger) (*resources, error) {
+	setupCtx, cancel := context.WithTimeout(ctx, setupTimeout)
+	defer cancel()
+
+	if err := ensureDDBTable(setupCtx, cls.ddb, log); err != nil {
+		return nil, fmt.Errorf("ddb table setup: %w", err)
+	}
+
+	buckets, err := ensureS3Buckets(setupCtx, cls.s3, log)
+	if err != nil {
+		return nil, fmt.Errorf("s3 bucket setup: %w", err)
+	}
+
+	res := &resources{s3Buckets: buckets}
+
+	if res.roleArn, err = ensureIAMRole(setupCtx, cls.iam, log); err != nil {
+		return nil, fmt.Errorf("iam role setup: %w", err)
+	}
+
+	if res.queueURL, res.queueArn, err = ensureSQSQueue(setupCtx, cls.sqs, log); err != nil {
+		return nil, fmt.Errorf("sqs queue setup: %w", err)
+	}
+
+	if res.topicArn, err = ensureSNSTopic(setupCtx, cls.sns, res.queueArn, log); err != nil {
+		return nil, fmt.Errorf("sns topic setup: %w", err)
+	}
+
+	if err = ensureSecrets(setupCtx, cls.secretsmanager, log); err != nil {
+		return nil, fmt.Errorf("secretsmanager setup: %w", err)
+	}
+
+	if res.kinesisShardID, err = ensureKinesisStream(setupCtx, cls.kinesis, log); err != nil {
+		return nil, fmt.Errorf("kinesis stream setup: %w", err)
+	}
+
+	if err = ensureLogGroup(setupCtx, cls.logs, log); err != nil {
+		return nil, fmt.Errorf("logs group setup: %w", err)
+	}
+
+	if res.kmsKeyID, err = ensureKMSKey(setupCtx, cls.kms, log); err != nil {
+		return nil, fmt.Errorf("kms key setup: %w", err)
+	}
+
+	if res.functionName, err = ensureLambdaFunction(setupCtx, cls.lambda, res.roleArn, log); err != nil {
+		return nil, fmt.Errorf("lambda function setup: %w", err)
+	}
+
+	if err = ensureEventBridgeRule(setupCtx, cls.eventbridge, res.queueArn, log); err != nil {
+		return nil, fmt.Errorf("eventbridge rule setup: %w", err)
+	}
+
+	if res.stateMachineArn, err = ensureStateMachine(setupCtx, cls.stepfunctions, res.roleArn, log); err != nil {
+		return nil, fmt.Errorf("step functions state machine setup: %w", err)
+	}
+
+	return res, nil
+}

cmd/pgoload/cloudwatch.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"strconv"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch"
+	cwtypes "github.com/aws/aws-sdk-go-v2/service/cloudwatch/types"
+)
+
+const (
+	cwNamespace          = "pgoload"
+	cwMetricName         = "Ops"
+	cwStatWindow         = 10 * time.Minute
+	cwStatPeriodSeconds  = 60
+	cwDatapointsPerBatch = 3
+)
+
+// cloudwatchWorker repeatedly runs a mix of CloudWatch operations, staggered
+// by workerID, until ctx is done.
+//
+//nolint:dupl // structurally mirrors eventBridgeWorker's op-table shape but wires an unrelated service
+func cloudwatchWorker(ctx context.Context, cl *cloudwatch.Client, workerID int, c *opCounter, log *slog.Logger) {
+	ops := []opFunc{
+		func(ctx context.Context, workerID, i int) error { return cwPutMetricDataOp(ctx, cl, workerID, i) },
+		func(ctx context.Context, workerID, i int) error { return cwPutMetricDataOp(ctx, cl, workerID, i) },
+		func(ctx context.Context, _, _ int) error { return cwListMetricsOp(ctx, cl) },
+		func(ctx context.Context, _, _ int) error { return cwGetMetricStatisticsOp(ctx, cl) },
+		func(ctx context.Context, _, _ int) error { return cwDescribeAlarmsOp(ctx, cl) },
+	}
+
+	runOpLoop(ctx, workerID, ops, c, "cloudwatch", log)
+}
+
+func cwPutMetricDataOp(ctx context.Context, cl *cloudwatch.Client, workerID, i int) error {
+	data := make([]cwtypes.MetricDatum, 0, cwDatapointsPerBatch)
+	for j := range cwDatapointsPerBatch {
+		data = append(data, cwtypes.MetricDatum{
+			MetricName: aws.String(cwMetricName),
+			Value:      aws.Float64(float64(i + j)),
+			Dimensions: []cwtypes.Dimension{
+				{Name: aws.String("Worker"), Value: aws.String(strconv.Itoa(workerID))},
+			},
+		})
+	}
+
+	_, err := cl.PutMetricData(ctx, &cloudwatch.PutMetricDataInput{
+		Namespace:  aws.String(cwNamespace),
+		MetricData: data,
+	})
+
+	return err
+}
+
+func cwListMetricsOp(ctx context.Context, cl *cloudwatch.Client) error {
+	_, err := cl.ListMetrics(ctx, &cloudwatch.ListMetricsInput{Namespace: aws.String(cwNamespace)})
+
+	return err
+}
+
+func cwGetMetricStatisticsOp(ctx context.Context, cl *cloudwatch.Client) error {
+	now := time.Now()
+
+	_, err := cl.GetMetricStatistics(ctx, &cloudwatch.GetMetricStatisticsInput{
+		Namespace:  aws.String(cwNamespace),
+		MetricName: aws.String(cwMetricName),
+		StartTime:  aws.Time(now.Add(-cwStatWindow)),
+		EndTime:    aws.Time(now),
+		Period:     aws.Int32(cwStatPeriodSeconds),
+		Statistics: []cwtypes.Statistic{cwtypes.StatisticSum, cwtypes.StatisticAverage},
+	})
+
+	return err
+}
+
+func cwDescribeAlarmsOp(ctx context.Context, cl *cloudwatch.Client) error {
+	_, err := cl.DescribeAlarms(ctx, &cloudwatch.DescribeAlarmsInput{})
+
+	return err
+}

cmd/pgoload/common.go

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"sync/atomic"
+)
+
+// opCounter tracks operation and error totals for one breadth-load scenario.
+type opCounter struct {
+	ops    atomic.Int64
+	errors atomic.Int64
+}
+
+func (o *opCounter) record(err error) {
+	if err != nil {
+		o.errors.Add(1)
+
+		return
+	}
+
+	o.ops.Add(1)
+}
+
+// opFunc is one operation exercised by a breadth-scenario worker. workerID
+// and i together select the resource/key space touched by the call.
+type opFunc func(ctx context.Context, workerID, i int) error
+
+// runOpLoop repeatedly runs a mix of operations, staggered by workerID so
+// concurrent workers hit different operations on each tick, until ctx is
+// done. It is the shared loop shape behind every breadth scenario added
+// beyond the original DDB/S3 load (ddbWorker and s3Worker keep their own
+// copy of this loop untouched).
+func runOpLoop(ctx context.Context, workerID int, ops []opFunc, c *opCounter, scenario string, log *slog.Logger) {
+	for i := 0; ; i++ {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		err := ops[(workerID+i)%len(ops)](ctx, workerID, i)
+		c.record(err)
+
+		if err != nil {
+			log.WarnContext(ctx, scenario+" operation failed", "worker", workerID, "iter", i, "error", err)
+		}
+	}
+}

cmd/pgoload/ec2.go

@@ -0,0 +1,100 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/ec2"
+	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
+)
+
+const (
+	ec2ImageID      = "ami-12345678"
+	ec2TagKey       = "Owner"
+	ec2TagValue     = "pgoload"
+	ec2TerminateCap = 5
+)
+
+// ec2Worker repeatedly runs a mix of EC2 operations, staggered by workerID,
+// until ctx is done. Instance churn is self-balancing: run and terminate are
+// both in the op mix, and terminate always clears out whatever the pgoload
+// tag currently has, so the instance count stays bounded without any
+// client-side bookkeeping across calls.
+func ec2Worker(ctx context.Context, cl *ec2.Client, workerID int, c *opCounter, log *slog.Logger) {
+	ops := []opFunc{
+		func(ctx context.Context, _, _ int) error { return ec2RunInstancesOp(ctx, cl) },
+		func(ctx context.Context, _, _ int) error { return ec2DescribeInstancesOp(ctx, cl) },
+		func(ctx context.Context, _, _ int) error { return ec2DescribeInstancesOp(ctx, cl) },
+		func(ctx context.Context, _, _ int) error { return ec2DescribeSecurityGroupsOp(ctx, cl) },
+		func(ctx context.Context, _, _ int) error { return ec2TerminateInstancesOp(ctx, cl) },
+	}
+
+	runOpLoop(ctx, workerID, ops, c, "ec2", log)
+}
+
+func ec2OwnerFilter() []ec2types.Filter {
+	return []ec2types.Filter{
+		{Name: aws.String("tag:" + ec2TagKey), Values: []string{ec2TagValue}},
+	}
+}
+
+func ec2RunInstancesOp(ctx context.Context, cl *ec2.Client) error {
+	const minMaxCount = 1
+
+	_, err := cl.RunInstances(ctx, &ec2.RunInstancesInput{
+		ImageId:      aws.String(ec2ImageID),
+		InstanceType: ec2types.InstanceTypeT2Micro,
+		MinCount:     aws.Int32(minMaxCount),
+		MaxCount:     aws.Int32(minMaxCount),
+		TagSpecifications: []ec2types.TagSpecification{
+			{
+				ResourceType: ec2types.ResourceTypeInstance,
+				Tags:         []ec2types.Tag{{Key: aws.String(ec2TagKey), Value: aws.String(ec2TagValue)}},
+			},
+		},
+	})
+
+	return err
+}
+
+func ec2DescribeInstancesOp(ctx context.Context, cl *ec2.Client) error {
+	_, err := cl.DescribeInstances(ctx, &ec2.DescribeInstancesInput{Filters: ec2OwnerFilter()})
+
+	return err
+}
+
+func ec2DescribeSecurityGroupsOp(ctx context.Context, cl *ec2.Client) error {
+	_, err := cl.DescribeSecurityGroups(ctx, &ec2.DescribeSecurityGroupsInput{})
+
+	return err
+}
+
+// ec2TerminateInstancesOp describes up to ec2TerminateCap pgoload-tagged
+// instances and terminates them, keeping the emulator's instance count from
+// growing without bound over a long run.
+func ec2TerminateInstancesOp(ctx context.Context, cl *ec2.Client) error {
+	out, err := cl.DescribeInstances(ctx, &ec2.DescribeInstancesInput{
+		Filters:    ec2OwnerFilter(),
+		MaxResults: aws.Int32(ec2TerminateCap),
+	})
+	if err != nil {
+		return err
+	}
+
+	ids := make([]string, 0, ec2TerminateCap)
+
+	for _, res := range out.Reservations {
+		for _, inst := range res.Instances {
+			ids = append(ids, aws.ToString(inst.InstanceId))
+		}
+	}
+
+	if len(ids) == 0 {
+		return nil
+	}
+
+	_, err = cl.TerminateInstances(ctx, &ec2.TerminateInstancesInput{InstanceIds: ids})
+
+	return err
+}