324 lines
11 KiB
Go
324 lines
11 KiB
Go
// Command gl-loadtest runs a stateful gRPC load test against the GL service.
|
|
// Wallet accounts are created implicitly by the first journal that references
|
|
// each generated owner ID, matching normal GL behavior.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"flag"
|
|
"fmt"
|
|
"os"
|
|
"os/signal"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"syscall"
|
|
"time"
|
|
|
|
domain "gl/domain/ledger"
|
|
basev1 "gl/gen/base/v1"
|
|
ledgerv1 "gl/gen/ledger/v1"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
)
|
|
|
|
type client interface {
|
|
AppendJournal(context.Context, *ledgerv1.AppendJournalRequest, ...grpc.CallOption) (*ledgerv1.AppendJournalResponse, error)
|
|
AppendTransactionEvent(context.Context, *ledgerv1.AppendTransactionEventRequest, ...grpc.CallOption) (*ledgerv1.AppendTransactionEventResponse, error)
|
|
GetJournal(context.Context, *ledgerv1.GetJournalRequest, ...grpc.CallOption) (*ledgerv1.Journal, error)
|
|
GetBalance(context.Context, *ledgerv1.GetBalanceRequest, ...grpc.CallOption) (*ledgerv1.GetBalanceResponse, error)
|
|
}
|
|
|
|
type config struct {
|
|
RunID string
|
|
Wallets int
|
|
AssetID int64
|
|
Amount string
|
|
RequestTimeout time.Duration
|
|
}
|
|
|
|
type sample struct {
|
|
Operation string
|
|
Latency time.Duration
|
|
Err error
|
|
}
|
|
|
|
type operationStats struct {
|
|
Requests int64
|
|
Errors int64
|
|
Latency []time.Duration
|
|
}
|
|
|
|
func main() {
|
|
target := flag.String("target", "127.0.0.1:8600", "GL gRPC target")
|
|
concurrency := flag.Int("concurrency", 20, "number of concurrent transaction workers")
|
|
duration := flag.Duration("duration", 30*time.Second, "load duration")
|
|
wallets := flag.Int("wallets", 1000, "number of simulated wallet accounts")
|
|
assetID := flag.Int64("asset-id", 1, "asset ID used by transfers")
|
|
amount := flag.String("amount", "1", "canonical decimal transfer amount")
|
|
requestTimeout := flag.Duration("request-timeout", 5*time.Second, "timeout for each gRPC call")
|
|
runID := flag.String("run-id", "", "idempotency namespace; generated when empty")
|
|
flag.Parse()
|
|
|
|
if *concurrency < 1 || *duration <= 0 || *wallets < 2 || *assetID <= 0 || *requestTimeout <= 0 || *amount == "" {
|
|
fmt.Fprintln(os.Stderr, "concurrency and durations must be positive; wallets must be at least 2; asset-id and amount are required")
|
|
os.Exit(2)
|
|
}
|
|
parsedAmount, err := domain.ParseAmount(*amount)
|
|
if err != nil || parsedAmount.IsZero() || strings.HasPrefix(*amount, "-") {
|
|
fmt.Fprintf(os.Stderr, "amount must be a positive canonical ledger decimal: %v\n", err)
|
|
os.Exit(2)
|
|
}
|
|
if *runID == "" {
|
|
*runID = "gl-load-" + strconv.FormatInt(time.Now().UTC().UnixNano(), 36)
|
|
}
|
|
|
|
connection, err := grpc.NewClient(*target, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "create gRPC client: %v\n", err)
|
|
os.Exit(2)
|
|
}
|
|
defer connection.Close()
|
|
ledger := ledgerv1.NewGeneralLedgerServiceClient(connection)
|
|
healthCtx, cancelHealth := context.WithTimeout(context.Background(), *requestTimeout)
|
|
health, err := ledger.Health(healthCtx, &basev1.Empty{})
|
|
cancelHealth()
|
|
if err != nil || !health.GetServing() || !health.GetDatabaseReady() {
|
|
fmt.Fprintf(os.Stderr, "GL is not ready at %s: response=%v error=%v\n", *target, health, err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
testConfig := config{
|
|
RunID: *runID,
|
|
Wallets: *wallets,
|
|
AssetID: *assetID,
|
|
Amount: *amount,
|
|
RequestTimeout: *requestTimeout,
|
|
}
|
|
signalCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
ctx, cancel := context.WithTimeout(signalCtx, *duration)
|
|
defer cancel()
|
|
|
|
fmt.Printf("GL load test: target=%s workers=%d duration=%s wallets=%d asset=%d run=%s\n", *target, *concurrency, *duration, *wallets, *assetID, *runID)
|
|
started := time.Now()
|
|
samples, transactions := run(ctx, ledger, testConfig, *concurrency)
|
|
elapsed := time.Since(started)
|
|
failed := printSummary(samples, transactions, elapsed)
|
|
if failed {
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
func run(ctx context.Context, ledger client, cfg config, concurrency int) ([]sample, uint64) {
|
|
var sequence atomic.Uint64
|
|
results := make(chan []sample, concurrency)
|
|
var workers sync.WaitGroup
|
|
workers.Add(concurrency)
|
|
for range concurrency {
|
|
go func() {
|
|
defer workers.Done()
|
|
workerSamples := make([]sample, 0, 256)
|
|
for ctx.Err() == nil {
|
|
transaction := sequence.Add(1)
|
|
workerSamples = append(workerSamples, runTransaction(ctx, ledger, cfg, transaction)...)
|
|
}
|
|
results <- workerSamples
|
|
}()
|
|
}
|
|
workers.Wait()
|
|
close(results)
|
|
|
|
all := make([]sample, 0)
|
|
for workerSamples := range results {
|
|
all = append(all, workerSamples...)
|
|
}
|
|
return all, sequence.Load()
|
|
}
|
|
|
|
func runTransaction(ctx context.Context, ledger client, cfg config, sequence uint64) []sample {
|
|
transactionID := fmt.Sprintf("%s-%d", cfg.RunID, sequence)
|
|
senderID := fmt.Sprintf("load-wallet-%d", sequence%uint64(cfg.Wallets))
|
|
receiverID := fmt.Sprintf("load-wallet-%d", (sequence+1)%uint64(cfg.Wallets))
|
|
correlationID := "load-correlation-" + transactionID
|
|
blockchainHash := transactionHash(transactionID)
|
|
occurredAt := timestamppb.Now()
|
|
|
|
created := &ledgerv1.AppendTransactionEventRequest{
|
|
SourceService: "gl-loadtest",
|
|
IdempotencyKey: "load:" + transactionID + ":event:v1",
|
|
SourceTransactionId: transactionID,
|
|
TrackingCode: "load-" + strconv.FormatUint(sequence, 10),
|
|
EventVersion: 1,
|
|
State: ledgerv1.TransactionState_TRANSACTION_STATE_CREATED,
|
|
OccurredAt: occurredAt,
|
|
CorrelationId: correlationID,
|
|
ActorId: senderID,
|
|
Metadata: map[string]string{"load_test": cfg.RunID},
|
|
}
|
|
samples := []sample{measure(ctx, cfg.RequestTimeout, "append_event", func(callCtx context.Context) error {
|
|
_, err := ledger.AppendTransactionEvent(callCtx, created)
|
|
return err
|
|
})}
|
|
|
|
journalRequest := &ledgerv1.AppendJournalRequest{
|
|
SourceService: "gl-loadtest",
|
|
IdempotencyKey: "load:" + transactionID + ":transfer:v1",
|
|
SourceTransactionId: transactionID,
|
|
TrackingCode: created.TrackingCode,
|
|
EffectKind: "internal-transfer",
|
|
EventVersion: 1,
|
|
OccurredAt: occurredAt,
|
|
CorrelationId: correlationID,
|
|
ActorId: senderID,
|
|
Blockchain: &ledgerv1.BlockchainReference{
|
|
Network: "load-test",
|
|
TransactionHash: blockchainHash,
|
|
LedgerSequence: strconv.FormatUint(sequence, 10),
|
|
},
|
|
Metadata: map[string]string{"load_test": cfg.RunID},
|
|
Entries: []*ledgerv1.JournalEntry{
|
|
{
|
|
LineNumber: 1,
|
|
Account: walletAccount(senderID, cfg.AssetID),
|
|
Amount: "-" + cfg.Amount,
|
|
Description: "load-test transfer debit",
|
|
},
|
|
{
|
|
LineNumber: 2,
|
|
Account: walletAccount(receiverID, cfg.AssetID),
|
|
Amount: cfg.Amount,
|
|
Description: "load-test transfer credit",
|
|
},
|
|
},
|
|
}
|
|
var journalID string
|
|
samples = append(samples, measure(ctx, cfg.RequestTimeout, "append_journal", func(callCtx context.Context) error {
|
|
response, err := ledger.AppendJournal(callCtx, journalRequest)
|
|
if err == nil && response.GetJournal() != nil {
|
|
journalID = response.GetJournal().GetJournalId()
|
|
}
|
|
return err
|
|
}))
|
|
|
|
if journalID != "" {
|
|
samples = append(samples, measure(ctx, cfg.RequestTimeout, "get_journal", func(callCtx context.Context) error {
|
|
_, err := ledger.GetJournal(callCtx, &ledgerv1.GetJournalRequest{
|
|
Lookup: &ledgerv1.GetJournalRequest_JournalId{JournalId: journalID},
|
|
})
|
|
return err
|
|
}))
|
|
}
|
|
|
|
samples = append(samples, measure(ctx, cfg.RequestTimeout, "get_balance", func(callCtx context.Context) error {
|
|
_, err := ledger.GetBalance(callCtx, &ledgerv1.GetBalanceRequest{Account: walletAccount(receiverID, cfg.AssetID)})
|
|
return err
|
|
}))
|
|
|
|
successful := &ledgerv1.AppendTransactionEventRequest{
|
|
SourceService: "gl-loadtest",
|
|
IdempotencyKey: "load:" + transactionID + ":event:v2",
|
|
SourceTransactionId: transactionID,
|
|
TrackingCode: created.TrackingCode,
|
|
EventVersion: 2,
|
|
State: ledgerv1.TransactionState_TRANSACTION_STATE_SUCCESSFUL,
|
|
OccurredAt: timestamppb.Now(),
|
|
CorrelationId: correlationID,
|
|
ActorId: senderID,
|
|
Blockchain: journalRequest.Blockchain,
|
|
Metadata: map[string]string{"load_test": cfg.RunID},
|
|
}
|
|
samples = append(samples, measure(ctx, cfg.RequestTimeout, "append_event", func(callCtx context.Context) error {
|
|
_, err := ledger.AppendTransactionEvent(callCtx, successful)
|
|
return err
|
|
}))
|
|
return samples
|
|
}
|
|
|
|
func walletAccount(ownerID string, assetID int64) *ledgerv1.AccountReference {
|
|
return &ledgerv1.AccountReference{
|
|
AccountClass: ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE,
|
|
OwnerType: "user",
|
|
OwnerId: ownerID,
|
|
AssetId: assetID,
|
|
}
|
|
}
|
|
|
|
func transactionHash(transactionID string) string {
|
|
value := sha256.Sum256([]byte(transactionID))
|
|
return hex.EncodeToString(value[:])
|
|
}
|
|
|
|
func measure(ctx context.Context, timeout time.Duration, operation string, call func(context.Context) error) sample {
|
|
callCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout)
|
|
defer cancel()
|
|
started := time.Now()
|
|
err := call(callCtx)
|
|
return sample{Operation: operation, Latency: time.Since(started), Err: err}
|
|
}
|
|
|
|
func printSummary(samples []sample, transactions uint64, elapsed time.Duration) bool {
|
|
stats := make(map[string]*operationStats)
|
|
var failed bool
|
|
for _, value := range samples {
|
|
operation := stats[value.Operation]
|
|
if operation == nil {
|
|
operation = &operationStats{}
|
|
stats[value.Operation] = operation
|
|
}
|
|
operation.Requests++
|
|
operation.Latency = append(operation.Latency, value.Latency)
|
|
if value.Err != nil {
|
|
operation.Errors++
|
|
failed = true
|
|
}
|
|
}
|
|
|
|
names := make([]string, 0, len(stats))
|
|
for name := range stats {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
fmt.Printf("transactions: %d (%.1f tx/s)\n", transactions, rate(float64(transactions), elapsed))
|
|
fmt.Printf("%-18s %10s %8s %10s %12s %12s %12s\n", "operation", "requests", "errors", "req/s", "p50", "p95", "p99")
|
|
for _, name := range names {
|
|
value := stats[name]
|
|
fmt.Printf("%-18s %10d %8d %10.1f %12s %12s %12s\n",
|
|
name,
|
|
value.Requests,
|
|
value.Errors,
|
|
rate(float64(value.Requests), elapsed),
|
|
percentile(value.Latency, 50),
|
|
percentile(value.Latency, 95),
|
|
percentile(value.Latency, 99),
|
|
)
|
|
}
|
|
return failed
|
|
}
|
|
|
|
func rate(count float64, elapsed time.Duration) float64 {
|
|
if elapsed <= 0 {
|
|
return 0
|
|
}
|
|
return count / elapsed.Seconds()
|
|
}
|
|
|
|
func percentile(values []time.Duration, percent int) time.Duration {
|
|
if len(values) == 0 {
|
|
return 0
|
|
}
|
|
ordered := append([]time.Duration(nil), values...)
|
|
sort.Slice(ordered, func(left, right int) bool { return ordered[left] < ordered[right] })
|
|
index := (len(ordered)*percent + 99) / 100
|
|
if index < 1 {
|
|
index = 1
|
|
}
|
|
return ordered[index-1]
|
|
}
|