diff --git a/Makefile b/Makefile index e2836fd5a6..397f24982b 100644 --- a/Makefile +++ b/Makefile @@ -251,8 +251,8 @@ build-linux: fi .PHONY: build-linux -# Auto-detect platform: use arm64 on ARM Macs, amd64 elsewhere -DOCKER_PLATFORM ?= $(shell if [ "$$(uname -m)" = "arm64" ]; then echo "linux/arm64"; else echo "linux/amd64"; fi) +# Auto-detect platform: arm64 on ARM Macs and Graviton (aarch64), amd64 elsewhere +DOCKER_PLATFORM ?= $(shell if [ "$$(uname -m)" = "arm64" ] || [ "$$(uname -m)" = "aarch64" ]; then echo "linux/arm64"; else echo "linux/amd64"; fi) export DOCKER_PLATFORM # Build docker image for detected platform @@ -542,6 +542,41 @@ docker-cluster-stop-monitoring: @cd docker && DOCKER_PLATFORM=$(DOCKER_PLATFORM) USERID=$(shell id -u) GROUPID=$(shell id -g) GOCACHE=$(shell go env GOCACHE) docker compose -f docker-compose.yml -f docker-compose.monitoring.yml down .PHONY: docker-cluster-stop-monitoring +# One Autobahn validator per AWS host. VALIDATOR_HOME is the persisted seid +# directory and must live outside build/ so step0's `make clean` cannot delete it. +AWS_VALIDATOR_COMPOSE = docker compose -f docker-compose.aws-validator.yml +VALIDATOR_HOME ?= $(HOME)/.sei-autobahn-e2e-home + +docker-aws-validator-init: + @mkdir -p $(VALIDATOR_HOME) + @cd docker && $(CLUSTER_ENV_VARS) VALIDATOR_HOME=$(VALIDATOR_HOME) ID=$(ID) ADVERTISE_IP=$(ADVERTISE_IP) AUTOBAHN_E2E_PHASE=init SKIP_BUILD=$(SKIP_BUILD) \ + $(AWS_VALIDATOR_COMPOSE) run --rm --no-deps node +.PHONY: docker-aws-validator-init + +docker-aws-validator-genesis: + @mkdir -p $(VALIDATOR_HOME) + @cd docker && $(CLUSTER_ENV_VARS) VALIDATOR_HOME=$(VALIDATOR_HOME) ID=0 AUTOBAHN_E2E_PHASE=genesis SKIP_BUILD=true \ + $(AWS_VALIDATOR_COMPOSE) run --rm --no-deps node +.PHONY: docker-aws-validator-genesis + +docker-aws-validator-start: + @mkdir -p $(VALIDATOR_HOME) + @cd docker && $(CLUSTER_ENV_VARS) VALIDATOR_HOME=$(VALIDATOR_HOME) ID=$(ID) ADVERTISE_IP=$(ADVERTISE_IP) AUTOBAHN_E2E_PHASE=start SKIP_BUILD=true \ + $(AWS_VALIDATOR_COMPOSE) up -d +.PHONY: docker-aws-validator-start + +docker-aws-validator-stop: + @cd docker && $(CLUSTER_ENV_VARS) VALIDATOR_HOME=$(VALIDATOR_HOME) $(AWS_VALIDATOR_COMPOSE) down +.PHONY: docker-aws-validator-stop + +docker-aws-load-start: + @cd docker && $(CLUSTER_ENV_VARS) docker compose -f docker-compose.aws-load.yml up -d +.PHONY: docker-aws-load-start + +docker-aws-load-stop: + @cd docker && $(CLUSTER_ENV_VARS) docker compose -f docker-compose.aws-load.yml down +.PHONY: docker-aws-load-stop + # Run GIGA EVM integration tests with a GIGA-enabled cluster # This starts a fresh cluster with GIGA_EXECUTOR and GIGA_OCC enabled, # runs the EVM GIGA tests, then stops the cluster. diff --git a/cmd/autobahn-e2e/aws.go b/cmd/autobahn-e2e/aws.go index fc88503dfb..6a951d5fa9 100644 --- a/cmd/autobahn-e2e/aws.go +++ b/cmd/autobahn-e2e/aws.go @@ -11,10 +11,13 @@ import ( "regexp" "strconv" "strings" + "sync" "time" + + "golang.org/x/sync/errgroup" ) -const ubuntuARM64AMIParameter = "/aws/service/canonical/ubuntu/server/24.04/stable/current/arm64/hvm/ebs-gp3/ami-id" +const ubuntuAMD64AMIParameter = "/aws/service/canonical/ubuntu/server/24.04/stable/current/amd64/hvm/ebs-gp3/ami-id" var sshUserPattern = regexp.MustCompile(`^[a-z_][a-z0-9_-]*$`) @@ -44,13 +47,19 @@ func (a *application) deployAWS(ctx context.Context, options deployOptions) erro if options.volumeSize < 20 { return fmt.Errorf("--volume-size must be at least 20 GiB") } + if options.volumeIOPS <= 0 { + return fmt.Errorf("--volume-iops must be positive") + } + if options.volumeThroughput <= 0 { + return fmt.Errorf("--volume-throughput must be positive") + } if options.keyName != "" && options.sshKeyPath == "" { return fmt.Errorf("--ssh-key is required with --key-name") } if !sshUserPattern.MatchString(options.sshUser) { return fmt.Errorf("invalid --ssh-user %q", options.sshUser) } - for _, name := range []string{"aws", "git", "ssh"} { + for _, name := range []string{"aws", "git", "ssh", "scp"} { if err := a.runner.lookPath(name); err != nil { return err } @@ -67,7 +76,7 @@ func (a *application) deployAWS(ctx context.Context, options deployOptions) erro if amiID == "" { amiID, err = client.output(ctx, "ssm", "get-parameter", - "--name", ubuntuARM64AMIParameter, + "--name", ubuntuAMD64AMIParameter, "--query", "Parameter.Value", "--output", "text", ) @@ -84,6 +93,10 @@ func (a *application) deployAWS(ctx context.Context, options deployOptions) erro if err != nil { return err } + grafanaCIDR, err := resolveGrafanaCIDR(options.grafanaCIDR, sshCIDR) + if err != nil { + return err + } state := clusterState{ Version: stateVersion, @@ -95,6 +108,7 @@ func (a *application) deployAWS(ctx context.Context, options deployOptions) erro AWS: &awsState{ Region: options.region, Profile: options.profile, + Topology: options.topology, SSHUser: options.sshUser, RemoteDir: filepath.Join("/home", options.sshUser, "sei-chain-"+options.name), RepoURL: repoURL, @@ -113,7 +127,7 @@ func (a *application) deployAWS(ctx context.Context, options deployOptions) erro securityGroupID, err := client.output(ctx, "ec2", "create-security-group", "--group-name", securityGroupName, - "--description", "SSH access for Sei Autobahn EVM-only E2E", + "--description", "SSH, Grafana, and intra-cluster traffic for Sei Autobahn E2E", "--vpc-id", vpcID, "--query", "GroupId", "--output", "text", @@ -132,15 +146,17 @@ func (a *application) deployAWS(ctx context.Context, options deployOptions) erro ); err != nil { return fail(err) } - if _, err := client.output(ctx, - "ec2", "authorize-security-group-ingress", - "--group-id", state.AWS.SecurityGroupID, - "--protocol", "tcp", - "--port", "22", - cidrFlag(sshCIDR), sshCIDR, - ); err != nil { + if err := client.authorizeTCP(ctx, state.AWS.SecurityGroupID, "22", sshCIDR); err != nil { + return fail(err) + } + if err := client.authorizeTCP(ctx, state.AWS.SecurityGroupID, strconv.Itoa(grafanaPublicPort), grafanaCIDR); err != nil { return fail(err) } + if !state.AWS.colocated() { + if err := client.authorizeTCPFromGroup(ctx, state.AWS.SecurityGroupID, "1", "65535"); err != nil { + return fail(err) + } + } if options.keyName == "" { if err := os.MkdirAll(a.stateDir, 0o700); err != nil { @@ -178,71 +194,146 @@ func (a *application) deployAWS(ctx context.Context, options deployOptions) erro return fail(err) } defer func() { _ = os.Remove(userDataPath) }() - runArgs := []string{ - "ec2", "run-instances", - "--image-id", amiID, - "--instance-type", options.instanceType, - "--key-name", state.AWS.KeyName, - "--security-group-ids", state.AWS.SecurityGroupID, - "--associate-public-ip-address", - "--metadata-options", "HttpTokens=required,HttpEndpoint=enabled", - "--block-device-mappings", fmt.Sprintf("DeviceName=/dev/sda1,Ebs={VolumeSize=%d,VolumeType=gp3,DeleteOnTermination=true}", options.volumeSize), - "--user-data", "file://" + userDataPath, - "--tag-specifications", fmt.Sprintf("ResourceType=instance,Tags=[{Key=Name,Value=sei-autobahn-e2e-%s},{Key=sei-autobahn-e2e-cluster,Value=%s}]", options.name, options.name), - "--query", "Instances[0].InstanceId", - "--output", "text", + + if state.AWS.colocated() { + if err := a.launchColocatedInstance(ctx, client, options, &state, amiID, userDataPath); err != nil { + return fail(err) + } + } else if err := a.launchDistributedInstances(ctx, client, options, &state, amiID, userDataPath); err != nil { + return fail(err) } - if options.subnetID != "" { - runArgs = append(runArgs, "--subnet-id", options.subnetID) + + readyCtx, cancel := context.WithTimeout(ctx, options.timeout) + defer cancel() + _, _ = fmt.Fprintln(a.stdout, "Waiting for cloud-init on every instance.") + if err := a.waitForAllBootstraps(readyCtx, state); err != nil { + return fail(err) } - instanceID, err := client.output(ctx, runArgs...) - if err != nil { + if err := a.startRemoteCluster(readyCtx, state); err != nil { return fail(err) } - state.AWS.InstanceID = strings.TrimSpace(instanceID) + if err := a.waitForRemoteCluster(readyCtx, state); err != nil { + return fail(err) + } + if err := a.waitForRemoteGrafana(readyCtx, state); err != nil { + return fail(err) + } + state.Status = "ready" if err := a.store().save(state); err != nil { return err } - if err := client.stream(ctx, "ec2", "wait", "instance-running", "--instance-ids", state.AWS.InstanceID); err != nil { - return fail(err) + if state.AWS.colocated() { + _, _ = fmt.Fprintf(a.stdout, "Cluster %s is ready on EC2 instance %s (%s) with four Docker validators.\n", state.Name, state.AWS.InstanceID, state.AWS.PublicIP) + } else { + _, _ = fmt.Fprintf(a.stdout, "Cluster %s is ready with %d validator instances and 1 load instance.\n", state.Name, len(state.AWS.validators())) } - if err := client.stream(ctx, "ec2", "wait", "instance-status-ok", "--instance-ids", state.AWS.InstanceID); err != nil { - return fail(err) + _, _ = fmt.Fprintf(a.stdout, "Grafana: %s (admin / admin)\n", grafanaPublicURL(state.AWS.PublicIP)) + if load, ok := state.AWS.loadHost(); ok && !state.AWS.colocated() { + _, _ = fmt.Fprintf(a.stdout, "sei-load is not running. Start it on the load instance when you want traffic:\n") + _, _ = fmt.Fprintf(a.stdout, " ssh -i %s %s@%s\n", expandHome(state.AWS.SSHKeyPath), state.AWS.SSHUser, load.PublicIP) + _, _ = fmt.Fprintf(a.stdout, " cd %s && GOBIN=\"$PWD/build/tools\" go install github.com/sei-protocol/sei-load@%s\n", state.AWS.RemoteDir, seiLoadVersion) + _, _ = fmt.Fprintf(a.stdout, " ./build/tools/sei-load --config integration_test/autobahn/sei-load.aws.json --metricsListenAddr 0.0.0.0:19698\n") } - publicIP, err := client.output(ctx, - "ec2", "describe-instances", - "--instance-ids", state.AWS.InstanceID, - "--query", "Reservations[0].Instances[0].PublicIpAddress", - "--output", "text", + return nil +} + +func (a *application) launchDistributedInstances(ctx context.Context, client awsClient, options deployOptions, state *clusterState, amiID, userDataPath string) error { + _, _ = fmt.Fprintf(a.stdout, "Launching %d validator instances and 1 load instance in %s.\n", awsValidatorCount, options.region) + var ( + launchMu sync.Mutex + validatorIDs []string + loadIDs []string ) + // One launch failure must not cancel a sibling run-instances. + var launch errgroup.Group + launch.Go(func() error { + ids, err := client.runInstances(ctx, options, *state, amiID, userDataPath, awsRoleValidator, awsValidatorCount, options.volumeSize, options.volumeIOPS, options.volumeThroughput) + if err != nil { + return err + } + launchMu.Lock() + defer launchMu.Unlock() + validatorIDs = ids + state.AWS.Hosts = append(state.AWS.Hosts, hostsFromIDs(awsRoleValidator, ids)...) + return a.store().save(*state) + }) + launch.Go(func() error { + ids, err := client.runInstances(ctx, options, *state, amiID, userDataPath, awsRoleLoad, 1, defaultLoadVolumeSizeGiB, defaultLoadVolumeIOPS, defaultLoadVolumeThroughputMB) + if err != nil { + return err + } + launchMu.Lock() + defer launchMu.Unlock() + loadIDs = ids + state.AWS.Hosts = append(state.AWS.Hosts, hostsFromIDs(awsRoleLoad, ids)...) + return a.store().save(*state) + }) + if err := launch.Wait(); err != nil { + return err + } + allIDs := append(append([]string{}, validatorIDs...), loadIDs...) + _, _ = fmt.Fprintf(a.stdout, "Launched %d instances; waiting for them to pass status checks.\n", len(allIDs)) + if err := client.waitInstances(ctx, allIDs); err != nil { + return err + } + _, _ = fmt.Fprintln(a.stdout, "All instances passed status checks.") + infos, err := client.describeInstanceIPs(ctx, allIDs) if err != nil { - return fail(err) + return err } - state.AWS.PublicIP = strings.TrimSpace(publicIP) - if state.AWS.PublicIP == "" || state.AWS.PublicIP == "None" { - return fail(fmt.Errorf("ec2 instance has no public IP; choose a subnet that assigns public addresses")) + hosts, err := assignAWSHosts(validatorIDs, loadIDs, infos) + if err != nil { + return err } - if err := a.store().save(state); err != nil { + state.AWS.Hosts = hosts + if load, ok := state.AWS.loadHost(); ok { + state.AWS.InstanceID = load.InstanceID + state.AWS.PublicIP = load.PublicIP + } + if err := a.store().save(*state); err != nil { return err } + for _, host := range state.AWS.validators() { + name := fmt.Sprintf("sei-autobahn-e2e-%s-validator-%d", options.name, host.Index) + if _, err := client.output(ctx, "ec2", "create-tags", "--resources", host.InstanceID, "--tags", "Key=Name,Value="+name); err != nil { + return err + } + } + return nil +} - readyCtx, cancel := context.WithTimeout(ctx, options.timeout) - defer cancel() - if err := a.waitForEC2Bootstrap(readyCtx, state); err != nil { - return fail(err) +func (a *application) launchColocatedInstance(ctx context.Context, client awsClient, options deployOptions, state *clusterState, amiID, userDataPath string) error { + _, _ = fmt.Fprintf(a.stdout, "Launching 1 colocated instance in %s.\n", options.region) + ids, err := client.runInstances(ctx, options, *state, amiID, userDataPath, awsRoleLoad, 1, options.volumeSize, options.volumeIOPS, options.volumeThroughput) + if err != nil { + return err } - if err := a.startRemoteCluster(readyCtx, state); err != nil { - return fail(err) + state.AWS.Hosts = hostsFromIDs(awsRoleLoad, ids) + if err := a.store().save(*state); err != nil { + return err } - if err := a.waitForRemoteCluster(readyCtx, state); err != nil { - return fail(err) + _, _ = fmt.Fprintf(a.stdout, "Launched %s; waiting for it to pass status checks.\n", ids[0]) + if err := client.waitInstances(ctx, ids); err != nil { + return err } - state.Status = "ready" - if err := a.store().save(state); err != nil { + _, _ = fmt.Fprintln(a.stdout, "Instance passed status checks.") + infos, err := client.describeInstanceIPs(ctx, ids) + if err != nil { return err } - _, _ = fmt.Fprintf(a.stdout, "Cluster %s is ready on EC2 instance %s (%s).\n", state.Name, state.AWS.InstanceID, state.AWS.PublicIP) - return nil + hosts, err := assignAWSHosts(nil, ids, infos) + if err != nil { + return err + } + state.AWS.Hosts = hosts + if load, ok := state.AWS.loadHost(); ok { + state.AWS.InstanceID = load.InstanceID + state.AWS.PublicIP = load.PublicIP + } + if _, err := client.output(ctx, "ec2", "create-tags", "--resources", state.AWS.InstanceID, "--tags", "Key=Name,Value=sei-autobahn-e2e-"+options.name); err != nil { + return err + } + return a.store().save(*state) } func (a *application) ensureAWSCredentials(ctx context.Context, client awsClient) error { @@ -324,6 +415,16 @@ func resolveVPC(ctx context.Context, client awsClient, subnetID string) (string, return vpcID, nil } +func resolveGrafanaCIDR(configured, sshCIDR string) (string, error) { + if configured == "" { + return sshCIDR, nil + } + if _, _, err := net.ParseCIDR(configured); err != nil { + return "", fmt.Errorf("invalid --grafana-cidr: %w", err) + } + return configured, nil +} + func resolveSSHCIDR(ctx context.Context, configured string) (string, error) { if configured != "" { if _, _, err := net.ParseCIDR(configured); err != nil { @@ -405,67 +506,257 @@ touch /var/lib/autobahn-e2e-ready return path, nil } -func (a *application) waitForEC2Bootstrap(ctx context.Context, state clusterState) error { +func (c awsClient) authorizeTCP(ctx context.Context, groupID, port, cidr string) error { + _, err := c.output(ctx, + "ec2", "authorize-security-group-ingress", + "--group-id", groupID, + "--protocol", "tcp", + "--port", port, + cidrFlag(cidr), cidr, + ) + return err +} + +func (c awsClient) authorizeTCPFromGroup(ctx context.Context, groupID, fromPort, toPort string) error { + permission := fmt.Sprintf("IpProtocol=tcp,FromPort=%s,ToPort=%s,UserIdGroupPairs=[{GroupId=%s}]", fromPort, toPort, groupID) + _, err := c.output(ctx, + "ec2", "authorize-security-group-ingress", + "--group-id", groupID, + "--ip-permissions", permission, + ) + return err +} + +func (c awsClient) runInstances(ctx context.Context, options deployOptions, state clusterState, amiID, userDataPath, role string, count, volumeSize, volumeIOPS, volumeThroughput int) ([]string, error) { + name := fmt.Sprintf("sei-autobahn-e2e-%s-%s", options.name, role) + runArgs := []string{ + "ec2", "run-instances", + "--image-id", amiID, + "--count", strconv.Itoa(count), + "--instance-type", options.instanceType, + "--key-name", state.AWS.KeyName, + "--security-group-ids", state.AWS.SecurityGroupID, + "--associate-public-ip-address", + "--metadata-options", "HttpTokens=required,HttpEndpoint=enabled", + "--block-device-mappings", ebsRootMapping(volumeSize, volumeIOPS, volumeThroughput), + "--user-data", "file://" + userDataPath, + "--tag-specifications", fmt.Sprintf("ResourceType=instance,Tags=[{Key=Name,Value=%s},{Key=sei-autobahn-e2e-cluster,Value=%s},{Key=sei-autobahn-e2e-role,Value=%s}]", name, options.name, role), + "--query", "Instances[*].InstanceId", + "--output", "text", + } + if options.subnetID != "" { + runArgs = append(runArgs, "--subnet-id", options.subnetID) + } + value, err := c.output(ctx, runArgs...) + if err != nil { + return nil, err + } + ids := strings.Fields(strings.TrimSpace(value)) + if len(ids) != count { + return nil, fmt.Errorf("run-instances for %s: expected %d instance IDs, got %d", role, count, len(ids)) + } + return ids, nil +} + +func (c awsClient) waitInstances(ctx context.Context, ids []string) error { + if len(ids) == 0 { + return fmt.Errorf("wait for instances: no instance IDs") + } + args := append([]string{"ec2", "wait", "instance-running", "--instance-ids"}, ids...) + if err := c.stream(ctx, args...); err != nil { + return err + } + args = append([]string{"ec2", "wait", "instance-status-ok", "--instance-ids"}, ids...) + return c.stream(ctx, args...) +} + +type instanceAddrs struct { + publicIP string + privateIP string +} + +func (c awsClient) describeInstanceIPs(ctx context.Context, ids []string) (map[string]instanceAddrs, error) { + args := append([]string{"ec2", "describe-instances", "--instance-ids"}, ids...) + args = append(args, "--query", "Reservations[].Instances[].[InstanceId,PublicIpAddress,PrivateIpAddress]", "--output", "text") + value, err := c.output(ctx, args...) + if err != nil { + return nil, err + } + infos := map[string]instanceAddrs{} + for _, line := range strings.Split(strings.TrimSpace(value), "\n") { + fields := strings.Fields(line) + if len(fields) < 3 { + continue + } + infos[fields[0]] = instanceAddrs{publicIP: fields[1], privateIP: fields[2]} + } + return infos, nil +} + +func hostsFromIDs(role string, ids []string) []awsHost { + hosts := make([]awsHost, len(ids)) + for i, id := range ids { + hosts[i] = awsHost{Role: role, Index: i, InstanceID: id} + } + if role == awsRoleLoad { + for i := range hosts { + hosts[i].Index = 0 + } + } + return hosts +} + +func assignAWSHosts(validatorIDs, loadIDs []string, infos map[string]instanceAddrs) ([]awsHost, error) { + hosts := make([]awsHost, 0, len(validatorIDs)+len(loadIDs)) + for i, id := range validatorIDs { + info, ok := infos[id] + if !ok { + return nil, fmt.Errorf("missing addresses for validator instance %s", id) + } + if info.publicIP == "" || info.publicIP == "None" || info.privateIP == "" || info.privateIP == "None" { + return nil, fmt.Errorf("ec2 instance %s has no public or private IP; choose a subnet that assigns public addresses", id) + } + hosts = append(hosts, awsHost{ + Role: awsRoleValidator, + Index: i, + InstanceID: id, + PublicIP: info.publicIP, + PrivateIP: info.privateIP, + }) + } + for _, id := range loadIDs { + info, ok := infos[id] + if !ok { + return nil, fmt.Errorf("missing addresses for load instance %s", id) + } + if info.publicIP == "" || info.publicIP == "None" || info.privateIP == "" || info.privateIP == "None" { + return nil, fmt.Errorf("ec2 instance %s has no public or private IP; choose a subnet that assigns public addresses", id) + } + hosts = append(hosts, awsHost{ + Role: awsRoleLoad, + InstanceID: id, + PublicIP: info.publicIP, + PrivateIP: info.privateIP, + }) + } + return hosts, nil +} + +func (a *application) waitForAllBootstraps(ctx context.Context, state clusterState) error { + return a.forEachHost(ctx, state.AWS.Hosts, func(ctx context.Context, host awsHost) error { + return a.waitForEC2Bootstrap(ctx, state, host) + }) +} + +func (a *application) waitForEC2Bootstrap(ctx context.Context, state clusterState, host awsHost) error { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() for { - _, err := a.runner.output(ctx, sshCommand(state, "test -f /var/lib/autobahn-e2e-ready")) + _, err := a.runner.output(ctx, sshCommandTo(state, host, "test -f /var/lib/autobahn-e2e-ready")) if err == nil { return nil } select { case <-ctx.Done(): - return fmt.Errorf("wait for EC2 bootstrap: %w", ctx.Err()) + return fmt.Errorf("wait for EC2 bootstrap on %s: %w", host.PublicIP, ctx.Err()) case <-ticker.C: } } } -func (a *application) startRemoteCluster(ctx context.Context, state clusterState) error { - aws := state.AWS - command := strings.Join([]string{ - "git clone --filter=blob:none " + shellQuote(aws.RepoURL) + " " + shellQuote(aws.RemoteDir), - "cd " + shellQuote(aws.RemoteDir), - "git checkout --detach " + shellQuote(aws.Ref), - "AUTOBAHN=true AUTOBAHN_EVMONLY=true DOCKER_DETACH=true make docker-cluster-start", - }, " && ") - if err := a.runner.stream(ctx, sshCommand(state, command)); err != nil { - return fmt.Errorf("start remote cluster: %w", err) +func (a *application) waitForRemoteGrafana(ctx context.Context, state clusterState) error { + load, ok := state.AWS.loadHost() + if !ok { + return fmt.Errorf("wait for Grafana: load instance is missing") } - return nil -} - -func (a *application) waitForRemoteCluster(ctx context.Context, state clusterState) error { - command := "test \"$(wc -l < " + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated/launch.complete")) + ")\" -ge " + strconv.Itoa(dockerClusterSize) + command := "curl -fsS -o /dev/null http://127.0.0.1:" + strconv.Itoa(grafanaPublicPort) + "/api/health" ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() for { - if _, err := a.runner.output(ctx, sshCommand(state, command)); err == nil { + if _, err := a.runner.output(ctx, sshCommandTo(state, load, command)); err == nil { return nil } select { case <-ctx.Done(): - return fmt.Errorf("wait for remote cluster: %w", ctx.Err()) + return fmt.Errorf("wait for Grafana: %w", ctx.Err()) case <-ticker.C: } } } +func (a *application) waitForRemoteCluster(ctx context.Context, state clusterState) error { + minLines := 1 + hosts := state.AWS.validators() + if state.AWS.colocated() { + minLines = dockerClusterSize + host, ok := state.AWS.loadHost() + if !ok { + return fmt.Errorf("wait for remote cluster: colocated instance is missing") + } + hosts = []awsHost{host} + } + command := "test \"$(wc -l < " + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated/launch.complete")) + ")\" -ge " + strconv.Itoa(minLines) + return a.forEachHost(ctx, hosts, func(ctx context.Context, host awsHost) error { + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + for { + if _, err := a.runner.output(ctx, sshCommandTo(state, host, command)); err == nil { + return nil + } + select { + case <-ctx.Done(): + return fmt.Errorf("wait for remote validator %d: %w", host.Index, ctx.Err()) + case <-ticker.C: + } + } + }) +} + func sshCommand(state clusterState, remoteCommand string) commandSpec { - return commandSpec{name: "ssh", args: append(sshBaseArgs(state), remoteCommand)} + return sshCommandTo(state, awsHost{PublicIP: state.AWS.PublicIP}, remoteCommand) +} + +func sshCommandTo(state clusterState, host awsHost, remoteCommand string) commandSpec { + return commandSpec{name: "ssh", args: append(sshBaseArgsTo(state, host), remoteCommand)} } func sshBaseArgs(state clusterState) []string { + return sshBaseArgsTo(state, awsHost{PublicIP: state.AWS.PublicIP}) +} + +func sshBaseArgsTo(state clusterState, host awsHost) []string { aws := state.AWS return []string{ "-i", expandHome(aws.SSHKeyPath), "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", "-o", "StrictHostKeyChecking=accept-new", - aws.SSHUser + "@" + aws.PublicIP, + aws.SSHUser + "@" + host.PublicIP, } } +func scpBaseArgs(state clusterState) []string { + return []string{ + "-3", + "-r", + "-i", expandHome(state.AWS.SSHKeyPath), + "-o", "BatchMode=yes", + "-o", "ConnectTimeout=10", + "-o", "StrictHostKeyChecking=accept-new", + } +} + +func remoteSSHPath(state clusterState, host awsHost, path string) string { + return state.AWS.SSHUser + "@" + host.PublicIP + ":" + path +} + +func ebsRootMapping(sizeGiB, iops, throughputMB int) string { + return fmt.Sprintf( + "DeviceName=/dev/sda1,Ebs={VolumeSize=%d,VolumeType=gp3,Iops=%d,Throughput=%d,DeleteOnTermination=true}", + sizeGiB, iops, throughputMB, + ) +} + func shellQuote(value string) string { return "'" + strings.ReplaceAll(value, "'", `'\''`) + "'" } diff --git a/cmd/autobahn-e2e/aws_remote.go b/cmd/autobahn-e2e/aws_remote.go new file mode 100644 index 0000000000..abbb53f61a --- /dev/null +++ b/cmd/autobahn-e2e/aws_remote.go @@ -0,0 +1,287 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "path/filepath" + "sort" + "strconv" + "strings" + + "golang.org/x/sync/errgroup" +) + +func (a *application) startRemoteCluster(ctx context.Context, state clusterState) error { + if state.AWS.colocated() { + return a.startRemoteColocatedCluster(ctx, state) + } + validators := state.AWS.validators() + if len(validators) != awsValidatorCount { + return fmt.Errorf("start remote cluster: expected %d validators, got %d", awsValidatorCount, len(validators)) + } + load, ok := state.AWS.loadHost() + if !ok { + return fmt.Errorf("start remote cluster: load instance is missing") + } + + _, _ = fmt.Fprintln(a.stdout, "Cloning, building, and initializing all validators in parallel.") + if err := a.forEachHost(ctx, validators, func(ctx context.Context, host awsHost) error { + if err := a.cloneRemoteRepo(ctx, state, host); err != nil { + return err + } + return a.buildAndInitValidator(ctx, state, host) + }); err != nil { + return err + } + if err := a.collectValidatorArtifacts(ctx, state, validators); err != nil { + return err + } + + _, _ = fmt.Fprintln(a.stdout, "Generating genesis on validator 0.") + if err := a.remoteStream(ctx, state, validators[0], remoteMake(state, "docker-aws-validator-genesis", nil)); err != nil { + return fmt.Errorf("generate genesis: %w", err) + } + if err := a.distributeGenesis(ctx, state, validators); err != nil { + return err + } + + _, _ = fmt.Fprintln(a.stdout, "Starting one validator process on each instance.") + if err := a.forEachHost(ctx, validators, func(ctx context.Context, host awsHost) error { + return a.remoteStream(ctx, state, host, remoteMake(state, "docker-aws-validator-start", map[string]string{ + "ID": strconv.Itoa(host.Index), + "ADVERTISE_IP": host.PrivateIP, + })) + }); err != nil { + return fmt.Errorf("start validators: %w", err) + } + + _, _ = fmt.Fprintln(a.stdout, "Setting up Grafana and Prometheus on the load instance.") + if err := a.cloneRemoteRepo(ctx, state, load); err != nil { + return err + } + if err := a.startLoadHost(ctx, state, load, validators); err != nil { + return err + } + return nil +} + +func (a *application) startRemoteColocatedCluster(ctx context.Context, state clusterState) error { + host, ok := state.AWS.loadHost() + if !ok { + return fmt.Errorf("start remote cluster: colocated instance is missing") + } + _, _ = fmt.Fprintln(a.stdout, "Cloning the repository and starting four Docker validators plus monitoring.") + if err := a.cloneRemoteRepo(ctx, state, host); err != nil { + return err + } + command := strings.Join([]string{ + "cd " + shellQuote(state.AWS.RemoteDir), + "AUTOBAHN=true AUTOBAHN_EVMONLY=true DOCKER_DETACH=true make docker-cluster-start-monitoring", + }, " && ") + if err := a.remoteStream(ctx, state, host, command); err != nil { + return fmt.Errorf("start colocated cluster: %w", err) + } + return nil +} + +func (a *application) cloneRemoteRepo(ctx context.Context, state clusterState, host awsHost) error { + command := strings.Join([]string{ + "git clone --filter=blob:none " + shellQuote(state.AWS.RepoURL) + " " + shellQuote(state.AWS.RemoteDir), + "cd " + shellQuote(state.AWS.RemoteDir), + "git checkout --detach " + shellQuote(state.AWS.Ref), + }, " && ") + if err := a.remoteStream(ctx, state, host, command); err != nil { + return fmt.Errorf("clone repository on %s: %w", host.PublicIP, err) + } + return nil +} + +func (a *application) buildAndInitValidator(ctx context.Context, state clusterState, host awsHost) error { + command := strings.Join([]string{ + "cd " + shellQuote(state.AWS.RemoteDir), + "make build-docker-node", + remoteMake(state, "docker-aws-validator-init", map[string]string{ + "ID": strconv.Itoa(host.Index), + "ADVERTISE_IP": host.PrivateIP, + }), + }, " && ") + if err := a.remoteStream(ctx, state, host, command); err != nil { + return fmt.Errorf("build and init validator %d: %w", host.Index, err) + } + return nil +} + +func (a *application) collectValidatorArtifacts(ctx context.Context, state clusterState, validators []awsHost) error { + primary := validators[0] + for _, host := range validators[1:] { + bundle := fmt.Sprintf("/tmp/autobahn-e2e-node-%d.tgz", host.Index) + create := strings.Join([]string{ + "tar -C " + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated")) + " -czf " + shellQuote(bundle), + fmt.Sprintf("node_%d gentx exported_keys genesis_accounts.txt persistent_peers.txt init.complete", host.Index), + }, " ") + if err := a.remoteStream(ctx, state, host, create); err != nil { + return fmt.Errorf("bundle validator %d artifacts: %w", host.Index, err) + } + dest := fmt.Sprintf("/tmp/autobahn-e2e-node-%d.tgz", host.Index) + if err := a.copyBetweenHosts(ctx, state, host, bundle, primary, dest); err != nil { + return fmt.Errorf("copy validator %d artifacts: %w", host.Index, err) + } + merge := strings.Join([]string{ + "set -euo pipefail", + fmt.Sprintf("mkdir -p /tmp/autobahn-e2e-merge-%d", host.Index), + fmt.Sprintf("tar -C /tmp/autobahn-e2e-merge-%d -xzf %s", host.Index, dest), + "GEN=" + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated")), + fmt.Sprintf("cp -a /tmp/autobahn-e2e-merge-%d/node_%d \"$GEN/\"", host.Index, host.Index), + "cp /tmp/autobahn-e2e-merge-" + strconv.Itoa(host.Index) + "/gentx/* \"$GEN/gentx/\"", + "cp /tmp/autobahn-e2e-merge-" + strconv.Itoa(host.Index) + "/exported_keys/* \"$GEN/exported_keys/\"", + fmt.Sprintf("cat /tmp/autobahn-e2e-merge-%d/genesis_accounts.txt >> \"$GEN/genesis_accounts.txt\"", host.Index), + fmt.Sprintf("cat /tmp/autobahn-e2e-merge-%d/persistent_peers.txt >> \"$GEN/persistent_peers.txt\"", host.Index), + fmt.Sprintf("cat /tmp/autobahn-e2e-merge-%d/init.complete >> \"$GEN/init.complete\"", host.Index), + }, "\n") + if err := a.remoteStream(ctx, state, primary, merge); err != nil { + return fmt.Errorf("merge validator %d artifacts: %w", host.Index, err) + } + } + return nil +} + +func (a *application) distributeGenesis(ctx context.Context, state clusterState, validators []awsHost) error { + primary := validators[0] + bundle := "/tmp/autobahn-e2e-genesis.tgz" + nodeDirs := make([]string, 0, len(validators)+2) + nodeDirs = append(nodeDirs, "genesis.json", "persistent_peers.txt") + for _, host := range validators { + nodeDirs = append(nodeDirs, fmt.Sprintf("node_%d", host.Index)) + } + create := "tar -C " + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated")) + " -czf " + shellQuote(bundle) + " " + strings.Join(nodeDirs, " ") + if err := a.remoteStream(ctx, state, primary, create); err != nil { + return fmt.Errorf("bundle genesis: %w", err) + } + for _, host := range validators[1:] { + if err := a.copyBetweenHosts(ctx, state, primary, bundle, host, bundle); err != nil { + return fmt.Errorf("copy genesis to validator %d: %w", host.Index, err) + } + extract := "mkdir -p " + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated")) + + " && tar -C " + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated")) + " -xzf " + shellQuote(bundle) + if err := a.remoteStream(ctx, state, host, extract); err != nil { + return fmt.Errorf("extract genesis on validator %d: %w", host.Index, err) + } + } + return nil +} + +func (a *application) startLoadHost(ctx context.Context, state clusterState, load awsHost, validators []awsHost) error { + privateIPs := make([]string, len(validators)) + for i, host := range validators { + privateIPs[i] = host.PrivateIP + } + prom := prometheusScrapeConfig(privateIPs) + loadCfg, err := seiLoadAWSConfig(privateIPs) + if err != nil { + return err + } + write := strings.Join([]string{ + "set -euo pipefail", + "mkdir -p " + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated")), + "cat > " + shellQuote(filepath.Join(state.AWS.RemoteDir, "build/generated/prometheus.yml")) + " <<'EOF'\n" + prom + "EOF", + "cat > " + shellQuote(filepath.Join(state.AWS.RemoteDir, "integration_test/autobahn/sei-load.aws.json")) + " <<'EOF'\n" + loadCfg + "EOF", + }, "\n") + if err := a.remoteStream(ctx, state, load, write); err != nil { + return fmt.Errorf("write load-host config: %w", err) + } + if err := a.remoteStream(ctx, state, load, remoteMake(state, "docker-aws-load-start", nil)); err != nil { + return fmt.Errorf("start load-host monitoring: %w", err) + } + return nil +} + +func remoteMake(state clusterState, target string, extra map[string]string) string { + parts := []string{ + "cd " + shellQuote(state.AWS.RemoteDir), + "AUTOBAHN=true AUTOBAHN_EVMONLY=true", + } + keys := make([]string, 0, len(extra)) + for key := range extra { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + parts[1] += " " + key + "=" + shellQuote(extra[key]) + } + parts[1] += " make " + target + return strings.Join(parts, " && ") +} + +func prometheusScrapeConfig(privateIPs []string) string { + var b strings.Builder + b.WriteString("global:\n") + b.WriteString(" scrape_interval: 15s\n") + b.WriteString(" evaluation_interval: 15s\n\n") + b.WriteString("scrape_configs:\n") + b.WriteString(" - job_name: autobahn-e2e\n") + b.WriteString(" metrics_path: /metrics\n") + b.WriteString(" static_configs:\n") + b.WriteString(" - targets:\n") + for _, ip := range privateIPs { + fmt.Fprintf(&b, " - '%s:%d'\n", ip, awsMetricsPort) + } + b.WriteString(" scrape_interval: 5s\n") + return b.String() +} + +func seiLoadAWSConfig(privateIPs []string) (string, error) { + endpoints := make([]string, len(privateIPs)) + for i, ip := range privateIPs { + endpoints[i] = fmt.Sprintf("http://%s:%d", ip, awsEVMPort) + } + cfg := map[string]any{ + "chainId": 713715, + "seiChainID": "autobahn-evmonly", + "endpoints": endpoints, + "accounts": map[string]any{ + "count": 5000, + "newAccountRate": 0, + }, + "scenarios": []map[string]any{ + {"name": "EVMTransfer", "weight": 1}, + }, + "settings": map[string]any{ + "workers": 1, + "tps": 250, + "statsInterval": "5s", + "bufferSize": 1000, + "dryRun": false, + "debug": false, + "trackReceipts": false, + "trackBlocks": false, + "trackUserLatency": false, + "prewarm": false, + "rampUp": false, + "postSummaryFlushDelay": "1s", + }, + } + data, err := json.MarshalIndent(cfg, "", " ") + if err != nil { + return "", fmt.Errorf("encode sei-load config: %w", err) + } + return string(data) + "\n", nil +} + +func (a *application) forEachHost(ctx context.Context, hosts []awsHost, fn func(context.Context, awsHost) error) error { + g, ctx := errgroup.WithContext(ctx) + for _, host := range hosts { + host := host + g.Go(func() error { return fn(ctx, host) }) + } + return g.Wait() +} + +func (a *application) remoteStream(ctx context.Context, state clusterState, host awsHost, command string) error { + return a.runner.stream(ctx, sshCommandTo(state, host, command)) +} + +func (a *application) copyBetweenHosts(ctx context.Context, state clusterState, src awsHost, srcPath string, dst awsHost, dstPath string) error { + args := append(scpBaseArgs(state), remoteSSHPath(state, src, srcPath), remoteSSHPath(state, dst, dstPath)) + return a.runner.stream(ctx, commandSpec{name: "scp", args: args}) +} diff --git a/cmd/autobahn-e2e/command.go b/cmd/autobahn-e2e/command.go index 13ceeca574..7fc6e0dbf2 100644 --- a/cmd/autobahn-e2e/command.go +++ b/cmd/autobahn-e2e/command.go @@ -14,8 +14,28 @@ const ( defaultClusterName = "autobahn-evmonly" targetLocal = "local" targetAWS = "aws" + + grafanaPublicPort = 3000 + + awsTopologyDistributed = "distributed" + awsTopologyColocated = "colocated" + + awsValidatorCount = 4 + awsEVMPort = 8545 + awsMetricsPort = 26660 + defaultLoadVolumeSizeGiB = 100 + defaultLoadVolumeIOPS = 3000 + defaultLoadVolumeThroughputMB = 125 + seiLoadVersion = "v0.0.1" ) +func grafanaPublicURL(host string) string { + if host == "" { + return "" + } + return fmt.Sprintf("http://%s:%d", host, grafanaPublicPort) +} + type application struct { runner commandRunner stdout io.Writer diff --git a/cmd/autobahn-e2e/command_test.go b/cmd/autobahn-e2e/command_test.go index 252761995b..8b17937220 100644 --- a/cmd/autobahn-e2e/command_test.go +++ b/cmd/autobahn-e2e/command_test.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -14,6 +15,7 @@ import ( ) type fakeRunner struct { + mu sync.Mutex commands []commandSpec outputFn func(commandSpec) (string, error) streamFn func(commandSpec) error @@ -21,6 +23,8 @@ type fakeRunner struct { } func (r *fakeRunner) output(_ context.Context, spec commandSpec) (string, error) { + r.mu.Lock() + defer r.mu.Unlock() r.commands = append(r.commands, spec) if r.outputFn == nil { return "", nil @@ -29,6 +33,8 @@ func (r *fakeRunner) output(_ context.Context, spec commandSpec) (string, error) } func (r *fakeRunner) stream(_ context.Context, spec commandSpec) error { + r.mu.Lock() + defer r.mu.Unlock() r.commands = append(r.commands, spec) if r.streamFn == nil { return nil @@ -104,9 +110,12 @@ func TestAWSDeployCreatesManagedResourcesAndReadyState(t *testing.T) { case strings.Contains(joined, "create-key-pair"): return "-----BEGIN OPENSSH PRIVATE KEY-----\ntest\n-----END OPENSSH PRIVATE KEY-----\n", nil case strings.Contains(joined, "run-instances"): - return "i-123\n", nil + if strings.Contains(joined, "Value=load") { + return "i-load\n", nil + } + return "i-v0\ti-v1\ti-v2\ti-v3\n", nil case strings.Contains(joined, "describe-instances"): - return "203.0.113.10\n", nil + return "i-v0\t203.0.113.10\t10.0.0.10\ni-v1\t203.0.113.11\t10.0.0.11\ni-v2\t203.0.113.12\t10.0.0.12\ni-v3\t203.0.113.13\t10.0.0.13\ni-load\t203.0.113.20\t10.0.0.20\n", nil case spec.name == "ssh": return "", nil default: @@ -116,37 +125,122 @@ func TestAWSDeployCreatesManagedResourcesAndReadyState(t *testing.T) { var stdout, stderr bytes.Buffer app := &application{runner: runner, stdout: &stdout, stderr: &stderr, stateDir: stateDir} options := deployOptions{ - name: "aws-test", - target: "aws", - timeout: time.Minute, - region: "us-west-2", - instanceType: "c7g.2xlarge", - amiID: "ami-123", - sshCIDR: "198.51.100.4/32", - sshUser: "ubuntu", - volumeSize: 100, - repoURL: "https://github.com/sei-protocol/sei-chain.git", - ref: "deadbeef", + name: "aws-test", + target: "aws", + timeout: time.Minute, + region: "us-west-2", + instanceType: "r7i.12xlarge", + amiID: "ami-123", + sshCIDR: "198.51.100.4/32", + sshUser: "ubuntu", + volumeSize: defaultVolumeSizeGiB, + volumeIOPS: defaultVolumeIOPS, + volumeThroughput: defaultVolumeThroughputMB, + repoURL: "https://github.com/sei-protocol/sei-chain.git", + ref: "deadbeef", + topology: awsTopologyDistributed, } require.NoError(t, app.deploy(context.Background(), options)) state, err := app.store().load(options.name) require.NoError(t, err) require.Equal(t, "ready", state.Status) - require.Equal(t, "i-123", state.AWS.InstanceID) - require.Equal(t, "203.0.113.10", state.AWS.PublicIP) + require.Equal(t, "i-load", state.AWS.InstanceID) + require.Equal(t, "203.0.113.20", state.AWS.PublicIP) + require.Len(t, state.AWS.Hosts, 5) + require.Len(t, state.AWS.validators(), 4) require.True(t, state.AWS.ManagedKey) require.FileExists(t, state.AWS.SSHKeyPath) keyInfo, err := os.Stat(state.AWS.SSHKeyPath) require.NoError(t, err) require.Equal(t, os.FileMode(0o600), keyInfo.Mode().Perm()) require.Contains(t, stdout.String(), "Cluster aws-test is ready") + require.Contains(t, stdout.String(), "Grafana: http://203.0.113.20:3000") + require.Contains(t, stdout.String(), "passed status checks") commands := joinedCommands(runner.commands) require.Contains(t, commands, "authorize-security-group-ingress") require.Contains(t, commands, "--cidr 198.51.100.4/32") + require.Contains(t, commands, "--port 3000") + require.Contains(t, commands, "--port 22") + require.NotContains(t, commands, "--cidr 0.0.0.0/0") + require.Contains(t, commands, "UserIdGroupPairs") + require.Contains(t, commands, "autobahn-e2e-genesis.tgz' genesis.json persistent_peers.txt") + require.Contains(t, commands, "--count 4") + require.Contains(t, commands, "--count 1") + require.Contains(t, commands, "docker-aws-validator-init") + require.Contains(t, commands, "docker-aws-validator-genesis") + require.Contains(t, commands, "docker-aws-validator-start") + require.Contains(t, commands, "docker-aws-load-start") + require.Contains(t, commands, "sei-load.aws.json") + require.NotContains(t, commands, "metricsListenAddr") + require.Contains(t, stdout.String(), "sei-load is not running") + require.Contains(t, commands, ebsRootMapping(defaultVolumeSizeGiB, defaultVolumeIOPS, defaultVolumeThroughputMB)) + require.Contains(t, commands, ebsRootMapping(defaultLoadVolumeSizeGiB, defaultLoadVolumeIOPS, defaultLoadVolumeThroughputMB)) require.Contains(t, commands, "AUTOBAHN_EVMONLY=true") require.Contains(t, commands, "-o StrictHostKeyChecking=accept-new") + require.Contains(t, commands, "curl -fsS -o /dev/null http://127.0.0.1:3000/api/health") + require.Equal(t, awsTopologyDistributed, state.AWS.Topology) +} + +func TestAWSDeployColocatedUsesOneInstanceAndSharedCompose(t *testing.T) { + stateDir := t.TempDir() + runner := &fakeRunner{} + runner.outputFn = func(spec commandSpec) (string, error) { + joined := strings.Join(spec.args, " ") + switch { + case strings.Contains(joined, "sts get-caller-identity"): + return `{}`, nil + case strings.Contains(joined, "describe-vpcs"): + return "vpc-123\n", nil + case strings.Contains(joined, "create-security-group"): + return "sg-123\n", nil + case strings.Contains(joined, "create-key-pair"): + return "-----BEGIN OPENSSH PRIVATE KEY-----\ntest\n-----END OPENSSH PRIVATE KEY-----\n", nil + case strings.Contains(joined, "run-instances"): + return "i-colo\n", nil + case strings.Contains(joined, "describe-instances"): + return "i-colo\t203.0.113.10\t10.0.0.10\n", nil + case spec.name == "ssh": + return "", nil + default: + return "", nil + } + } + var stdout bytes.Buffer + app := &application{runner: runner, stdout: &stdout, stderr: &bytes.Buffer{}, stateDir: stateDir} + require.NoError(t, app.deploy(context.Background(), deployOptions{ + name: "colo-test", + target: "aws", + timeout: time.Minute, + region: "us-west-2", + instanceType: "r7i.12xlarge", + amiID: "ami-123", + sshCIDR: "198.51.100.4/32", + sshUser: "ubuntu", + volumeSize: defaultVolumeSizeGiB, + volumeIOPS: defaultVolumeIOPS, + volumeThroughput: defaultVolumeThroughputMB, + repoURL: "https://github.com/sei-protocol/sei-chain.git", + ref: "deadbeef", + topology: awsTopologyColocated, + })) + state, err := app.store().load("colo-test") + require.NoError(t, err) + require.Equal(t, "ready", state.Status) + require.Equal(t, awsTopologyColocated, state.AWS.Topology) + require.Equal(t, "i-colo", state.AWS.InstanceID) + require.Equal(t, "203.0.113.10", state.AWS.PublicIP) + require.Len(t, state.AWS.Hosts, 1) + require.Empty(t, state.AWS.validators()) + require.Contains(t, stdout.String(), "four Docker validators") + require.Contains(t, stdout.String(), "Grafana: http://203.0.113.10:3000") + + commands := joinedCommands(runner.commands) + require.Contains(t, commands, "docker-cluster-start-monitoring") + require.NotContains(t, commands, "docker-aws-validator-init") + require.NotContains(t, commands, "UserIdGroupPairs") + require.NotContains(t, commands, "--count 4") } func TestAWSDeployRetainsFailedState(t *testing.T) { @@ -171,17 +265,19 @@ func TestAWSDeployRetainsFailedState(t *testing.T) { } app := &application{runner: runner, stdout: &bytes.Buffer{}, stderr: &bytes.Buffer{}, stateDir: stateDir} err := app.deploy(context.Background(), deployOptions{ - name: "failed-aws", - target: "aws", - timeout: time.Minute, - region: "us-west-2", - instanceType: "c7g.2xlarge", - amiID: "ami-123", - sshCIDR: "198.51.100.4/32", - sshUser: "ubuntu", - volumeSize: 100, - repoURL: "https://example.com/repo.git", - ref: "deadbeef", + name: "failed-aws", + target: "aws", + timeout: time.Minute, + region: "us-west-2", + instanceType: "r7i.12xlarge", + amiID: "ami-123", + sshCIDR: "198.51.100.4/32", + sshUser: "ubuntu", + volumeSize: defaultVolumeSizeGiB, + volumeIOPS: defaultVolumeIOPS, + volumeThroughput: defaultVolumeThroughputMB, + repoURL: "https://example.com/repo.git", + ref: "deadbeef", }) require.Error(t, err) state, loadErr := app.store().load("failed-aws") @@ -202,6 +298,12 @@ func TestAWSForwardUsesChosenNodePort(t *testing.T) { PublicIP: "203.0.113.10", SSHUser: "ubuntu", SSHKeyPath: "/tmp/test.pem", + Hosts: []awsHost{ + {Role: awsRoleValidator, Index: 0, PublicIP: "203.0.113.10"}, + {Role: awsRoleValidator, Index: 1, PublicIP: "203.0.113.11"}, + {Role: awsRoleValidator, Index: 2, PublicIP: "203.0.113.12"}, + {Role: awsRoleValidator, Index: 3, PublicIP: "203.0.113.13"}, + }, }, } require.NoError(t, newStateStore(stateDir).save(state)) @@ -218,8 +320,8 @@ func TestAWSForwardUsesChosenNodePort(t *testing.T) { require.Len(t, runner.commands, 1) require.Equal(t, "ssh", runner.commands[0].name) joined := strings.Join(runner.commands[0].args, " ") - require.Contains(t, joined, "-L 127.0.0.1:18545:127.0.0.1:8551") - require.True(t, strings.HasSuffix(joined, "ubuntu@203.0.113.10")) + require.Contains(t, joined, "-L 127.0.0.1:18545:127.0.0.1:8545") + require.True(t, strings.HasSuffix(joined, "ubuntu@203.0.113.13")) } func TestListShowsPartialAWSDeploymentWithoutCredentials(t *testing.T) { @@ -247,6 +349,37 @@ func TestListShowsPartialAWSDeploymentWithoutCredentials(t *testing.T) { require.Empty(t, runner.commands) } +func TestAWSTeardownStopsMonitoringStack(t *testing.T) { + stateDir := t.TempDir() + state := clusterState{ + Version: stateVersion, + Name: "monitored-aws", + Target: targetAWS, + Status: "ready", + Nodes: clusterNodes(4), + AWS: &awsState{ + Region: "us-west-2", + PublicIP: "203.0.113.10", + SSHUser: "ubuntu", + SSHKeyPath: "/tmp/test.pem", + RemoteDir: "/home/ubuntu/sei-chain-monitored-aws", + }, + } + store := newStateStore(stateDir) + require.NoError(t, store.save(state)) + runner := &fakeRunner{outputFn: func(spec commandSpec) (string, error) { + if strings.Contains(strings.Join(spec.args, " "), "sts get-caller-identity") { + return `{}`, nil + } + return "", nil + }} + app := &application{runner: runner, stdout: &bytes.Buffer{}, stderr: &bytes.Buffer{}, stateDir: stateDir} + + require.NoError(t, app.teardown(context.Background(), teardownOptions{name: state.Name})) + require.Contains(t, joinedCommands(runner.commands), "docker-cluster-stop-monitoring") + require.Contains(t, joinedCommands(runner.commands), "if [ -d '/home/ubuntu/sei-chain-monitored-aws' ]") +} + func TestAWSTeardownToleratesAlreadyDeletedManagedResources(t *testing.T) { stateDir := t.TempDir() keyPath := filepath.Join(stateDir, "managed.pem") @@ -321,6 +454,109 @@ func TestWriteUserDataUsesSelectedSSHUser(t *testing.T) { require.Contains(t, string(data), "/var/lib/autobahn-e2e-ready") } +func TestEBSRootMapping(t *testing.T) { + require.Equal(t, + "DeviceName=/dev/sda1,Ebs={VolumeSize=1024,VolumeType=gp3,Iops=10000,Throughput=1000,DeleteOnTermination=true}", + ebsRootMapping(1024, 10000, 1000), + ) +} + +func TestAssignAWSHostsRequiresPublicAndPrivateIPs(t *testing.T) { + _, err := assignAWSHosts([]string{"i-v0"}, []string{"i-load"}, map[string]instanceAddrs{ + "i-v0": {publicIP: "203.0.113.10", privateIP: "10.0.0.10"}, + "i-load": {publicIP: "None", privateIP: "10.0.0.20"}, + }) + require.Error(t, err) + + hosts, err := assignAWSHosts([]string{"i-v0"}, []string{"i-load"}, map[string]instanceAddrs{ + "i-v0": {publicIP: "203.0.113.10", privateIP: "10.0.0.10"}, + "i-load": {publicIP: "203.0.113.20", privateIP: "10.0.0.20"}, + }) + require.NoError(t, err) + require.Equal(t, awsRoleValidator, hosts[0].Role) + require.Equal(t, "10.0.0.10", hosts[0].PrivateIP) + require.Equal(t, awsRoleLoad, hosts[1].Role) + require.Equal(t, "203.0.113.20", hosts[1].PublicIP) +} + +func TestPrometheusAndLoadConfigUsePrivateEVMEndpoints(t *testing.T) { + prom := prometheusScrapeConfig([]string{"10.0.0.10", "10.0.0.11"}) + require.Contains(t, prom, "10.0.0.10:26660") + require.Contains(t, prom, "10.0.0.11:26660") + + cfg, err := seiLoadAWSConfig([]string{"10.0.0.10", "10.0.0.11"}) + require.NoError(t, err) + require.Contains(t, cfg, "http://10.0.0.10:8545") + require.Contains(t, cfg, "http://10.0.0.11:8545") +} + +func TestGrafanaPublicURL(t *testing.T) { + require.Equal(t, "", grafanaPublicURL("")) + require.Equal(t, "http://203.0.113.10:3000", grafanaPublicURL("203.0.113.10")) +} + +func TestResolveGrafanaCIDR(t *testing.T) { + got, err := resolveGrafanaCIDR("", "198.51.100.4/32") + require.NoError(t, err) + require.Equal(t, "198.51.100.4/32", got) + + got, err = resolveGrafanaCIDR("0.0.0.0/0", "198.51.100.4/32") + require.NoError(t, err) + require.Equal(t, "0.0.0.0/0", got) + + _, err = resolveGrafanaCIDR("not-a-cidr", "198.51.100.4/32") + require.Error(t, err) +} + +func TestAWSDeployGrafanaCIDRCanBeWidened(t *testing.T) { + stateDir := t.TempDir() + runner := &fakeRunner{} + runner.outputFn = func(spec commandSpec) (string, error) { + joined := strings.Join(spec.args, " ") + switch { + case strings.Contains(joined, "sts get-caller-identity"): + return `{}`, nil + case strings.Contains(joined, "describe-vpcs"): + return "vpc-123\n", nil + case strings.Contains(joined, "create-security-group"): + return "sg-123\n", nil + case strings.Contains(joined, "create-key-pair"): + return "-----BEGIN OPENSSH PRIVATE KEY-----\ntest\n-----END OPENSSH PRIVATE KEY-----\n", nil + case strings.Contains(joined, "run-instances"): + if strings.Contains(joined, "Value=load") { + return "i-load\n", nil + } + return "i-v0\ti-v1\ti-v2\ti-v3\n", nil + case strings.Contains(joined, "describe-instances"): + return "i-v0\t203.0.113.10\t10.0.0.10\ni-v1\t203.0.113.11\t10.0.0.11\ni-v2\t203.0.113.12\t10.0.0.12\ni-v3\t203.0.113.13\t10.0.0.13\ni-load\t203.0.113.20\t10.0.0.20\n", nil + case spec.name == "ssh": + return "", nil + default: + return "", nil + } + } + app := &application{runner: runner, stdout: &bytes.Buffer{}, stderr: &bytes.Buffer{}, stateDir: stateDir} + require.NoError(t, app.deploy(context.Background(), deployOptions{ + name: "grafana-open", + target: "aws", + timeout: time.Minute, + region: "us-west-2", + instanceType: "r7i.12xlarge", + amiID: "ami-123", + sshCIDR: "198.51.100.4/32", + grafanaCIDR: "0.0.0.0/0", + sshUser: "ubuntu", + volumeSize: defaultVolumeSizeGiB, + volumeIOPS: defaultVolumeIOPS, + volumeThroughput: defaultVolumeThroughputMB, + repoURL: "https://github.com/sei-protocol/sei-chain.git", + ref: "deadbeef", + topology: awsTopologyDistributed, + })) + require.Contains(t, joinedCommands(runner.commands), "--cidr 0.0.0.0/0") + require.Contains(t, joinedCommands(runner.commands), "--port 3000") +} + func TestShellQuote(t *testing.T) { require.Equal(t, `'a'\''b'`, shellQuote("a'b")) } diff --git a/cmd/autobahn-e2e/deploy.go b/cmd/autobahn-e2e/deploy.go index 0ce3119c71..bbb2714f6f 100644 --- a/cmd/autobahn-e2e/deploy.go +++ b/cmd/autobahn-e2e/deploy.go @@ -12,24 +12,33 @@ import ( "github.com/spf13/cobra" ) -const dockerClusterSize = 4 +const ( + dockerClusterSize = 4 + defaultVolumeSizeGiB = 1024 + defaultVolumeIOPS = 10000 + defaultVolumeThroughputMB = 1000 +) type deployOptions struct { - name string - target string - timeout time.Duration - region string - profile string - instanceType string - amiID string - subnetID string - sshCIDR string - sshUser string - keyName string - sshKeyPath string - volumeSize int - repoURL string - ref string + name string + target string + timeout time.Duration + region string + profile string + instanceType string + amiID string + subnetID string + sshCIDR string + grafanaCIDR string + sshUser string + keyName string + sshKeyPath string + volumeSize int + volumeIOPS int + volumeThroughput int + repoURL string + ref string + topology string } func (a *application) newDeployCommand() *cobra.Command { @@ -44,19 +53,23 @@ func (a *application) newDeployCommand() *cobra.Command { flags := cmd.Flags() flags.StringVar(&options.name, "name", defaultClusterName, "cluster name") flags.StringVar(&options.target, "target", targetLocal, "deployment target: local or aws") - flags.DurationVar(&options.timeout, "timeout", 20*time.Minute, "deployment readiness timeout") + flags.DurationVar(&options.timeout, "timeout", 40*time.Minute, "deployment readiness timeout") flags.StringVar(&options.region, "region", "us-west-2", "AWS region") flags.StringVar(&options.profile, "profile", "", "AWS CLI profile") - flags.StringVar(&options.instanceType, "instance-type", "c7g.2xlarge", "EC2 instance type") - flags.StringVar(&options.amiID, "ami-id", "", "EC2 AMI ID; defaults to Ubuntu 24.04 ARM64") + flags.StringVar(&options.instanceType, "instance-type", "r7i.12xlarge", "EC2 instance type") + flags.StringVar(&options.amiID, "ami-id", "", "EC2 AMI ID; defaults to Ubuntu 24.04 AMD64") flags.StringVar(&options.subnetID, "subnet-id", "", "EC2 subnet; defaults to a default VPC subnet") flags.StringVar(&options.sshCIDR, "ssh-cidr", "", "CIDR allowed to SSH; defaults to the caller's public IP") + flags.StringVar(&options.grafanaCIDR, "grafana-cidr", "", "CIDR allowed to reach Grafana :3000; defaults to --ssh-cidr") flags.StringVar(&options.sshUser, "ssh-user", "ubuntu", "EC2 SSH user") flags.StringVar(&options.keyName, "key-name", "", "existing EC2 key pair name; omitted creates a managed key") flags.StringVar(&options.sshKeyPath, "ssh-key", "", "private key for --key-name") - flags.IntVar(&options.volumeSize, "volume-size", 100, "EC2 root volume size in GiB") + flags.IntVar(&options.volumeSize, "volume-size", defaultVolumeSizeGiB, "EC2 root volume size in GiB") + flags.IntVar(&options.volumeIOPS, "volume-iops", defaultVolumeIOPS, "EC2 root gp3 IOPS") + flags.IntVar(&options.volumeThroughput, "volume-throughput", defaultVolumeThroughputMB, "EC2 root gp3 throughput in MB/s") flags.StringVar(&options.repoURL, "repo-url", "", "Git repository cloned on EC2; defaults to origin") flags.StringVar(&options.ref, "ref", "", "Git ref deployed on EC2; defaults to the current commit") + flags.StringVar(&options.topology, "topology", awsTopologyDistributed, "AWS topology: distributed (one validator per EC2 plus a load host) or colocated (four Docker validators on one EC2)") return cmd } @@ -67,6 +80,13 @@ func (a *application) deploy(ctx context.Context, options deployOptions) error { if options.timeout <= 0 { return fmt.Errorf("--timeout must be positive") } + if options.target == targetAWS { + topology, err := normalizeAWSTopology(options.topology) + if err != nil { + return err + } + options.topology = topology + } exists, err := a.store().exists(options.name) if err != nil { return err @@ -151,6 +171,17 @@ func (a *application) deployLocal(ctx context.Context, options deployOptions) er return nil } +func normalizeAWSTopology(topology string) (string, error) { + switch topology { + case "", awsTopologyDistributed: + return awsTopologyDistributed, nil + case awsTopologyColocated: + return awsTopologyColocated, nil + default: + return "", fmt.Errorf("unsupported --topology %q; use %s or %s", topology, awsTopologyDistributed, awsTopologyColocated) + } +} + func waitForLaunchFile(ctx context.Context, path string, count int) error { ticker := time.NewTicker(time.Second) defer ticker.Stop() diff --git a/cmd/autobahn-e2e/forward.go b/cmd/autobahn-e2e/forward.go index 804ac86b9f..978dc88d60 100644 --- a/cmd/autobahn-e2e/forward.go +++ b/cmd/autobahn-e2e/forward.go @@ -60,13 +60,21 @@ func (a *application) forward(ctx context.Context, options forwardOptions) error if state.AWS == nil { return fmt.Errorf("aws metadata is missing") } - _, _ = fmt.Fprintf(a.stdout, "Forwarding %s to %s:8545 through %s. Press Ctrl-C to stop.\n", localAddress, node.Name, state.AWS.PublicIP) - baseArgs := sshBaseArgs(state) + host, ok := state.AWS.validatorByIndex(node.Index) + if !ok { + host = awsHost{PublicIP: state.AWS.PublicIP} + } + if host.PublicIP == "" { + return fmt.Errorf("validator %s has no public IP", node.Name) + } + evmPort := state.AWS.evmPort(node) + _, _ = fmt.Fprintf(a.stdout, "Forwarding %s to %s:%d through %s. Press Ctrl-C to stop.\n", localAddress, node.Name, evmPort, host.PublicIP) + baseArgs := sshBaseArgsTo(state, host) destination := baseArgs[len(baseArgs)-1] args := append(baseArgs[:len(baseArgs)-1], "-o", "ExitOnForwardFailure=yes", "-N", - "-L", fmt.Sprintf("%s:127.0.0.1:%d", localAddress, node.EVMHostPort), + "-L", fmt.Sprintf("%s:127.0.0.1:%d", localAddress, evmPort), destination, ) return a.runner.stream(ctx, commandSpec{name: "ssh", args: args}) diff --git a/cmd/autobahn-e2e/list.go b/cmd/autobahn-e2e/list.go index 1d114a4442..c7faab850f 100644 --- a/cmd/autobahn-e2e/list.go +++ b/cmd/autobahn-e2e/list.go @@ -24,6 +24,7 @@ type nodeReport struct { Status string `json:"status"` Height string `json:"height"` EVMTarget string `json:"evm_target"` + Dashboard string `json:"dashboard,omitempty"` InstanceID string `json:"instance_id,omitempty"` PublicIP string `json:"public_ip,omitempty"` } @@ -91,7 +92,21 @@ func (a *application) list(ctx context.Context, options listOptions) error { report.InstanceID, ) } - return writer.Flush() + if err := writer.Flush(); err != nil { + return err + } + printed := map[string]struct{}{} + for _, report := range reports { + if report.Dashboard == "" { + continue + } + if _, ok := printed[report.Dashboard]; ok { + continue + } + printed[report.Dashboard] = struct{}{} + _, _ = fmt.Fprintf(a.stdout, "\nDASHBOARD %s (admin / admin)\n", report.Dashboard) + } + return nil } func (a *application) inspectCluster(ctx context.Context, state clusterState) ([]nodeReport, error) { @@ -142,7 +157,8 @@ func (a *application) inspectAWSCluster(ctx context.Context, state clusterState) if state.AWS == nil { return nil, fmt.Errorf("aws metadata is missing") } - if state.AWS.InstanceID == "" { + ids := state.AWS.instanceIDs() + if len(ids) == 0 { reports := make([]nodeReport, len(state.Nodes)) for i, node := range state.Nodes { reports[i] = nodeReport{ @@ -151,7 +167,7 @@ func (a *application) inspectAWSCluster(ctx context.Context, state clusterState) Node: node.Name, Status: state.Status, Height: "-", - EVMTarget: fmt.Sprintf("SSH→127.0.0.1:%d", node.EVMHostPort), + EVMTarget: fmt.Sprintf("SSH→127.0.0.1:%d", state.AWS.evmPort(node)), } } return reports, nil @@ -160,27 +176,41 @@ func (a *application) inspectAWSCluster(ctx context.Context, state clusterState) if err := a.ensureAWSCredentials(ctx, client); err != nil { return nil, err } - instanceStatus, err := client.output(ctx, - "ec2", "describe-instances", - "--instance-ids", state.AWS.InstanceID, - "--query", "Reservations[0].Instances[0].State.Name", - "--output", "text", - ) + args := append([]string{"ec2", "describe-instances", "--instance-ids"}, ids...) + args = append(args, "--query", "Reservations[].Instances[].[InstanceId,State.Name]", "--output", "text") + value, err := client.output(ctx, args...) if err != nil { return nil, err } - instanceStatus = strings.TrimSpace(instanceStatus) + instanceStatus := map[string]string{} + for _, line := range strings.Split(strings.TrimSpace(value), "\n") { + fields := strings.Fields(line) + if len(fields) >= 2 { + instanceStatus[fields[0]] = fields[1] + } + } + dashboard := "" + if load, ok := state.AWS.loadHost(); ok { + dashboard = grafanaPublicURL(load.PublicIP) + } reports := make([]nodeReport, len(state.Nodes)) for i, node := range state.Nodes { - status := instanceStatus + host, ok := state.AWS.validatorByIndex(node.Index) + if !ok { + host = awsHost{InstanceID: state.AWS.InstanceID, PublicIP: state.AWS.PublicIP} + } + status := instanceStatus[host.InstanceID] + if status == "" { + status = state.Status + } height := "-" - if instanceStatus == "running" && state.AWS.PublicIP != "" { - value, inspectErr := a.runner.output(ctx, sshCommand(state, + if status == "running" && host.PublicIP != "" { + value, inspectErr := a.runner.output(ctx, sshCommandTo(state, host, "docker inspect --format '{{.State.Status}}' "+shellQuote(node.Container))) if inspectErr == nil { status = strings.TrimSpace(value) } - value, heightErr := a.runner.output(ctx, sshCommand(state, + value, heightErr := a.runner.output(ctx, sshCommandTo(state, host, "docker exec "+shellQuote(node.Container)+" curl -fsS http://127.0.0.1:26660/metrics")) if heightErr == nil { height = parseAutobahnExecutedHeight(value) @@ -192,9 +222,10 @@ func (a *application) inspectAWSCluster(ctx context.Context, state clusterState) Node: node.Name, Status: status, Height: height, - EVMTarget: fmt.Sprintf("SSH→127.0.0.1:%d", node.EVMHostPort), - InstanceID: state.AWS.InstanceID, - PublicIP: state.AWS.PublicIP, + EVMTarget: fmt.Sprintf("SSH→127.0.0.1:%d", state.AWS.evmPort(node)), + Dashboard: dashboard, + InstanceID: host.InstanceID, + PublicIP: host.PublicIP, } } return reports, nil diff --git a/cmd/autobahn-e2e/state.go b/cmd/autobahn-e2e/state.go index 582e8d7e48..658df4687b 100644 --- a/cmd/autobahn-e2e/state.go +++ b/cmd/autobahn-e2e/state.go @@ -32,19 +32,122 @@ type node struct { EVMHostPort int `json:"evm_host_port"` } +const ( + awsRoleValidator = "validator" + awsRoleLoad = "load" +) + +type awsHost struct { + Role string `json:"role"` + Index int `json:"index,omitempty"` + InstanceID string `json:"instance_id"` + PublicIP string `json:"public_ip"` + PrivateIP string `json:"private_ip,omitempty"` +} + type awsState struct { - Region string `json:"region"` - Profile string `json:"profile,omitempty"` - InstanceID string `json:"instance_id,omitempty"` - PublicIP string `json:"public_ip,omitempty"` - SecurityGroupID string `json:"security_group_id,omitempty"` - KeyName string `json:"key_name,omitempty"` - SSHKeyPath string `json:"ssh_key_path,omitempty"` - SSHUser string `json:"ssh_user"` - RemoteDir string `json:"remote_dir"` - ManagedKey bool `json:"managed_key"` - RepoURL string `json:"repo_url"` - Ref string `json:"ref"` + Region string `json:"region"` + Profile string `json:"profile,omitempty"` + Topology string `json:"topology,omitempty"` + InstanceID string `json:"instance_id,omitempty"` + PublicIP string `json:"public_ip,omitempty"` + Hosts []awsHost `json:"hosts,omitempty"` + SecurityGroupID string `json:"security_group_id,omitempty"` + KeyName string `json:"key_name,omitempty"` + SSHKeyPath string `json:"ssh_key_path,omitempty"` + SSHUser string `json:"ssh_user"` + RemoteDir string `json:"remote_dir"` + ManagedKey bool `json:"managed_key"` + RepoURL string `json:"repo_url"` + Ref string `json:"ref"` +} + +func (s *awsState) topology() string { + if s == nil { + return awsTopologyDistributed + } + switch s.Topology { + case awsTopologyColocated, awsTopologyDistributed: + return s.Topology + } + if len(s.validators()) >= 2 { + return awsTopologyDistributed + } + return awsTopologyColocated +} + +func (s *awsState) colocated() bool { + return s.topology() == awsTopologyColocated +} + +func (s *awsState) evmPort(n node) int { + if s.colocated() { + return n.EVMHostPort + } + return awsEVMPort +} + +func (s *awsState) validators() []awsHost { + if s == nil { + return nil + } + hosts := make([]awsHost, 0, len(s.Hosts)) + for _, host := range s.Hosts { + if host.Role == awsRoleValidator { + hosts = append(hosts, host) + } + } + sort.Slice(hosts, func(i, j int) bool { return hosts[i].Index < hosts[j].Index }) + return hosts +} + +func (s *awsState) loadHost() (awsHost, bool) { + if s == nil { + return awsHost{}, false + } + for _, host := range s.Hosts { + if host.Role == awsRoleLoad { + return host, true + } + } + if s.PublicIP != "" || s.InstanceID != "" { + return awsHost{Role: awsRoleLoad, InstanceID: s.InstanceID, PublicIP: s.PublicIP}, true + } + return awsHost{}, false +} + +func (s *awsState) validatorByIndex(index int) (awsHost, bool) { + for _, host := range s.validators() { + if host.Index == index { + return host, true + } + } + return awsHost{}, false +} + +func (s *awsState) instanceIDs() []string { + if s == nil { + return nil + } + if len(s.Hosts) > 0 { + ids := make([]string, 0, len(s.Hosts)) + seen := map[string]struct{}{} + for _, host := range s.Hosts { + if host.InstanceID == "" { + continue + } + if _, ok := seen[host.InstanceID]; ok { + continue + } + seen[host.InstanceID] = struct{}{} + ids = append(ids, host.InstanceID) + } + return ids + } + if s.InstanceID != "" { + return []string{s.InstanceID} + } + return nil } type stateStore struct { diff --git a/cmd/autobahn-e2e/teardown.go b/cmd/autobahn-e2e/teardown.go index 4f0235af94..46098c8dbd 100644 --- a/cmd/autobahn-e2e/teardown.go +++ b/cmd/autobahn-e2e/teardown.go @@ -59,18 +59,42 @@ func (a *application) teardownAWS(ctx context.Context, state clusterState) error if err := a.ensureAWSCredentials(ctx, client); err != nil { return err } - if state.AWS.PublicIP != "" && state.AWS.RemoteDir != "" { - command := "cd " + shellQuote(state.AWS.RemoteDir) + " && make docker-cluster-stop" - if err := a.runner.stream(ctx, sshCommand(state, command)); err != nil { - _, _ = fmt.Fprintf(a.stderr, "warning: remote Docker teardown failed: %v\n", err) + if state.AWS.RemoteDir != "" { + if state.AWS.colocated() { + if host, ok := state.AWS.loadHost(); ok && host.PublicIP != "" { + command := "if [ -d " + shellQuote(state.AWS.RemoteDir) + " ]; then cd " + shellQuote(state.AWS.RemoteDir) + " && make docker-cluster-stop-monitoring; fi" + if err := a.runner.stream(ctx, sshCommandTo(state, host, command)); err != nil { + _, _ = fmt.Fprintf(a.stderr, "warning: remote Docker teardown failed: %v\n", err) + } + } + } else { + for _, host := range state.AWS.validators() { + if host.PublicIP == "" { + continue + } + command := "if [ -d " + shellQuote(state.AWS.RemoteDir) + " ]; then cd " + shellQuote(state.AWS.RemoteDir) + " && make docker-aws-validator-stop; fi" + if err := a.runner.stream(ctx, sshCommandTo(state, host, command)); err != nil { + _, _ = fmt.Fprintf(a.stderr, "warning: remote validator teardown failed: %v\n", err) + } + } + if load, ok := state.AWS.loadHost(); ok && load.PublicIP != "" { + command := "if [ -d " + shellQuote(state.AWS.RemoteDir) + " ]; then cd " + shellQuote(state.AWS.RemoteDir) + " && make docker-aws-load-stop; fi" + if err := a.runner.stream(ctx, sshCommandTo(state, load, command)); err != nil { + _, _ = fmt.Fprintf(a.stderr, "warning: remote load-host teardown failed: %v\n", err) + } + } } } var errs []error - if state.AWS.InstanceID != "" { - if _, err := client.output(ctx, "ec2", "terminate-instances", "--instance-ids", state.AWS.InstanceID); err != nil { - errs = append(errs, err) - } else if err := client.stream(ctx, "ec2", "wait", "instance-terminated", "--instance-ids", state.AWS.InstanceID); err != nil { + if ids := state.AWS.instanceIDs(); len(ids) > 0 { + args := append([]string{"ec2", "terminate-instances", "--instance-ids"}, ids...) + if _, err := client.output(ctx, args...); err != nil { errs = append(errs, err) + } else { + waitArgs := append([]string{"ec2", "wait", "instance-terminated", "--instance-ids"}, ids...) + if err := client.stream(ctx, waitArgs...); err != nil { + errs = append(errs, err) + } } } if state.AWS.SecurityGroupID != "" { diff --git a/docker/docker-compose.aws-load.yml b/docker/docker-compose.aws-load.yml new file mode 100644 index 0000000000..8e784ea192 --- /dev/null +++ b/docker/docker-compose.aws-load.yml @@ -0,0 +1,28 @@ +services: + prometheus: + container_name: sei-prometheus + image: prom/prometheus:latest + ports: + - "9099:9090" + volumes: + - ${PROJECT_HOME}/build/generated/prometheus.yml:/etc/prometheus/prometheus.yml:ro + command: + - --config.file=/etc/prometheus/prometheus.yml + - --storage.tsdb.path=/prometheus + - --web.enable-lifecycle + + grafana: + container_name: sei-grafana + image: grafana/grafana:latest + ports: + - "0.0.0.0:3000:3000" + volumes: + - ./docker_compose_monitoring/grafana-datasource.yaml:/etc/grafana/provisioning/datasources/grafana-datasource.yaml:ro + - ./monitornode/config/grafana-dashboards.yaml:/etc/grafana/provisioning/dashboards/grafana-dashboards.yaml:ro + - ./monitornode/dashboards:/var/lib/grafana/dashboards:ro + environment: + - GF_SECURITY_ADMIN_USER=admin + - GF_SECURITY_ADMIN_PASSWORD=admin + - GF_USERS_ALLOW_SIGN_UP=false + depends_on: + - prometheus diff --git a/docker/docker-compose.aws-validator.yml b/docker/docker-compose.aws-validator.yml new file mode 100644 index 0000000000..4d0d8f82d1 --- /dev/null +++ b/docker/docker-compose.aws-validator.yml @@ -0,0 +1,37 @@ +services: + node: + platform: ${DOCKER_PLATFORM:-linux/amd64} + container_name: sei-node-${ID:-0} + image: "sei-chain/localnode" + user: "${USERID}:${GROUPID}" + ports: + - "26656:26656" + - "8545:8545" + - "26660:26660" + environment: + - ID=${ID:-0} + - CLUSTER_SIZE=4 + - ADVERTISE_IP + - AUTOBAHN_E2E_PHASE + - NUM_ACCOUNTS + - SKIP_BUILD + - INVARIANT_CHECK_INTERVAL + - UPGRADE_VERSION_LIST + - MOCK_BALANCES + - GIGA_EXECUTOR + - GIGA_OCC + - RECEIPT_BACKEND + - AUTOBAHN + - AUTOBAHN_EVMONLY + - GIGA_STORAGE + - GIGA_MIGRATE_FROM_MEMIAVL + - GIGA_FLATKV_ONLY + - GOCACHE=/tmp/go-cache + - GOMODCACHE=/tmp/go-mod + volumes: + - "${PROJECT_HOME}:/sei-protocol/sei-chain:Z" + - "${PROJECT_HOME}/../sei-tendermint:/sei-protocol/sei-tendermint:Z" + - "${PROJECT_HOME}/../sei-cosmos:/sei-protocol/sei-cosmos:Z" + - "${PROJECT_HOME}/../sei-db:/sei-protocol/sei-db:Z" + - "${PROJECT_HOME}/../go-ethereum:/sei-protocol/go-ethereum:Z" + - "${VALIDATOR_HOME}:/root/.sei:Z" diff --git a/docker/docker-compose.monitoring.yml b/docker/docker-compose.monitoring.yml index 84a3e6f06e..8f47720b5a 100644 --- a/docker/docker-compose.monitoring.yml +++ b/docker/docker-compose.monitoring.yml @@ -17,9 +17,11 @@ services: container_name: sei-grafana image: grafana/grafana:latest ports: - - "3000:3000" + - "0.0.0.0:3000:3000" volumes: - ./docker_compose_monitoring/grafana-datasource.yaml:/etc/grafana/provisioning/datasources/grafana-datasource.yaml:ro + - ./monitornode/config/grafana-dashboards.yaml:/etc/grafana/provisioning/dashboards/grafana-dashboards.yaml:ro + - ./monitornode/dashboards:/var/lib/grafana/dashboards:ro environment: - GF_SECURITY_ADMIN_USER=admin - GF_SECURITY_ADMIN_PASSWORD=admin diff --git a/docker/docker_compose_monitoring/grafana-datasource.yaml b/docker/docker_compose_monitoring/grafana-datasource.yaml index 774c799693..f3d48a8926 100644 --- a/docker/docker_compose_monitoring/grafana-datasource.yaml +++ b/docker/docker_compose_monitoring/grafana-datasource.yaml @@ -1,6 +1,8 @@ apiVersion: 1 datasources: + # The uid is pinned because the dashboards in docker/monitornode/dashboards reference it directly. - name: Prometheus + uid: PBFA97CFB590B2093 type: prometheus access: proxy url: http://prometheus:9090 diff --git a/docker/localnode/scripts/deploy.sh b/docker/localnode/scripts/deploy.sh index c2c1227891..26c4b384c4 100755 --- a/docker/localnode/scripts/deploy.sh +++ b/docker/localnode/scripts/deploy.sh @@ -1,7 +1,9 @@ #!/usr/bin/env sh +set -e NODE_ID=${ID:-0} CLUSTER_SIZE=${CLUSTER_SIZE:-1} +PHASE=${AUTOBAHN_E2E_PHASE:-all} # Clean up and env set up export GOPATH=$HOME/go @@ -19,56 +21,100 @@ mkdir -p $GOBIN # here races with other nodes writing init/genesis/launch coordination files. mkdir -p build/generated -# Step 0: Build on node 0 -if [ "$NODE_ID" = 0 ] && [ -z "$SKIP_BUILD" ] -then - /usr/bin/build.sh $MOCK_BALANCES -fi +ensure_seid() { + if [ -f build/seid ]; then + cp build/seid "$GOBIN"/ + fi +} -if ! [ "$SKIP_BUILD" ] -then - until [ -f build/generated/build.complete ] - do - sleep 1 - done -fi +run_build() { + if [ -n "$SKIP_BUILD" ] + then + return + fi + # Local 4-in-1 compose shares build/; only node 0 compiles. AWS init + # runs one node per host, so every PHASE=init container compiles itself. + if [ "$NODE_ID" = 0 ] || [ "$PHASE" = "init" ] + then + /usr/bin/build.sh $MOCK_BALANCES + fi + if [ "$PHASE" = "all" ] + then + until [ -f build/generated/build.complete ] + do + sleep 1 + done + fi +} -# Step 1: Run init on all nodes -/usr/bin/configure_init.sh +run_init() { + /usr/bin/configure_init.sh +} -# Step 2&3: Genesis on node 0 -if [ "$NODE_ID" = 0 ] -then - # wait for other nodes init complete - until [ -f build/generated/init.complete ] - do - sleep 1 - done - while [ $(cat build/generated/init.complete |wc -l) -lt "$CLUSTER_SIZE" ] - do - sleep 1 - done +run_genesis() { + if [ "$NODE_ID" != 0 ] + then + return + fi + if [ "$PHASE" = "all" ] + then + until [ -f build/generated/init.complete ] + do + sleep 1 + done + while [ $(cat build/generated/init.complete |wc -l) -lt "$CLUSTER_SIZE" ] + do + sleep 1 + done + fi echo "Running genesis on node 0" /usr/bin/genesis.sh -fi - -until [ -f build/generated/genesis.json ] -do - sleep 1 -done +} -# Step 4: Config overrides -/usr/bin/config_override.sh +run_start() { + until [ -f build/generated/genesis.json ] + do + sleep 1 + done -# Step 5: Start the chain -/usr/bin/start_sei.sh + /usr/bin/config_override.sh + /usr/bin/start_sei.sh -# Wait until the chain started -while [ $(cat build/generated/launch.complete |wc -l) -lt "$CLUSTER_SIZE" ] -do - sleep 1 -done -sleep 5 -echo "All $CLUSTER_SIZE Nodes started successfully." + if [ "$PHASE" = "all" ] + then + while [ $(cat build/generated/launch.complete |wc -l) -lt "$CLUSTER_SIZE" ] + do + sleep 1 + done + sleep 5 + echo "All $CLUSTER_SIZE Nodes started successfully." + fi + tail -f /dev/null +} -tail -f /dev/null +case "$PHASE" in + init) + run_build + run_init + echo "init phase complete for node $NODE_ID" + ;; + genesis) + ensure_seid + run_genesis + echo "genesis phase complete" + ;; + start) + ensure_seid + run_start + ;; + all) + run_build + run_init + run_genesis + run_start + ;; + *) + echo "unknown AUTOBAHN_E2E_PHASE=$PHASE" >&2 + exit 1 + ;; +esac diff --git a/docker/localnode/scripts/step1_configure_init.sh b/docker/localnode/scripts/step1_configure_init.sh index fa2a7f1681..1a5bff6788 100755 --- a/docker/localnode/scripts/step1_configure_init.sh +++ b/docker/localnode/scripts/step1_configure_init.sh @@ -33,7 +33,11 @@ cp docker/localnode/config/config.toml "$TENDERMINT_CONFIG_FILE" # Set up persistent peers SEI_NODE_ID=$(seid tendermint show-node-id) -NODE_IP=$(hostname -i | awk '{print $1}') +if [ -n "$ADVERTISE_IP" ]; then + NODE_IP="$ADVERTISE_IP" +else + NODE_IP=$(hostname -i | awk '{print $1}') +fi P2P_PORT=26656 # Must match [p2p] laddr in config.toml EVMRPC_PORT=8545 # Must match the EVM RPC HTTP port (evmrpc DefaultConfig HTTPPort). echo "$SEI_NODE_ID@$NODE_IP:$P2P_PORT" >> build/generated/persistent_peers.txt diff --git a/docker/localnode/scripts/step4_config_override.sh b/docker/localnode/scripts/step4_config_override.sh index 7523d00f73..1024997707 100755 --- a/docker/localnode/scripts/step4_config_override.sh +++ b/docker/localnode/scripts/step4_config_override.sh @@ -38,7 +38,11 @@ if [ "$VALIDATOR" != "true" ]; then fi # Override up persistent peers -NODE_IP=$(hostname -i | awk '{print $1}') +if [ -n "$ADVERTISE_IP" ]; then + NODE_IP="$ADVERTISE_IP" +else + NODE_IP=$(hostname -i | awk '{print $1}') +fi PEERS=$(cat build/generated/persistent_peers.txt |grep -v "$NODE_IP" | paste -sd "," -) sed -i'' -e 's/persistent-peers = ""/persistent-peers = "'$PEERS'"/g' ~/.sei/config/config.toml diff --git a/docker/monitornode/config/prometheus.yaml b/docker/monitornode/config/prometheus.yaml index 8d8ca8aced..d0e7e8fe26 100644 --- a/docker/monitornode/config/prometheus.yaml +++ b/docker/monitornode/config/prometheus.yaml @@ -40,3 +40,15 @@ scrape_configs: # target_label: instance # - target_label: __address__ # replacement: 'host.docker.internal:9300' + + # Autobahn e2e validators (seid prometheus-listen-addr). start-prometheus.sh attaches this + # container to the cluster network so these names resolve. + - job_name: 'autobahn-e2e' + metrics_path: /metrics + static_configs: + - targets: + - 'sei-node-0:26660' + - 'sei-node-1:26660' + - 'sei-node-2:26660' + - 'sei-node-3:26660' + scrape_interval: 5s diff --git a/docker/monitornode/dashboards/autobahn-e2e-dashboard.json b/docker/monitornode/dashboards/autobahn-e2e-dashboard.json new file mode 100644 index 0000000000..fe37ecd01c --- /dev/null +++ b/docker/monitornode/dashboards/autobahn-e2e-dashboard.json @@ -0,0 +1,334 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "description": "Autobahn EVM-only e2e: executed throughput and the main-loop split across consensus, execution, and storage.", + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "links": [], + "panels": [ + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 0 }, + "id": 1, + "panels": [], + "title": "Overview", + "type": "row" + }, + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [{ "color": "green", "value": null }] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { "h": 6, "w": 6, "x": 0, "y": 1 }, + "id": 2, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.4.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "editorMode": "code", + "expr": "avg(rate(tendermint_internal_autobahn_data_tx_size_count[$__rate_interval]))", + "legendFormat": "TPS", + "range": true, + "refId": "A" + } + ], + "title": "TPS", + "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [{ "color": "green", "value": null }] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { "h": 6, "w": 6, "x": 6, "y": 1 }, + "id": 3, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.4.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "editorMode": "code", + "expr": "avg(rate(tendermint_internal_autobahn_data_next_block{stage=\"execute\"}[$__rate_interval]))", + "legendFormat": "blocks/s", + "range": true, + "refId": "A" + } + ], + "title": "Blocks / sec", + "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [{ "color": "green", "value": null }] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { "h": 6, "w": 6, "x": 12, "y": 1 }, + "id": 4, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.4.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "editorMode": "code", + "expr": "avg(histogram_quantile(0.50, sum by (le, instance) (rate(tendermint_internal_autobahn_data_latency_bucket{resource=\"blocks\",stage=\"execute\"}[$__rate_interval]))))", + "legendFormat": "p50", + "range": true, + "refId": "A" + } + ], + "title": "Block finalize time (p50)", + "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [{ "color": "green", "value": null }] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { "h": 6, "w": 6, "x": 18, "y": 1 }, + "id": 5, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.4.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "editorMode": "code", + "expr": "avg(histogram_quantile(0.99, sum by (le, instance) (rate(tendermint_internal_autobahn_data_latency_bucket{resource=\"blocks\",stage=\"execute\"}[$__rate_interval]))))", + "legendFormat": "p99", + "range": true, + "refId": "A" + } + ], + "title": "Block finalize time (p99)", + "type": "stat" + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 7 }, + "id": 6, + "panels": [], + "title": "Latency breakdown", + "type": "row" + }, + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "description": "Share of the execute goroutine's wall clock. Consensus is the wait for the next committed block; execution is the EVM; storage is receipts, state commit, and the app commit. The three phases sum to 100%.", + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { + "hideFrom": { "legend": false, "tooltip": false, "viz": false } + }, + "mappings": [], + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { "h": 14, "w": 12, "x": 0, "y": 8 }, + "id": 7, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "values": ["percent"] + }, + "pieType": "pie", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "sort": "desc", + "tooltip": { "hideZeros": false, "mode": "single", "sort": "none" } + }, + "pluginVersion": "12.4.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "editorMode": "code", + "expr": "avg by (phase) (rate(sei_chain_autobahn_main_loop_phase_duration_seconds_total[$__rate_interval]))", + "legendFormat": "{{phase}}", + "range": true, + "refId": "A" + } + ], + "title": "Main loop — time spent", + "type": "piechart" + }, + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "description": "The same main-loop phases as the pie, stacked over time. A phase that stays near 1.0 s/s is occupying the execute goroutine.", + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 100, + "gradientMode": "none", + "hideFrom": { "legend": false, "tooltip": false, "viz": false }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { "type": "linear" }, + "showPoints": "never", + "showValues": false, + "spanNulls": false, + "stacking": { "group": "A", "mode": "normal" }, + "thresholdsStyle": { "mode": "off" } + }, + "mappings": [], + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { "h": 14, "w": 12, "x": 12, "y": 8 }, + "id": 8, + "options": { + "legend": { + "calcs": ["lastNotNull"], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { "hideZeros": false, "mode": "multi", "sort": "desc" } + }, + "pluginVersion": "12.4.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, + "editorMode": "code", + "expr": "avg by (phase) (rate(sei_chain_autobahn_main_loop_phase_duration_seconds_total[$__rate_interval]))", + "legendFormat": "{{phase}}", + "range": true, + "refId": "A" + } + ], + "title": "Main loop — time spent", + "type": "timeseries" + } + ], + "preload": false, + "refresh": "5s", + "schemaVersion": 42, + "tags": ["autobahn", "e2e"], + "templating": { "list": [] }, + "time": { "from": "now-15m", "to": "now" }, + "timepicker": {}, + "timezone": "browser", + "title": "Autobahn E2E", + "uid": "autobahn-e2e", + "version": 1, + "weekStart": "" +} diff --git a/docker/monitornode/scripts/start-prometheus.sh b/docker/monitornode/scripts/start-prometheus.sh index 98cdfd4fdb..37fd6081e0 100755 --- a/docker/monitornode/scripts/start-prometheus.sh +++ b/docker/monitornode/scripts/start-prometheus.sh @@ -34,8 +34,29 @@ if [[ ! -f "${PROMETHEUS_CONFIG}" ]]; then exit 1 fi +connect_autobahn_network() { + local net + net="$(docker inspect sei-node-0 --format '{{range $k, $v := .NetworkSettings.Networks}}{{$k}} {{end}}' 2>/dev/null | awk '{print $1}')" || true + if [[ -z "${net}" ]]; then + return 0 + fi + if docker inspect "$CONTAINER_NAME" --format '{{range $k, $v := .NetworkSettings.Networks}}{{$k}} {{end}}' 2>/dev/null | grep -qw "${net}"; then + return 0 + fi + echo "Attaching ${CONTAINER_NAME} to ${net} so Autobahn e2e nodes can be scraped." + docker network connect "${net}" "$CONTAINER_NAME" +} + +reload_prometheus() { + if curl -fsS -X POST "http://127.0.0.1:${PROMETHEUS_UI_PORT}/-/reload" >/dev/null 2>&1; then + echo "Reloaded Prometheus scrape config." + fi +} + # If container exists and is running, we're done if docker ps -q -f "name=^${CONTAINER_NAME}$" | grep -q .; then + connect_autobahn_network + reload_prometheus echo "Prometheus is already running." echo " UI: http://localhost:${PROMETHEUS_UI_PORT}" exit 0 @@ -45,6 +66,8 @@ fi if docker ps -aq -f "name=^${CONTAINER_NAME}$" | grep -q .; then echo "Starting existing Prometheus container..." docker start "$CONTAINER_NAME" + connect_autobahn_network + reload_prometheus echo "" echo "Prometheus is running." echo " UI: http://localhost:${PROMETHEUS_UI_PORT}" @@ -69,6 +92,8 @@ docker run -d \ --storage.tsdb.path=/prometheus \ --web.enable-lifecycle +connect_autobahn_network + echo "" echo "Prometheus is running." echo " UI: http://localhost:${PROMETHEUS_UI_PORT}" diff --git a/giga/evmonly/giga_store.go b/giga/evmonly/giga_store.go index 7fc68b366d..7c7190eec4 100644 --- a/giga/evmonly/giga_store.go +++ b/giga/evmonly/giga_store.go @@ -8,6 +8,7 @@ import ( "github.com/ethereum/go-ethereum/common" + gigametrics "github.com/sei-protocol/sei-chain/giga/metrics" "github.com/sei-protocol/sei-chain/sei-db/proto" gigatypes "github.com/sei-protocol/sei-chain/sei-db/state_db/giga/types" ) @@ -54,6 +55,7 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar if err := ctx.Err(); err != nil { return nil, err } + gigametrics.SetPhase(gigametrics.PhaseExecution) snapshot := stateStore.OpenView() if snapshot == nil { return nil, errors.New("giga store returned a nil snapshot") @@ -77,6 +79,7 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar if err := ctx.Err(); err != nil { return nil, err } + gigametrics.SetPhase(gigametrics.PhaseStorage) changesets, err := e.changeSetEncoder(result.ChangeSet) if err != nil { return nil, fmt.Errorf("encode state changes for block %d: %w", req.Context.Number, err) diff --git a/giga/metrics/autobahn_loop.go b/giga/metrics/autobahn_loop.go new file mode 100644 index 0000000000..72ece1efdc --- /dev/null +++ b/giga/metrics/autobahn_loop.go @@ -0,0 +1,66 @@ +package metrics + +import ( + "sync" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/otlptranslator" + seidbmetrics "github.com/sei-protocol/sei-chain/sei-db/common/metrics" + "go.opentelemetry.io/otel" + otelprometheus "go.opentelemetry.io/otel/exporters/prometheus" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" +) + +const ( + meterName = "autobahn" + timerName = "autobahn_main_loop" + + // PhaseConsensus is time spent waiting for the next committed Autobahn block. + PhaseConsensus = "consensus" + // PhaseExecution is time spent executing that block's transactions. + PhaseExecution = "execution" + // PhaseStorage is time spent persisting receipts, state, and the app commit. + PhaseStorage = "storage" +) + +var ( + setupOnce sync.Once + setupErr error + + loopOnce sync.Once + loop *seidbmetrics.PhaseTimer + loopMu sync.Mutex +) + +// SetupPrometheus installs a Prometheus MeterProvider on the default registerer. +func SetupPrometheus() error { + setupOnce.Do(func() { + exporter, err := otelprometheus.New( + otelprometheus.WithRegisterer(prometheus.DefaultRegisterer), + otelprometheus.WithTranslationStrategy(otlptranslator.UnderscoreEscapingWithSuffixes), + ) + if err != nil { + setupErr = err + return + } + otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(exporter))) + }) + return setupErr +} + +// MainLoop is the phase timer for Autobahn's execute loop. +func MainLoop() *seidbmetrics.PhaseTimer { + loopOnce.Do(func() { + loop = seidbmetrics.NewPhaseTimerFactory(otel.Meter(meterName), timerName). + RecordLatencies(). + Build() + }) + return loop +} + +// SetPhase records a transition on Autobahn's execute-loop timer. +func SetPhase(phase string) { + loopMu.Lock() + defer loopMu.Unlock() + MainLoop().SetPhase(phase) +} diff --git a/giga/metrics/autobahn_loop_test.go b/giga/metrics/autobahn_loop_test.go new file mode 100644 index 0000000000..a93da983c6 --- /dev/null +++ b/giga/metrics/autobahn_loop_test.go @@ -0,0 +1,29 @@ +package metrics + +import ( + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestSetupPrometheusIsIdempotent(t *testing.T) { + require.NoError(t, SetupPrometheus()) + require.NoError(t, SetupPrometheus()) + require.NotNil(t, MainLoop()) +} + +func TestSetPhaseIsSafeForConcurrentCallers(t *testing.T) { + require.NoError(t, SetupPrometheus()) + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + SetPhase(PhaseConsensus) + SetPhase(PhaseExecution) + SetPhase(PhaseStorage) + }() + } + wg.Wait() +} diff --git a/integration_test/autobahn/README.md b/integration_test/autobahn/README.md index db139f77be..e8359e865b 100644 --- a/integration_test/autobahn/README.md +++ b/integration_test/autobahn/README.md @@ -1,8 +1,10 @@ # Autobahn EVM-only E2E clusters `autobahn-e2e` manages the four-validator, disk-backed EVM-only Autobahn -topology used for integration and load testing. It can run the topology in -local Docker or on one AWS EC2 instance. +topology used for integration and load testing. Locally it runs all four +validators in Docker on one host. On AWS it places each validator on its +own EC2 instance and adds a fifth instance that scrapes metrics. Start +`sei-load` on that host when you want traffic. Run every command in this document from the root of a `sei-chain` checkout. The examples use the default cluster name, `autobahn-evmonly`. If `--name` is @@ -69,19 +71,55 @@ replace existing `sei-node-*` containers or existing manager metadata. ## Start a cluster on AWS -The AWS target creates one Ubuntu EC2 host and runs the same four-node Docker -topology on it. Only SSH is opened in the managed security group. EVM JSON-RPC -stays private and is accessed through `forward`. +AWS deploy has two topologies, selected with `--topology`: + +- `distributed` (default): five Ubuntu EC2 hosts — four validators and one + load/monitoring instance. Each validator runs a single `seid` container + that advertises the instance's private IP. The four validators clone, + compile, and initialize in parallel. The load instance is brought up + afterward with Prometheus and Grafana; `sei-load` is left for you to + start. The security group admits SSH from the caller, Grafana (`:3000`) + from the internet, and all TCP between the five instances. +- `colocated`: one Ubuntu EC2 host running the same four-container Docker + topology used locally, plus Prometheus and Grafana. Use this when you + want the cheaper single-instance setup. + +EVM JSON-RPC stays off the public internet and is accessed through +`forward`. ```sh ./autobahn-e2e deploy --target aws \ --name my-autobahn \ - --region us-west-2 + --region us-west-2 \ + --topology distributed + +./autobahn-e2e deploy --target aws \ + --name my-autobahn-colo \ + --region us-west-2 \ + --topology colocated ./autobahn-e2e list --name my-autobahn ``` -In another terminal, forward one node to the load-generator host: +Deploy prints the public Grafana URL (`http://:3000`, admin / +admin). `list` repeats it under `DASHBOARD`. Open **Autobahn E2E**. The +login is the default Grafana pair on a temporary test host; tear the +cluster down when finished. + +Deploy writes `integration_test/autobahn/sei-load.aws.json` on the load +instance with `http://:8545` for every validator. +It does not start `sei-load`. When you want traffic, SSH in and run it: + +```sh +ssh -i ~/.sei/autobahn-e2e/my-autobahn.pem ubuntu@ +cd ~/sei-chain-my-autobahn +GOBIN="$PWD/build/tools" go install github.com/sei-protocol/sei-load@v0.0.1 +./build/tools/sei-load \ + --config integration_test/autobahn/sei-load.aws.json \ + --metricsListenAddr 0.0.0.0:19698 +``` + +In another terminal, forward one validator to the laptop: ```sh ./autobahn-e2e forward \ @@ -90,11 +128,10 @@ In another terminal, forward one node to the load-generator host: --local-port 18545 ``` -One forwarded endpoint is sufficient: validator EVM proxying is enabled by -default, so transactions submitted to node 0 are forwarded to the Autobahn -validator that owns the sender's shard. To distribute load across all four -entry points, start four `forward` processes with distinct local ports and put -all four URLs in the `sei-load` configuration. +One forwarded endpoint is sufficient for `cast` and similar tools: +validator EVM proxying is enabled by default, so transactions submitted +to node 0 are forwarded to the Autobahn validator that owns the sender's +shard. AWS credentials use the AWS CLI credential chain. Use `--profile NAME` to select a profile. If no credentials work in an interactive terminal, the @@ -111,14 +148,20 @@ manager state directory with mode `0600`. To use an existing key pair instead: --ssh-key ~/.ssh/my-key-pair.pem ``` -The default security-group rule admits SSH only from the public IP detected at -deployment time. Use `--ssh-cidr` when a VPN, NAT, or IPv6 setup makes that -incorrect. Use `--subnet-id` if the region has no default VPC or the instance -needs a specific public subnet. - -The default instance is `c7g.2xlarge` with 100 GiB of gp3 storage and the -current Ubuntu 24.04 ARM64 AMI from AWS Systems Manager. When changing -architecture, override `--instance-type` and `--ami-id` together. +The default security-group rule admits SSH and Grafana from the public IP +detected at deployment time. Use `--ssh-cidr` when a VPN, NAT, or IPv6 setup +makes that source incorrect. Use `--grafana-cidr` to widen Grafana +independently (for example `0.0.0.0/0`). Use `--subnet-id` if the +region has no default VPC or the instance needs a specific public subnet. + +The default validator instance is `r7i.12xlarge` with 1024 GiB of gp3 storage +(10000 IOPS, 1000 MB/s) and the +current Ubuntu 24.04 AMD64 AMI from AWS Systems Manager. The load instance +uses the same AMI and instance type with a 100 GiB gp3 root volume. Override +the validator disk with `--volume-size`, `--volume-iops`, and +`--volume-throughput`. When changing architecture, override `--instance-type` +and `--ami-id` together. `--timeout` defaults to 40 minutes to cover the +image build, `seid` compile, and five-instance bootstrap. `--repo-url` and `--ref` select the source built remotely; they default to this checkout's origin and current commit. The selected commit must be reachable from the EC2 host, so uncommitted local changes are not deployed. @@ -203,8 +246,11 @@ GOBIN="$PWD/build/tools" go install github.com/sei-protocol/sei-load@v0.0.1 ``` The checked-in [`sei-load.local.json`](sei-load.local.json) is a ready local -four-endpoint configuration. For AWS with the single tunnel shown above, copy -it and change `endpoints` to only `http://127.0.0.1:18545`. +four-endpoint configuration. An AWS deploy writes +`integration_test/autobahn/sei-load.aws.json` on the load instance with the +four private EVM URLs and leaves `sei-load` stopped. To drive load from +the laptop instead, copy the local file and point `endpoints` at one or +more `forward` tunnels. Start load and press Ctrl-C to stop it cleanly: @@ -279,6 +325,34 @@ current EVM-only RPC. `sei-load` implements receipt tracking by subscribing to new heads and fetching blocks, rather than polling individual receipts, and those methods are not exposed yet. +## Watch the dashboard + +An AWS deploy starts Prometheus and Grafana on the load instance and prints +a URL reachable from the same CIDR as SSH. Open that address (admin / admin) +and select **Autobahn E2E**. Prometheus scrapes each validator at +`:26660`. + +For a local cluster, start the monitornode containers after the nodes are +up. Prometheus scrapes each validator at `:26660` and Grafana provisions +**Autobahn E2E** from `docker/monitornode/dashboards`. + +```sh +docker/monitornode/scripts/start-prometheus.sh +docker/monitornode/scripts/start-grafana.sh +``` + +Open http://localhost:3000 (admin / admin) and select **Autobahn E2E**. +The overview line is executed TPS, blocks/sec, and produce-to-execute +finalize time. The pie and stacked line are the execute goroutine split +across consensus wait, EVM execution, and storage. + +If Prometheus was already running from a gigasim or cryptosim session, +run `start-prometheus.sh` again after the cluster is up so it joins the +node network and reloads scrape targets. + +`make docker-cluster-start-monitoring` provisions the same dashboard +(Grafana at http://localhost:3000, Prometheus UI at http://localhost:9099). + ## Interact with a running cluster Inspect node health and the last executed height at any time: @@ -379,7 +453,7 @@ Stop the local containers and remove their manager metadata: ./autobahn-e2e teardown --name autobahn-evmonly ``` -Stop an AWS cluster and remove the EC2 instance, security group, managed key +Stop an AWS cluster and remove the five EC2 instances, security group, managed key pair, local managed private key, and manager metadata: ```sh diff --git a/sei-db/bench/gigasim/README.md b/sei-db/bench/gigasim/README.md index 329bf6878d..bb1ab10909 100644 --- a/sei-db/bench/gigasim/README.md +++ b/sei-db/bench/gigasim/README.md @@ -5,6 +5,11 @@ state DB alone, gigasim runs both together with the receipt store, through the s engines behave *together* — whether pruning, checkpointing and hashing on one store show up as latency on another. +Gigasim does not start validators or accept RPC traffic. For a four-validator Autobahn EVM-only +cluster, local or on AWS, use [`autobahn-e2e`](../../../integration_test/autobahn/README.md). AWS +deploy takes `--topology distributed` (default: one validator per EC2, plus a load/monitoring host) +or `--topology colocated` (all four Docker validators on a single EC2). + # Running Gigasim Run from anywhere in the repository; the script builds what it needs first: diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go index 0e4d44f746..57b8a746f2 100644 --- a/sei-tendermint/internal/p2p/giga_router_common.go +++ b/sei-tendermint/internal/p2p/giga_router_common.go @@ -11,6 +11,7 @@ import ( "sync/atomic" ethrpc "github.com/ethereum/go-ethereum/rpc" + gigametrics "github.com/sei-protocol/sei-chain/giga/metrics" "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/hashvault" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" @@ -230,7 +231,6 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo proposerAddress = key.Address() } - // TODO: add metrics to understand execution latency. resp, err := app.FinalizeBlock(ctx, &abci.RequestFinalizeBlock{ Txs: b.Payload.Txs(), // Empty DecidedLastCommit does not indicate missing votes. @@ -252,6 +252,8 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo return nil, fmt.Errorf("app.FinalizeBlock(): %w", err) } + gigametrics.SetPhase(gigametrics.PhaseStorage) + // Commit this height's app hash to the equivocation guard before persisting app state, so the // vault always records our commitment to a height before the state it implies is committed (and // before the hash is proposed for AppQC voting via PushAppHash below). On restart the block is @@ -457,10 +459,12 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { } for n := next; ; n += 1 { + gigametrics.SetPhase(gigametrics.PhaseConsensus) b, err := r.data.GlobalBlock(ctx, n) if err != nil { return fmt.Errorf("r.data.GlobalBlock(%v): %w", n, err) } + gigametrics.SetPhase(gigametrics.PhaseExecution) commitResp, err := r.executeBlock(ctx, b, hashVault) if err != nil { return fmt.Errorf("r.executeBlock(%v): %w", n, err)