ci: ARC load test [do not merge]

tests/fixture/tmpnet/network.go

@@ -11,7 +11,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"net/netip"
+	"net/url"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -1113,3 +1115,51 @@ func MetricsLinkForNetwork(networkUUID string, startTime string, endTime string)
 		endTime,
 	)
 }
+
+// GrafanaFilterOptions contains filters to apply to Grafana link in form of a query
+// Example: https://grafana-poc.avax-dev.network/?start_time=now-1h&endTime=now
+type GrafanaFilterOptions struct {
+	StartTime string
+	EndTime   string
+	Filters   map[string]string
+}
+
+func BuildMonitoringURLForNetwork(dashboardID, dashboardName, networkUUID string, options GrafanaFilterOptions) string {
+	// Set defaults for options if not provided
+	startTime := "now-1h"
+	if options.StartTime != "" {
+		startTime = options.StartTime
+	}
+
+	endTime := "now"
+	if options.EndTime != "" {
+		endTime = options.EndTime
+	}
+
+	baseURL := url.URL{
+		Scheme: "https",
+		Host:   grafanaURI,
+		Path:   fmt.Sprintf("/d/%s/%s", dashboardID, dashboardName),
+	}
+
+	query := baseURL.Query()
+
+	query.Add("from", startTime)
+	query.Add("to", endTime)
+
+	filters := make(map[string]string)
+	if options.Filters != nil {
+		filters = maps.Clone(options.Filters)
+	}
+
+	if _, exists := filters["network_uuid"]; !exists {
+		filters["network_uuid"] = networkUUID
+	}
+
+	for key, value := range filters {
+		query.Add("var-filter", fmt.Sprintf("%s|=|%s", key, value))
+	}
+
+	baseURL.RawQuery = query.Encode()
+	return baseURL.String()
+}

tests/fixture/tmpnet/process_runtime.go

@@ -372,7 +372,7 @@ func (p *ProcessRuntime) writeMonitoringConfig() error {
 func (p *ProcessRuntime) getMonitoringConfigPath(name string) (string, error) {
 	// Ensure a unique filename to allow config files to be added and removed
 	// by multiple nodes without conflict.
-	serviceDiscoveryDir, err := getServiceDiscoveryDir(name)
+	serviceDiscoveryDir, err := GetServiceDiscoveryDir(name)
 	if err != nil {
 		return "", err
 	}

tests/load/c/execute.go

@@ -0,0 +1,126 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package c
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+
+	"github.com/ava-labs/libevm/common"
+	"github.com/ava-labs/libevm/ethclient"
+
+	"github.com/ava-labs/avalanchego/tests/load"
+	"github.com/ava-labs/avalanchego/tests/load/c/listener"
+	"github.com/ava-labs/avalanchego/utils/crypto/secp256k1"
+	"github.com/ava-labs/avalanchego/utils/logging"
+)
+
+type loadConfig struct {
+	endpoints []string
+	agents    uint
+	minTPS    int64
+	maxTPS    int64
+	step      int64
+}
+
+func execute(
+	ctx context.Context,
+	keys []*secp256k1.PrivateKey,
+	config loadConfig,
+	metrics *load.Metrics,
+	logger logging.Logger,
+) error {
+	tracker := load.NewTracker[common.Hash](metrics)
+	agents, err := createAgents(ctx, config, keys, tracker)
+	if err != nil {
+		return fmt.Errorf("creating agents: %w", err)
+	}
+
+	orchestratorCtx, orchestratorCancel := context.WithCancel(ctx)
+	defer orchestratorCancel()
+	orchestratorConfig := load.NewOrchestratorConfig()
+	orchestratorConfig.MinTPS = config.minTPS
+	orchestratorConfig.MaxTPS = config.maxTPS
+	orchestratorConfig.Step = config.step
+	orchestratorConfig.TxRateMultiplier = 1.1
+	orchestrator := load.NewOrchestrator(agents, tracker, logger, orchestratorConfig)
+
+	return orchestrator.Execute(orchestratorCtx)
+}
+
+// createAgents creates agents for the given configuration and keys.
+// It creates them in parallel because creating issuers can sometimes take a while,
+// and this adds up for many agents. For example, deploying the Opcoder contract
+// takes a few seconds. Running the creation in parallel can reduce the time significantly.
+func createAgents(
+	ctx context.Context,
+	config loadConfig,
+	keys []*secp256k1.PrivateKey,
+	tracker *load.Tracker[common.Hash],
+) ([]load.Agent[common.Hash], error) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	type result struct {
+		agent load.Agent[common.Hash]
+		err   error
+	}
+	ch := make(chan result)
+	for i := range int(config.agents) {
+		key := keys[i]
+		endpoint := config.endpoints[i%len(config.endpoints)]
+		go func(key *secp256k1.PrivateKey, endpoint string) {
+			agent, err := createAgent(ctx, endpoint, key, tracker)
+			ch <- result{agent: agent, err: err}
+		}(key, endpoint)
+	}
+
+	var err error
+	agents := make([]load.Agent[common.Hash], 0, int(config.agents))
+	for range int(config.agents) {
+		result := <-ch
+		switch {
+		case result.err == nil && err == nil: // no previous error or new error
+			agents = append(agents, result.agent)
+		case err != nil: // error already occurred
+			continue
+		case result.err != nil: // first error
+			err = result.err
+			cancel()
+		default:
+			panic("unreachable")
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+	return agents, nil
+}
+
+func createAgent(
+	ctx context.Context,
+	endpoint string,
+	key *secp256k1.PrivateKey,
+	tracker *load.Tracker[common.Hash],
+) (load.Agent[common.Hash], error) {
+	client, err := ethclient.DialContext(ctx, endpoint)
+	if err != nil {
+		return load.Agent[common.Hash]{}, fmt.Errorf("dialing %s: %w", endpoint, err)
+	}
+
+	address := key.EthAddress()
+	blockNumber := (*big.Int)(nil)
+	nonce, err := client.NonceAt(ctx, address, blockNumber)
+	if err != nil {
+		return load.Agent[common.Hash]{}, fmt.Errorf("getting nonce for address %s: %w", address, err)
+	}
+
+	issuer, err := createIssuer(ctx, client, nonce, key.ToECDSA())
+	if err != nil {
+		return load.Agent[common.Hash]{}, fmt.Errorf("creating issuer: %w", err)
+	}
+	listener := listener.New(client, tracker, address, nonce)
+	return load.NewAgent(issuer, listener), nil
+}

tests/load/c/load_test.go

@@ -0,0 +1,117 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package c
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ava-labs/avalanchego/tests"
+	"github.com/ava-labs/avalanchego/tests/fixture/e2e"
+	"github.com/ava-labs/avalanchego/tests/fixture/tmpnet"
+	"github.com/ava-labs/avalanchego/tests/load"
+)
+
+const (
+	blockchainID = "C"
+	// invariant: nodesCount >= 5
+	nodesCount    = 5
+	agentsPerNode = 50
+	agentsCount   = nodesCount * agentsPerNode
+	logPrefix     = "avalanchego-load-test"
+)
+
+var flagVars *e2e.FlagVars
+
+func init() {
+	flagVars = e2e.RegisterFlagsWithDefaultOwner("avalanchego-load")
+
+	// Disable default metrics link generation to prevent duplicate links.
+	// We generate load specific links.
+	e2e.EmitMetricsLink = false
+}
+
+func TestLoad(t *testing.T) {
+	require := require.New(t)
+	ctx := context.Background()
+	log := tests.NewDefaultLogger(logPrefix)
+	tc := tests.NewTestContext(log)
+
+	startTime := time.Now()
+	nodes := tmpnet.NewNodesOrPanic(nodesCount)
+	network := &tmpnet.Network{
+		Owner: "avalanchego-load-test",
+		Nodes: nodes,
+	}
+
+	setPrefundedKeys(t, network, agentsCount)
+
+	testEnv := e2e.NewTestEnvironment(tc, flagVars, network)
+	defer func() {
+		require.NoError(network.Stop(ctx), "failed to stop network")
+	}()
+
+	registry := prometheus.NewRegistry()
+	metrics, err := load.NewMetrics(registry)
+	require.NoError(err, "failed to register load metrics")
+
+	metricsServer := load.NewPrometheusServer("127.0.0.1:0", registry)
+	merticsErrCh, err := metricsServer.Start()
+	require.NoError(err, "failed to start load metrics server")
+
+	monitoringConfigFilePath, err := metricsServer.GenerateMonitoringConfig(network.UUID)
+	require.NoError(err, "failed to generate monitoring config file")
+
+	defer func() {
+		select {
+		case err := <-merticsErrCh:
+			require.NoError(err, "metrics server exited with error")
+		default:
+			require.NoError(metricsServer.Stop(), "failed to stop metrics server")
+		}
+	}()
+
+	defer func() {
+		require.NoError(
+			os.Remove(monitoringConfigFilePath),
+			"failed †o remove monitoring config file",
+		)
+	}()
+
+	endpoints, err := tmpnet.GetNodeWebsocketURIs(ctx, network.Nodes, blockchainID)
+	require.NoError(err, "failed †o get node websocket URIs")
+	config := loadConfig{
+		endpoints: endpoints,
+		agents:    agentsPerNode,
+		minTPS:    100,
+		maxTPS:    500,
+		step:      100,
+	}
+
+	require.NoError(
+		execute(ctx, network.PreFundedKeys, config, metrics, log),
+		"failed to execute load test",
+	)
+
+	load.GenerateMetricsLink(testEnv, log, startTime)
+}
+
+// setPrefundedKeys sets the pre-funded keys for the network, and keeps
+// keys already set if any. If there are more keys than required, it
+// keeps the already set keys as they are.
+func setPrefundedKeys(t require.TestingT, network *tmpnet.Network, minKeys int) {
+	if len(network.PreFundedKeys) >= minKeys {
+		return
+	}
+
+	require := require.New(t)
+	missingPreFundedKeys, err := tmpnet.NewPrivateKeys(minKeys - len(network.PreFundedKeys))
+	require.NoError(err, "creating pre-funded keys")
+	network.PreFundedKeys = append(network.PreFundedKeys, missingPreFundedKeys...)
+}

tests/load/c/prometheus.template.yml

@@ -0,0 +1,17 @@
+global:
+  scrape_interval: 10s
+  evaluation_interval: 5s
+  scrape_timeout: 5s
+
+scrape_configs:
+  - job_name: "load"
+    metrics_path: "/ext/metrics"
+    static_configs:
+      - targets: ["localhost:8082"]
+
+  - job_name: "tmpnet"
+    metrics_path: "/ext/metrics"
+    file_sd_configs:
+      - files:
+          - "${HOME}/.tmpnet/prometheus/file_sd_configs/*.json"
+        refresh_interval: 1m