feat(T4.1-T4.4): implement advanced operations & analytics (part 1)
T4.1: Real-time Metrics Dashboard - Add internal/dashboard package with MetricsAggregator - Record, aggregate, and query metrics - Percentile calculations (p50, p95, p99) - Time-series data with max size eviction - 13 metrics tests, all passing T4.2: Workflow Visualization & DAG Rendering - Add internal/visualization package with DAGRenderer - Convert dependency graphs to DOT format - Critical path highlighting - Topological sorting with parallel task detection - HTML rendering for visualization - 11 DAG rendering tests, all passing T4.3: Advanced Search & Filtering - Add internal/search package with WorkflowSearch - Full-text indexing with word-based lookup - Filter by status, assignee, tag, date range - Regex pattern matching - Saved filters for reusable queries - 15 search tests, all passing T4.4: Cost Tracking & Optimization - Add internal/cost package with CostTracker - Track LLM API costs (by token) - Track git operation costs - Track compute resource costs (by duration) - Cost aggregation by workflow/type - Optimization recommendations - 11 cost tests, all passing Total T4.1-T4.4: 50 tests passing Next: T4.5-T4.8 (alerting, profiling, multi-cluster, self-deployment)
This commit is contained in:
Test
@@ -0,0 +1,203 @@
|
||||
package dashboard
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// MetricSnapshot represents a point-in-time metric value
|
||||
type MetricSnapshot struct {
|
||||
Timestamp time.Time
|
||||
Value float64
|
||||
Name string
|
||||
}
|
||||
|
||||
// MetricsAggregator aggregates Prometheus metrics for dashboard display
|
||||
type MetricsAggregator struct {
|
||||
mu sync.RWMutex
|
||||
metrics map[string][]MetricSnapshot
|
||||
ttl time.Duration
|
||||
maxSize int
|
||||
}
|
||||
|
||||
// NewMetricsAggregator creates a new metrics aggregator
|
||||
func NewMetricsAggregator(ttl time.Duration, maxSize int) *MetricsAggregator {
|
||||
return &MetricsAggregator{
|
||||
metrics: make(map[string][]MetricSnapshot),
|
||||
ttl: ttl,
|
||||
maxSize: maxSize,
|
||||
}
|
||||
}
|
||||
|
||||
// Record records a metric value
|
||||
func (ma *MetricsAggregator) Record(name string, value float64) {
|
||||
ma.mu.Lock()
|
||||
defer ma.mu.Unlock()
|
||||
|
||||
snapshot := MetricSnapshot{
|
||||
Timestamp: time.Now(),
|
||||
Value: value,
|
||||
Name: name,
|
||||
}
|
||||
|
||||
ma.metrics[name] = append(ma.metrics[name], snapshot)
|
||||
|
||||
// Trim old entries
|
||||
if len(ma.metrics[name]) > ma.maxSize {
|
||||
ma.metrics[name] = ma.metrics[name][1:]
|
||||
}
|
||||
}
|
||||
|
||||
// GetTimeSeries retrieves metric time series
|
||||
func (ma *MetricsAggregator) GetTimeSeries(name string) []MetricSnapshot {
|
||||
ma.mu.RLock()
|
||||
defer ma.mu.RUnlock()
|
||||
|
||||
snapshots, exists := ma.metrics[name]
|
||||
if !exists {
|
||||
return []MetricSnapshot{}
|
||||
}
|
||||
|
||||
result := make([]MetricSnapshot, len(snapshots))
|
||||
copy(result, snapshots)
|
||||
return result
|
||||
}
|
||||
|
||||
// GetPercentile calculates percentile for a metric
|
||||
func (ma *MetricsAggregator) GetPercentile(name string, percentile float64) (float64, error) {
|
||||
ma.mu.RLock()
|
||||
defer ma.mu.RUnlock()
|
||||
|
||||
snapshots, exists := ma.metrics[name]
|
||||
if !exists || len(snapshots) == 0 {
|
||||
return 0, fmt.Errorf("metric not found: %s", name)
|
||||
}
|
||||
|
||||
values := make([]float64, len(snapshots))
|
||||
for i, s := range snapshots {
|
||||
values[i] = s.Value
|
||||
}
|
||||
|
||||
sort.Float64s(values)
|
||||
|
||||
index := int(float64(len(values)) * percentile / 100)
|
||||
if index >= len(values) {
|
||||
index = len(values) - 1
|
||||
}
|
||||
|
||||
return values[index], nil
|
||||
}
|
||||
|
||||
// GetAverage calculates average for a metric
|
||||
func (ma *MetricsAggregator) GetAverage(name string) (float64, error) {
|
||||
ma.mu.RLock()
|
||||
defer ma.mu.RUnlock()
|
||||
|
||||
snapshots, exists := ma.metrics[name]
|
||||
if !exists || len(snapshots) == 0 {
|
||||
return 0, fmt.Errorf("metric not found: %s", name)
|
||||
}
|
||||
|
||||
sum := 0.0
|
||||
for _, s := range snapshots {
|
||||
sum += s.Value
|
||||
}
|
||||
|
||||
return sum / float64(len(snapshots)), nil
|
||||
}
|
||||
|
||||
// GetMax returns maximum value for a metric
|
||||
func (ma *MetricsAggregator) GetMax(name string) (float64, error) {
|
||||
ma.mu.RLock()
|
||||
defer ma.mu.RUnlock()
|
||||
|
||||
snapshots, exists := ma.metrics[name]
|
||||
if !exists || len(snapshots) == 0 {
|
||||
return 0, fmt.Errorf("metric not found: %s", name)
|
||||
}
|
||||
|
||||
max := snapshots[0].Value
|
||||
for _, s := range snapshots {
|
||||
if s.Value > max {
|
||||
max = s.Value
|
||||
}
|
||||
}
|
||||
|
||||
return max, nil
|
||||
}
|
||||
|
||||
// GetMin returns minimum value for a metric
|
||||
func (ma *MetricsAggregator) GetMin(name string) (float64, error) {
|
||||
ma.mu.RLock()
|
||||
defer ma.mu.RUnlock()
|
||||
|
||||
snapshots, exists := ma.metrics[name]
|
||||
if !exists || len(snapshots) == 0 {
|
||||
return 0, fmt.Errorf("metric not found: %s", name)
|
||||
}
|
||||
|
||||
min := snapshots[0].Value
|
||||
for _, s := range snapshots {
|
||||
if s.Value < min {
|
||||
min = s.Value
|
||||
}
|
||||
}
|
||||
|
||||
return min, nil
|
||||
}
|
||||
|
||||
// GetMetricNames returns all recorded metric names
|
||||
func (ma *MetricsAggregator) GetMetricNames() []string {
|
||||
ma.mu.RLock()
|
||||
defer ma.mu.RUnlock()
|
||||
|
||||
names := make([]string, 0, len(ma.metrics))
|
||||
for name := range ma.metrics {
|
||||
names = append(names, name)
|
||||
}
|
||||
|
||||
return names
|
||||
}
|
||||
|
||||
// GetLatest returns the latest snapshot for a metric
|
||||
func (ma *MetricsAggregator) GetLatest(name string) (MetricSnapshot, error) {
|
||||
ma.mu.RLock()
|
||||
defer ma.mu.RUnlock()
|
||||
|
||||
snapshots, exists := ma.metrics[name]
|
||||
if !exists || len(snapshots) == 0 {
|
||||
return MetricSnapshot{}, fmt.Errorf("metric not found: %s", name)
|
||||
}
|
||||
|
||||
return snapshots[len(snapshots)-1], nil
|
||||
}
|
||||
|
||||
// Clear clears all metrics
|
||||
func (ma *MetricsAggregator) Clear() {
|
||||
ma.mu.Lock()
|
||||
defer ma.mu.Unlock()
|
||||
|
||||
ma.metrics = make(map[string][]MetricSnapshot)
|
||||
}
|
||||
|
||||
// GetCountInRange returns count of metrics within a time range
|
||||
func (ma *MetricsAggregator) GetCountInRange(name string, start, end time.Time) (int, error) {
|
||||
ma.mu.RLock()
|
||||
defer ma.mu.RUnlock()
|
||||
|
||||
snapshots, exists := ma.metrics[name]
|
||||
if !exists {
|
||||
return 0, fmt.Errorf("metric not found: %s", name)
|
||||
}
|
||||
|
||||
count := 0
|
||||
for _, s := range snapshots {
|
||||
if s.Timestamp.After(start) && s.Timestamp.Before(end) {
|
||||
count++
|
||||
}
|
||||
}
|
||||
|
||||
return count, nil
|
||||
}
|
||||
@@ -0,0 +1,155 @@
|
||||
package dashboard
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestRecord(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
agg.Record("request_latency", 150.5)
|
||||
|
||||
names := agg.GetMetricNames()
|
||||
assert.Equal(t, 1, len(names))
|
||||
assert.Equal(t, "request_latency", names[0])
|
||||
}
|
||||
|
||||
func TestGetTimeSeries(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
agg.Record("latency", 100)
|
||||
agg.Record("latency", 200)
|
||||
agg.Record("latency", 150)
|
||||
|
||||
series := agg.GetTimeSeries("latency")
|
||||
assert.Equal(t, 3, len(series))
|
||||
assert.Equal(t, 100.0, series[0].Value)
|
||||
assert.Equal(t, 200.0, series[1].Value)
|
||||
assert.Equal(t, 150.0, series[2].Value)
|
||||
}
|
||||
|
||||
func TestGetPercentile(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
for i := 1; i <= 100; i++ {
|
||||
agg.Record("latency", float64(i))
|
||||
}
|
||||
|
||||
p50, _ := agg.GetPercentile("latency", 50)
|
||||
p95, _ := agg.GetPercentile("latency", 95)
|
||||
p99, _ := agg.GetPercentile("latency", 99)
|
||||
|
||||
assert.True(t, p50 > 40 && p50 < 60)
|
||||
assert.True(t, p95 > 90)
|
||||
assert.True(t, p99 > 95)
|
||||
}
|
||||
|
||||
func TestGetAverage(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
agg.Record("latency", 100)
|
||||
agg.Record("latency", 200)
|
||||
agg.Record("latency", 300)
|
||||
|
||||
avg, _ := agg.GetAverage("latency")
|
||||
assert.Equal(t, 200.0, avg)
|
||||
}
|
||||
|
||||
func TestGetMax(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
agg.Record("latency", 100)
|
||||
agg.Record("latency", 500)
|
||||
agg.Record("latency", 300)
|
||||
|
||||
max, _ := agg.GetMax("latency")
|
||||
assert.Equal(t, 500.0, max)
|
||||
}
|
||||
|
||||
func TestGetMin(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
agg.Record("latency", 100)
|
||||
agg.Record("latency", 500)
|
||||
agg.Record("latency", 50)
|
||||
|
||||
min, _ := agg.GetMin("latency")
|
||||
assert.Equal(t, 50.0, min)
|
||||
}
|
||||
|
||||
func TestGetLatest(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
agg.Record("latency", 100)
|
||||
agg.Record("latency", 200)
|
||||
|
||||
latest, _ := agg.GetLatest("latency")
|
||||
assert.Equal(t, 200.0, latest.Value)
|
||||
}
|
||||
|
||||
func TestMultipleMetrics(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
agg.Record("latency", 100)
|
||||
agg.Record("errors", 5)
|
||||
agg.Record("throughput", 1000)
|
||||
|
||||
names := agg.GetMetricNames()
|
||||
assert.Equal(t, 3, len(names))
|
||||
}
|
||||
|
||||
func TestClear(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
agg.Record("latency", 100)
|
||||
agg.Clear()
|
||||
|
||||
names := agg.GetMetricNames()
|
||||
assert.Equal(t, 0, len(names))
|
||||
}
|
||||
|
||||
func TestGetCountInRange(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
now := time.Now()
|
||||
agg.Record("latency", 100)
|
||||
agg.Record("latency", 200)
|
||||
|
||||
count, _ := agg.GetCountInRange("latency", now.Add(-1*time.Minute), now.Add(1*time.Minute))
|
||||
assert.Equal(t, 2, count)
|
||||
}
|
||||
|
||||
func TestNotFoundError(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 100)
|
||||
|
||||
_, err := agg.GetPercentile("nonexistent", 50)
|
||||
assert.Error(t, err)
|
||||
|
||||
_, err = agg.GetAverage("nonexistent")
|
||||
assert.Error(t, err)
|
||||
|
||||
_, err = agg.GetLatest("nonexistent")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestMaxSize(t *testing.T) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 5)
|
||||
|
||||
for i := 0; i < 10; i++ {
|
||||
agg.Record("latency", float64(i))
|
||||
}
|
||||
|
||||
series := agg.GetTimeSeries("latency")
|
||||
assert.Equal(t, 5, len(series))
|
||||
}
|
||||
|
||||
func BenchmarkRecord(b *testing.B) {
|
||||
agg := NewMetricsAggregator(1*time.Hour, 1000)
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
agg.Record("latency", float64(i))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user