feat: add localized ledger explorer dashboard

2026-08-15 00:47:56 +03:30
parent ce5f8b4a84
commit 5b4cb5a2a3
31 changed files with 5458 additions and 42 deletions
@@ -0,0 +1,165 @@
+// Command dashboard-loadtest runs a concurrent read-only HTTP load test against
+// a deployed GL dashboard.
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"os/signal"
+	"sort"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+)
+
+type result struct {
+	Duration time.Duration
+	Requests int64
+	Errors   int64
+	Non2xx   int64
+	Latency  []time.Duration
+}
+
+func main() {
+	baseURL := flag.String("url", "http://127.0.0.1:8080", "dashboard base URL")
+	paths := flag.String("paths", "/", "comma-separated request paths")
+	concurrency := flag.Int("concurrency", 20, "number of concurrent workers")
+	duration := flag.Duration("duration", 15*time.Second, "test duration")
+	requestTimeout := flag.Duration("request-timeout", 5*time.Second, "timeout for each request")
+	flag.Parse()
+
+	if *concurrency < 1 || *duration <= 0 || *requestTimeout <= 0 {
+		fmt.Fprintln(os.Stderr, "concurrency and durations must be positive")
+		os.Exit(2)
+	}
+	targets, err := targets(*baseURL, *paths)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(2)
+	}
+
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+	transport.MaxIdleConns = *concurrency
+	transport.MaxIdleConnsPerHost = *concurrency
+	client := &http.Client{Transport: transport, Timeout: *requestTimeout}
+	defer transport.CloseIdleConnections()
+
+	signalCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	ctx, cancel := context.WithTimeout(signalCtx, *duration)
+	defer cancel()
+
+	fmt.Printf("Dashboard load test: %d workers for %s against %s\n", *concurrency, *duration, *baseURL)
+	started := time.Now()
+	result := run(ctx, client, targets, *concurrency)
+	result.Duration = time.Since(started)
+	printResult(result)
+	if result.Errors > 0 || result.Non2xx > 0 {
+		os.Exit(1)
+	}
+}
+
+func targets(baseURL, pathList string) ([]string, error) {
+	base, err := url.Parse(baseURL)
+	if err != nil || base.Scheme == "" || base.Host == "" {
+		return nil, fmt.Errorf("invalid dashboard URL %q", baseURL)
+	}
+	paths := strings.Split(pathList, ",")
+	result := make([]string, 0, len(paths))
+	for _, value := range paths {
+		value = strings.TrimSpace(value)
+		if value == "" {
+			continue
+		}
+		reference, parseErr := url.Parse(value)
+		if parseErr != nil || reference.IsAbs() || !strings.HasPrefix(reference.Path, "/") {
+			return nil, fmt.Errorf("invalid request path %q", value)
+		}
+		result = append(result, base.ResolveReference(reference).String())
+	}
+	if len(result) == 0 {
+		return nil, errors.New("at least one request path is required")
+	}
+	return result, nil
+}
+
+func run(ctx context.Context, client *http.Client, targets []string, concurrency int) result {
+	var (
+		requests int64
+		failures int64
+		non2xx   int64
+		sequence uint64
+		latency  = make([]time.Duration, 0, concurrency*100)
+		lock     sync.Mutex
+		workers  sync.WaitGroup
+	)
+	workers.Add(concurrency)
+	for range concurrency {
+		go func() {
+			defer workers.Done()
+			for ctx.Err() == nil {
+				target := targets[(atomic.AddUint64(&sequence, 1)-1)%uint64(len(targets))]
+				request, err := http.NewRequestWithContext(context.WithoutCancel(ctx), http.MethodGet, target, nil)
+				if err != nil {
+					atomic.AddInt64(&failures, 1)
+					continue
+				}
+				started := time.Now()
+				response, err := client.Do(request)
+				elapsed := time.Since(started)
+				if err != nil {
+					atomic.AddInt64(&requests, 1)
+					atomic.AddInt64(&failures, 1)
+					continue
+				}
+				_, copyErr := io.Copy(io.Discard, io.LimitReader(response.Body, 2<<20))
+				closeErr := response.Body.Close()
+				atomic.AddInt64(&requests, 1)
+				if response.StatusCode < 200 || response.StatusCode >= 300 {
+					atomic.AddInt64(&non2xx, 1)
+				}
+				if copyErr != nil || closeErr != nil {
+					atomic.AddInt64(&failures, 1)
+				}
+				lock.Lock()
+				latency = append(latency, elapsed)
+				lock.Unlock()
+			}
+		}()
+	}
+	workers.Wait()
+	return result{Requests: requests, Errors: failures, Non2xx: non2xx, Latency: latency}
+}
+
+func printResult(value result) {
+	seconds := value.Duration.Seconds()
+	requestsPerSecond := float64(value.Requests)
+	if seconds > 0 {
+		requestsPerSecond /= seconds
+	}
+	fmt.Printf("requests: %d\n", value.Requests)
+	fmt.Printf("throughput: %.1f req/s\n", requestsPerSecond)
+	fmt.Printf("latency: p50=%s p95=%s p99=%s\n", percentile(value.Latency, 50), percentile(value.Latency, 95), percentile(value.Latency, 99))
+	fmt.Printf("errors: %d transport, %d non-2xx\n", value.Errors, value.Non2xx)
+}
+
+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]
+}
@@ -0,0 +1,68 @@
+package main
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestRunSendsConcurrentRequestsAcrossTargets(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) {
+		response.WriteHeader(http.StatusOK)
+		_, _ = response.Write([]byte("ok"))
+	}))
+	defer server.Close()
+
+	requestTargets, err := targets(server.URL, "/,/transactions")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	result := run(ctx, server.Client(), requestTargets, 4)
+
+	if result.Requests == 0 || result.Errors != 0 || result.Non2xx != 0 {
+		t.Fatalf("unexpected load result: %+v", result)
+	}
+	if len(result.Latency) != int(result.Requests) {
+		t.Fatalf("expected one latency per request, got %d for %d", len(result.Latency), result.Requests)
+	}
+}
+
+func TestTargetsRejectsExternalAndRelativePaths(t *testing.T) {
+	for _, paths := range []string{"relative", "https://example.com/"} {
+		if _, err := targets("http://localhost:8080", paths); err == nil {
+			t.Fatalf("expected %q to be rejected", paths)
+		}
+	}
+}
+
+func TestPercentileUsesNearestRank(t *testing.T) {
+	values := []time.Duration{time.Millisecond, 4 * time.Millisecond, 2 * time.Millisecond, 3 * time.Millisecond}
+	if got := percentile(values, 95); got != 4*time.Millisecond {
+		t.Fatalf("unexpected p95: %s", got)
+	}
+}
+
+func TestRunDrainsInflightRequestAfterDuration(t *testing.T) {
+	var canceled atomic.Bool
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		time.Sleep(20 * time.Millisecond)
+		if request.Context().Err() != nil {
+			canceled.Store(true)
+		}
+		response.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Millisecond)
+	defer cancel()
+	result := run(ctx, server.Client(), []string{server.URL}, 1)
+
+	if result.Requests != 1 || result.Errors != 0 || canceled.Load() {
+		t.Fatalf("in-flight request was not drained cleanly: %+v", result)
+	}
+}
@@ -0,0 +1,54 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"gl/application/explorer"
+	"gl/infrastructure/config"
+	"gl/infrastructure/postgres"
+	webadapter "gl/interface/web"
+)
+
+func main() {
+	configPath := flag.String("conf", "./dashboard.cfg.toml", "path to the dashboard TOML configuration file")
+	flag.Parse()
+
+	cfg, err := config.LoadDashboard(*configPath)
+	if err != nil {
+		slog.Error("load dashboard configuration", "error", err)
+		os.Exit(1)
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	database, err := postgres.Open(ctx, cfg.Database)
+	if err != nil {
+		slog.Error("open GL database", "error", err)
+		os.Exit(1)
+	}
+	defer database.Close()
+	if err := database.Ping(ctx); err != nil {
+		slog.Error("connect to GL database", "error", err)
+		os.Exit(1)
+	}
+
+	repository := postgres.NewJournalRepository(database)
+	handler := webadapter.NewHandler(explorer.NewService(repository))
+	serverConfig := webadapter.ServerConfig{
+		Host:              cfg.HTTP.Host,
+		Port:              cfg.HTTP.Port,
+		ReadHeaderTimeout: cfg.HTTP.ReadHeaderTimeout,
+		ShutdownTimeout:   cfg.HTTP.ShutdownTimeout,
+	}
+	slog.Info("starting GL explorer", "host", cfg.HTTP.Host, "port", cfg.HTTP.Port)
+	if err := webadapter.Run(ctx, serverConfig, handler); err != nil {
+		slog.Error("GL explorer stopped", "error", err)
+		os.Exit(1)
+	}
+}
@@ -39,7 +39,8 @@ func main() {
 		os.Exit(1)
 	}

-	handler := grpcadapter.NewHandler(health.NewService(database), applicationledger.NewService(postgres.NewJournalRepository(database), nil))
+	repository := postgres.NewJournalRepository(database)
+	handler := grpcadapter.NewHandler(health.NewService(database), applicationledger.NewService(repository, nil))
 	slog.Info("starting GL gRPC service", "host", cfg.GRPC.Host, "port", cfg.GRPC.Port)
 	serverConfig := grpcadapter.ServerConfig{
 		Host:            cfg.GRPC.Host,