internal/controller/dashboard.go - master - t3/cli

internal/controller/dashboard.go
545 строк15 KB
Ivan
fix metrics
09 июл 2026, 16:16
09 июл 2026, 16:16009b39c
package controller

import (
	"context"
	"fmt"
	"math"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	pb "gitverse.ru/t3/cli/pkg/pb"
)

// agentMetrics holds the latest snapshot for a single agent.
type agentMetrics struct {
	hostname        string
	address         string
	weight          int
	activeUsers     int64
	totalIterations int64
	errorIterations int64
	customMetrics   []*pb.CustomMetric
}

// dashboard manages the live TUI display.
type dashboard struct {
	agents      []*agentMetrics
	stopCh      chan struct{}
	stopped     atomic.Bool
	mu          sync.RWMutex
	currentStep string
	stepIndex   int
	totalSteps  int
	startTime   time.Time
}

func newDashboard(agentCount int) *dashboard {
	d := &dashboard{
		agents: make([]*agentMetrics, agentCount),
		stopCh: make(chan struct{}),
	}
	for i := range d.agents {
		d.agents[i] = &agentMetrics{}
	}
	return d
}

// setStartTime records the test start time.
func (d *dashboard) setStartTime(t time.Time) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.startTime = t
}

// setCurrentStep sets the current step name and position.
func (d *dashboard) setCurrentStep(name string, index, total int) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.currentStep = name
	d.stepIndex = index
	d.totalSteps = total
}

// setAgentSnapshot updates the latest snapshot for agent idx.
func (d *dashboard) setAgentSnapshot(idx int, snapshot *pb.MetricsSnapshot) {
	d.agents[idx].hostname = snapshot.Hostname
	d.agents[idx].activeUsers = snapshot.ActiveUsers
	d.agents[idx].totalIterations = snapshot.TotalIterations
	d.agents[idx].errorIterations = snapshot.ErrorIterations
	d.agents[idx].customMetrics = snapshot.CustomMetrics
}

// agentsSnapshot returns the current agent metrics state.
func (d *dashboard) agentsSnapshot() []*agentMetrics {
	return d.agents
}

// setAddress sets the full address (host:port) for agent idx (called once at startup).
func (d *dashboard) setAddress(idx int, address string) {
	d.agents[idx].address = address
}

// setWeight sets the weight for agent idx (called once at startup).
func (d *dashboard) setWeight(idx int, weight int) {
	d.agents[idx].weight = weight
}

// start begins the live dashboard update loop.
func (d *dashboard) start(ctx context.Context, agents []*agentConn) {
	go func() {
		// Wait a moment for first metrics to arrive
		time.Sleep(200 * time.Millisecond)

		ticker := time.NewTicker(500 * time.Millisecond)
		defer ticker.Stop()

		for {
			select {
			case <-ctx.Done():
				d.printFinal(agents)
				return
			case <-d.stopCh:
				d.renderFinal()
				return
			case <-ticker.C:
				// Non-blocking read from all agent metricsCh
				for i, a := range agents {
					if a == nil {
						continue
					}
					select {
					case snapshot := <-a.metricsCh:
						d.setAgentSnapshot(i, snapshot)
					default:
					}
				}
				d.render()
			}
		}
	}()
}

// stop stops the dashboard.
func (d *dashboard) stop() {
	if d.stopped.CompareAndSwap(false, true) {
		close(d.stopCh)
	}
}

// formatDuration formats a duration as HH:MM:SS or MM:SS.
func formatDuration(d time.Duration) string {
	d = d.Round(time.Second)
	h := int(d.Hours())
	m := int(d.Minutes()) % 60
	s := int(d.Seconds()) % 60
	if h > 0 {
		return fmt.Sprintf("%02d:%02d:%02d", h, m, s)
	}
	return fmt.Sprintf("%02d:%02d", m, s)
}

// render clears and redraws the full dashboard.
func (d *dashboard) render() {
	var buf strings.Builder

	// Clear screen and move cursor to top
	buf.WriteString("\033[H\033[J")

	d.mu.RLock()
	currentStep := d.currentStep
	stepIndex := d.stepIndex
	totalSteps := d.totalSteps
	startTime := d.startTime
	d.mu.RUnlock()

	// Header — widened to 90 chars
	buf.WriteString("\033[1;36m") // bold cyan
	buf.WriteString("╔════════════════════════════════════════════════════════════════════════════════════════╗\n")
	buf.WriteString("║                              T3 — Live Metrics Dashboard                               ║\n")

	// Step info and elapsed time
	if !startTime.IsZero() {
		elapsed := time.Since(startTime)
		elapsedStr := formatDuration(elapsed)
		stepStr := ""
		if currentStep != "" {
			if totalSteps > 0 {
				stepStr = fmt.Sprintf("Step: %s (%d/%d)", currentStep, stepIndex, totalSteps)
			} else {
				stepStr = fmt.Sprintf("Step: %s", currentStep)
			}
		}
		buf.WriteString("\033[0m") // reset
		// Inner width between ║ chars is 88; fixed "  " prefix + "  " suffix = 4
		left := fmt.Sprintf("║  %s", stepStr)
		right := fmt.Sprintf("Elapsed: %s  ", elapsedStr)
		padding := 86 - len(stepStr) - len(right)
		if padding < 1 {
			padding = 1
		}
		buf.WriteString(left)
		buf.WriteString(strings.Repeat(" ", padding))
		buf.WriteString(right)
		buf.WriteString("║\n")
	}

	buf.WriteString("\033[1;36m") // bold cyan
	buf.WriteString("╠══════════════════════╤════════╪════════════╪══════════════╪══════════╪═════════════════╣\n")
	buf.WriteString("║        Agent         │ Weight │ Act. Users │  Iterations  │  Errors  │     Errors, %   ║\n")
	buf.WriteString("╠══════════════════════╪════════╪════════════╪══════════════╪══════════╪═════════════════╣\n")
	buf.WriteString("\033[0m") // reset

	for _, m := range d.agents {
		if m.hostname == "" {
			continue
		}
		errPct := 0.0
		if m.totalIterations > 0 {
			errPct = float64(m.errorIterations) / float64(m.totalIterations) * 100.0
		}

		displayName := truncateString(m.address, 20)
		if displayName == "" {
			displayName = truncateString(m.hostname, 20)
		}
		weight := fmt.Sprintf("%d", m.weight)
		activeUsers := fmt.Sprintf("%d", m.activeUsers)
		totalIter := fmt.Sprintf("%d", m.totalIterations)
		errCount := fmt.Sprintf("%d", m.errorIterations)
		errPctStr := fmt.Sprintf("%.1f %%", errPct)

		// Color error percentage red if > 5%
		if errPct > 5.0 {
			buf.WriteString("\033[0;31m") // red
		}

		fmt.Fprintf(&buf, "║ %-20s │ %-6s │ %-10s │ %-12s │ %-8s │ %-15s ║\n",
			displayName, weight, activeUsers, totalIter, errCount, errPctStr)

		buf.WriteString("\033[0m") // reset
	}

	buf.WriteString("\033[1;36m") // bold cyan
	buf.WriteString("╚══════════════════════╧════════╧════════════╧══════════════╧══════════╧═════════════════╝\n")
	buf.WriteString("\033[0m") // reset

	// Custom metrics section (only if any agent has custom metrics)
	d.renderCustomMetrics(&buf)

	// Footer with timestamp
	buf.WriteString(fmt.Sprintf("\033[2mLast updated: %s\033[0m\n", time.Now().Format("15:04:05")))

	fmt.Print(buf.String())
}

// aggregatedMetric holds aggregated values for a unique (name, type, tags) pair
// across all agents.
type aggregatedMetric struct {
	name, typ string
	tags      map[string]string
	// counter
	counterSum int64
	// gauge
	gaugeMin, gaugeMax, gaugeSum float64
	gaugeCount                   int
	// histogram
	histoCount                   int64
	histoSum, histoMin, histoMax float64
	histoFirst                   bool
	histoQuantiles               map[int32]float64 // percentage -> value, aggregated across agents (latest wins)
}

// aggregateCustomMetrics collects custom metrics from all agents and
// aggregates them by (name, type, tags), returning a sorted slice.
func (d *dashboard) aggregateCustomMetrics() []*aggregatedMetric {
	m := make(map[string]*aggregatedMetric)
	for _, a := range d.agents {
		for _, cm := range a.customMetrics {
			tagStr := tagMapString(cm.Tags)
			key := cm.Name + "|" + cm.Type + "|" + tagStr
			agg, ok := m[key]
			if !ok {
				agg = &aggregatedMetric{
					name:     cm.Name,
					typ:      cm.Type,
					tags:     cloneTags(cm.Tags),
					gaugeMin: math.MaxFloat64,
					gaugeMax: -math.MaxFloat64,
					histoMin: math.MaxFloat64,
					histoMax: -math.MaxFloat64,
				}
				m[key] = agg
			}
			switch cm.Type {
			case "counter":
				agg.counterSum += cm.Value.GetCounter()
			case "gauge":
				v := cm.Value.GetGauge()
				agg.gaugeSum += v
				agg.gaugeCount++
				if v < agg.gaugeMin {
					agg.gaugeMin = v
				}
				if v > agg.gaugeMax {
					agg.gaugeMax = v
				}
			case "histogram":
				h := cm.Value.GetHistogram()
				if h == nil {
					continue
				}
				agg.histoCount += h.Count
				agg.histoSum += h.Sum
				if !agg.histoFirst {
					agg.histoMin = h.Min
					agg.histoMax = h.Max
					agg.histoFirst = true
				} else {
					if h.Min < agg.histoMin {
						agg.histoMin = h.Min
					}
					if h.Max > agg.histoMax {
						agg.histoMax = h.Max
					}
				}
				// Aggregate quantiles: take the first non-nil set we see
				if agg.histoQuantiles == nil && len(h.Quantiles) > 0 {
					agg.histoQuantiles = make(map[int32]float64, len(h.Quantiles))
					for k, v := range h.Quantiles {
						agg.histoQuantiles[k] = v
					}
				}
			}
		}
	}

	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	result := make([]*aggregatedMetric, 0, len(m))
	for _, k := range keys {
		result = append(result, m[k])
	}
	return result
}

// tagMapString produces a deterministic string representation of a tags map.
func tagMapString(tags map[string]string) string {
	if len(tags) == 0 {
		return ""
	}
	keys := make([]string, 0, len(tags))
	for k := range tags {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	var b strings.Builder
	for _, k := range keys {
		if b.Len() > 0 {
			b.WriteByte(',')
		}
		b.WriteString(k)
		b.WriteByte('=')
		b.WriteString(tags[k])
	}
	return b.String()
}

// cloneTags returns a copy of the tags map.
func cloneTags(src map[string]string) map[string]string {
	if len(src) == 0 {
		return nil
	}
	dst := make(map[string]string, len(src))
	for k, v := range src {
		dst[k] = v
	}
	return dst
}

// formatAggregatedValue formats an aggregated metric value for display.
func formatAggregatedValue(agg *aggregatedMetric) string {
	switch agg.typ {
	case "counter":
		return fmt.Sprintf("%d", agg.counterSum)
	case "gauge":
		if agg.gaugeCount == 0 {
			return "no data"
		}
		avg := agg.gaugeSum / float64(agg.gaugeCount)
		return fmt.Sprintf("min=%.2f avg=%.2f max=%.2f", agg.gaugeMin, avg, agg.gaugeMax)
	case "histogram":
		if agg.histoCount == 0 {
			return "no data"
		}
		avg := agg.histoSum / float64(agg.histoCount)
		histoMin := safeFloat64(agg.histoMin)
		histoMax := safeFloat64(agg.histoMax)
		base := fmt.Sprintf("count=%d min=%.2f avg=%.2f max=%.2f", agg.histoCount, histoMin, avg, histoMax)
		// Append quantiles if present
		if len(agg.histoQuantiles) > 0 {
			pctKeys := make([]int32, 0, len(agg.histoQuantiles))
			for k := range agg.histoQuantiles {
				pctKeys = append(pctKeys, k)
			}
			sort.Slice(pctKeys, func(i, j int) bool { return pctKeys[i] < pctKeys[j] })
			qParts := make([]string, 0, len(pctKeys))
			for _, pct := range pctKeys {
				qParts = append(qParts, fmt.Sprintf("p%d=%.2f", pct, agg.histoQuantiles[pct]))
			}
			base += " " + strings.Join(qParts, " ")
		}
		return base
	default:
		return "unknown"
	}
}

// renderCustomMetrics appends a custom metrics section to buf.
// Only displayed when at least one agent has custom metrics.
func (d *dashboard) renderCustomMetrics(buf *strings.Builder) {
	aggregated := d.aggregateCustomMetrics()
	if len(aggregated) == 0 {
		return
	}

	innerWidth := 88
	buf.WriteString("\033[1;36m") // bold cyan
	buf.WriteString("╔════════════════════════════════════════════════════════════════════════════════════════╗\n")
	title := "Custom Metrics"
	padding := innerWidth - len(title)
	leftPad := padding / 2
	rightPad := padding - leftPad
	buf.WriteString("║")
	buf.WriteString(strings.Repeat(" ", leftPad))
	buf.WriteString(title)
	buf.WriteString(strings.Repeat(" ", rightPad))
	buf.WriteString("║\n")
	buf.WriteString("\033[0m") // reset

	for _, agg := range aggregated {
		valueStr := formatAggregatedValue(agg)

		// Build display name with tags
		displayName := agg.name
		if len(agg.tags) > 0 {
			tagParts := make([]string, 0, len(agg.tags))
			keys := make([]string, 0, len(agg.tags))
			for k := range agg.tags {
				keys = append(keys, k)
			}
			sort.Strings(keys)
			for _, k := range keys {
				tagParts = append(tagParts, k+"="+agg.tags[k])
			}
			displayName += "[" + strings.Join(tagParts, ",") + "]"
		}

		line := fmt.Sprintf("  %s (%s) = %s", displayName, agg.typ, valueStr)
		buf.WriteString("║ ")
		buf.WriteString(line)
		pad := innerWidth - 2 - len(line)
		if pad < 0 {
			pad = 0
		}
		buf.WriteString(strings.Repeat(" ", pad))
		buf.WriteString(" ║\n")
	}

	buf.WriteString("\033[1;36m") // bold cyan
	buf.WriteString("╚════════════════════════════════════════════════════════════════════════════════════════╝\n")
	buf.WriteString("\033[0m") // reset
}

// formatCustomMetricValue formats a single custom metric value for display.
func formatCustomMetricValue(cm *pb.CustomMetric) string {
	if cm.Type == "counter" {
		return fmt.Sprintf("%d", cm.Value.GetCounter())
	}
	if cm.Type == "gauge" {
		return fmt.Sprintf("%.2f", cm.Value.GetGauge())
	}
	if cm.Type == "histogram" {
		h := cm.Value.GetHistogram()
		if h == nil {
			return "no data"
		}
		avg := 0.0
		if h.Count > 0 {
			avg = h.Sum / float64(h.Count)
		}
		histoMin := safeFloat64(h.Min)
		histoMax := safeFloat64(h.Max)
		base := fmt.Sprintf("count=%d min=%.2f avg=%.2f max=%.2f", h.Count, histoMin, avg, histoMax)
		// Append quantiles if present
		if len(h.Quantiles) > 0 {
			pctKeys := make([]int32, 0, len(h.Quantiles))
			for k := range h.Quantiles {
				pctKeys = append(pctKeys, k)
			}
			sort.Slice(pctKeys, func(i, j int) bool { return pctKeys[i] < pctKeys[j] })
			qParts := make([]string, 0, len(pctKeys))
			for _, pct := range pctKeys {
				qParts = append(qParts, fmt.Sprintf("p%d=%.2f", pct, h.Quantiles[pct]))
			}
			base += " " + strings.Join(qParts, " ")
		}
		return base
	}
	return "unknown"
}

// renderFinal clears active users to zero and renders the final state.
func (d *dashboard) renderFinal() {
	for _, m := range d.agents {
		m.activeUsers = 0
	}
	d.render()
}

// printFinal prints a final summary when the test ends.
func (d *dashboard) printFinal(agents []*agentConn) {
	// Read final snapshots
	for i, a := range agents {
		if a == nil {
			continue
		}
		// Drain the channel for final values
		for {
			select {
			case snapshot := <-a.metricsCh:
				d.setAgentSnapshot(i, snapshot)
			default:
				goto nextAgent
			}
		}
	nextAgent:
	}

	d.renderFinal()
	fmt.Println("\n\033[1;32m✓ Test completed\033[0m")
}

func truncateString(s string, maxLen int) string {
	if len(s) <= maxLen {
		return s
	}
	return s[:maxLen-3] + "..."
}

// safeFloat64 replaces Inf/NaN with 0 for safe display in formatting functions.
func safeFloat64(v float64) float64 {
	if math.IsInf(v, 0) || math.IsNaN(v) {
		return 0
	}
	return v
}

// Helper for sorting int32 keys in map iteration.
var _ = strconv.Itoa // keep unused import if needed