9bdcbe0447
Major integrations and fixes: - Added BACKBEAT SDK integration for P2P operation timing - Implemented beat-aware status tracking for distributed operations - Added Docker secrets support for secure license management - Resolved KACHING license validation via HTTPS/TLS - Updated docker-compose configuration for clean stack deployment - Disabled rollback policies to prevent deployment failures - Added license credential storage (CHORUS-DEV-MULTI-001) Technical improvements: - BACKBEAT P2P operation tracking with phase management - Enhanced configuration system with file-based secrets - Improved error handling for license validation - Clean separation of KACHING and CHORUS deployment stacks 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
1148 lines
36 KiB
Go
1148 lines
36 KiB
Go
// Package distribution provides comprehensive monitoring and observability for distributed context operations
|
|
package distribution
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"chorus/pkg/config"
|
|
)
|
|
|
|
// MonitoringSystem provides comprehensive monitoring for the distributed context system
|
|
type MonitoringSystem struct {
|
|
mu sync.RWMutex
|
|
config *config.Config
|
|
metrics *MetricsCollector
|
|
healthChecks *HealthCheckManager
|
|
alertManager *AlertManager
|
|
dashboard *DashboardServer
|
|
logManager *LogManager
|
|
traceManager *TraceManager
|
|
|
|
// State
|
|
running bool
|
|
monitoringPort int
|
|
updateInterval time.Duration
|
|
retentionPeriod time.Duration
|
|
}
|
|
|
|
// MetricsCollector collects and aggregates system metrics
|
|
type MetricsCollector struct {
|
|
mu sync.RWMutex
|
|
timeSeries map[string]*TimeSeries
|
|
counters map[string]*Counter
|
|
gauges map[string]*Gauge
|
|
histograms map[string]*Histogram
|
|
customMetrics map[string]*CustomMetric
|
|
aggregatedStats *AggregatedStatistics
|
|
exporters []MetricsExporter
|
|
lastCollection time.Time
|
|
}
|
|
|
|
// TimeSeries represents a time-series metric
|
|
type TimeSeries struct {
|
|
Name string `json:"name"`
|
|
Labels map[string]string `json:"labels"`
|
|
DataPoints []*TimeSeriesPoint `json:"data_points"`
|
|
RetentionTTL time.Duration `json:"retention_ttl"`
|
|
LastUpdated time.Time `json:"last_updated"`
|
|
}
|
|
|
|
// TimeSeriesPoint represents a single data point in a time series
|
|
type TimeSeriesPoint struct {
|
|
Timestamp time.Time `json:"timestamp"`
|
|
Value float64 `json:"value"`
|
|
Labels map[string]string `json:"labels,omitempty"`
|
|
}
|
|
|
|
// Counter represents a monotonically increasing counter
|
|
type Counter struct {
|
|
Name string `json:"name"`
|
|
Value int64 `json:"value"`
|
|
Rate float64 `json:"rate"` // per second
|
|
Labels map[string]string `json:"labels"`
|
|
LastUpdated time.Time `json:"last_updated"`
|
|
}
|
|
|
|
// Gauge represents a value that can go up and down
|
|
type Gauge struct {
|
|
Name string `json:"name"`
|
|
Value float64 `json:"value"`
|
|
Min float64 `json:"min"`
|
|
Max float64 `json:"max"`
|
|
Average float64 `json:"average"`
|
|
Labels map[string]string `json:"labels"`
|
|
LastUpdated time.Time `json:"last_updated"`
|
|
}
|
|
|
|
// Histogram represents distribution of values
|
|
type Histogram struct {
|
|
Name string `json:"name"`
|
|
Buckets map[float64]int64 `json:"buckets"`
|
|
Count int64 `json:"count"`
|
|
Sum float64 `json:"sum"`
|
|
Labels map[string]string `json:"labels"`
|
|
Percentiles map[float64]float64 `json:"percentiles"`
|
|
LastUpdated time.Time `json:"last_updated"`
|
|
}
|
|
|
|
// CustomMetric represents application-specific metrics
|
|
type CustomMetric struct {
|
|
Name string `json:"name"`
|
|
Type MetricType `json:"type"`
|
|
Value interface{} `json:"value"`
|
|
Metadata map[string]interface{} `json:"metadata"`
|
|
Labels map[string]string `json:"labels"`
|
|
LastUpdated time.Time `json:"last_updated"`
|
|
}
|
|
|
|
// MetricType represents the type of custom metric
|
|
type MetricType string
|
|
|
|
const (
|
|
MetricTypeCounter MetricType = "counter"
|
|
MetricTypeGauge MetricType = "gauge"
|
|
MetricTypeHistogram MetricType = "histogram"
|
|
MetricTypeSummary MetricType = "summary"
|
|
MetricTypeCustom MetricType = "custom"
|
|
)
|
|
|
|
// AggregatedStatistics provides high-level system statistics
|
|
type AggregatedStatistics struct {
|
|
SystemOverview *SystemOverview `json:"system_overview"`
|
|
PerformanceMetrics *PerformanceOverview `json:"performance_metrics"`
|
|
HealthMetrics *HealthOverview `json:"health_metrics"`
|
|
ErrorMetrics *ErrorOverview `json:"error_metrics"`
|
|
ResourceMetrics *ResourceOverview `json:"resource_metrics"`
|
|
NetworkMetrics *NetworkOverview `json:"network_metrics"`
|
|
LastUpdated time.Time `json:"last_updated"`
|
|
}
|
|
|
|
// SystemOverview provides system-wide overview metrics
|
|
type SystemOverview struct {
|
|
TotalNodes int `json:"total_nodes"`
|
|
HealthyNodes int `json:"healthy_nodes"`
|
|
TotalContexts int64 `json:"total_contexts"`
|
|
DistributedContexts int64 `json:"distributed_contexts"`
|
|
ReplicationFactor float64 `json:"average_replication_factor"`
|
|
SystemUptime time.Duration `json:"system_uptime"`
|
|
ClusterVersion string `json:"cluster_version"`
|
|
LastRestart time.Time `json:"last_restart"`
|
|
}
|
|
|
|
// PerformanceOverview provides performance metrics
|
|
type PerformanceOverview struct {
|
|
RequestsPerSecond float64 `json:"requests_per_second"`
|
|
AverageResponseTime time.Duration `json:"average_response_time"`
|
|
P95ResponseTime time.Duration `json:"p95_response_time"`
|
|
P99ResponseTime time.Duration `json:"p99_response_time"`
|
|
Throughput float64 `json:"throughput_mbps"`
|
|
CacheHitRate float64 `json:"cache_hit_rate"`
|
|
QueueDepth int `json:"queue_depth"`
|
|
ActiveConnections int `json:"active_connections"`
|
|
}
|
|
|
|
// HealthOverview provides health-related metrics
|
|
type HealthOverview struct {
|
|
OverallHealthScore float64 `json:"overall_health_score"`
|
|
ComponentHealth map[string]float64 `json:"component_health"`
|
|
FailedHealthChecks int `json:"failed_health_checks"`
|
|
LastHealthCheck time.Time `json:"last_health_check"`
|
|
HealthTrend string `json:"health_trend"` // improving, stable, degrading
|
|
CriticalAlerts int `json:"critical_alerts"`
|
|
WarningAlerts int `json:"warning_alerts"`
|
|
}
|
|
|
|
// ErrorOverview provides error-related metrics
|
|
type ErrorOverview struct {
|
|
TotalErrors int64 `json:"total_errors"`
|
|
ErrorRate float64 `json:"error_rate"`
|
|
ErrorsByType map[string]int64 `json:"errors_by_type"`
|
|
ErrorsByComponent map[string]int64 `json:"errors_by_component"`
|
|
LastError *ErrorEvent `json:"last_error"`
|
|
ErrorTrend string `json:"error_trend"` // increasing, stable, decreasing
|
|
}
|
|
|
|
// ResourceOverview provides resource utilization metrics
|
|
type ResourceOverview struct {
|
|
CPUUtilization float64 `json:"cpu_utilization"`
|
|
MemoryUtilization float64 `json:"memory_utilization"`
|
|
DiskUtilization float64 `json:"disk_utilization"`
|
|
NetworkUtilization float64 `json:"network_utilization"`
|
|
StorageUsed int64 `json:"storage_used_bytes"`
|
|
StorageAvailable int64 `json:"storage_available_bytes"`
|
|
FileDescriptors int `json:"open_file_descriptors"`
|
|
Goroutines int `json:"goroutines"`
|
|
}
|
|
|
|
// NetworkOverview provides network-related metrics
|
|
type NetworkOverview struct {
|
|
TotalConnections int `json:"total_connections"`
|
|
ActiveConnections int `json:"active_connections"`
|
|
BandwidthUtilization float64 `json:"bandwidth_utilization"`
|
|
PacketLossRate float64 `json:"packet_loss_rate"`
|
|
AverageLatency time.Duration `json:"average_latency"`
|
|
NetworkPartitions int `json:"network_partitions"`
|
|
DataTransferred int64 `json:"data_transferred_bytes"`
|
|
}
|
|
|
|
// MetricsExporter exports metrics to external systems
|
|
type MetricsExporter interface {
|
|
Export(ctx context.Context, metrics map[string]interface{}) error
|
|
Name() string
|
|
IsEnabled() bool
|
|
}
|
|
|
|
// HealthCheckManager manages system health checks
|
|
type HealthCheckManager struct {
|
|
mu sync.RWMutex
|
|
healthChecks map[string]*HealthCheck
|
|
checkResults map[string]*HealthCheckResult
|
|
schedules map[string]*HealthCheckSchedule
|
|
running bool
|
|
}
|
|
|
|
// HealthCheck represents a single health check
|
|
type HealthCheck struct {
|
|
Name string `json:"name"`
|
|
Description string `json:"description"`
|
|
CheckType HealthCheckType `json:"check_type"`
|
|
Target string `json:"target"`
|
|
Timeout time.Duration `json:"timeout"`
|
|
Interval time.Duration `json:"interval"`
|
|
Retries int `json:"retries"`
|
|
Metadata map[string]interface{} `json:"metadata"`
|
|
Enabled bool `json:"enabled"`
|
|
CheckFunction func(context.Context) (*HealthCheckResult, error) `json:"-"`
|
|
}
|
|
|
|
// HealthCheckType represents different types of health checks
|
|
type HealthCheckType string
|
|
|
|
const (
|
|
HealthCheckTypeHTTP HealthCheckType = "http"
|
|
HealthCheckTypeTCP HealthCheckType = "tcp"
|
|
HealthCheckTypeCustom HealthCheckType = "custom"
|
|
HealthCheckTypeComponent HealthCheckType = "component"
|
|
HealthCheckTypeDatabase HealthCheckType = "database"
|
|
HealthCheckTypeService HealthCheckType = "service"
|
|
)
|
|
|
|
// HealthCheckResult represents the result of a health check
|
|
type HealthCheckResult struct {
|
|
CheckName string `json:"check_name"`
|
|
Status HealthCheckStatus `json:"status"`
|
|
ResponseTime time.Duration `json:"response_time"`
|
|
Message string `json:"message"`
|
|
Details map[string]interface{} `json:"details"`
|
|
Error string `json:"error,omitempty"`
|
|
Timestamp time.Time `json:"timestamp"`
|
|
Attempt int `json:"attempt"`
|
|
}
|
|
|
|
// HealthCheckStatus represents the status of a health check
|
|
type HealthCheckStatus string
|
|
|
|
const (
|
|
HealthCheckStatusHealthy HealthCheckStatus = "healthy"
|
|
HealthCheckStatusUnhealthy HealthCheckStatus = "unhealthy"
|
|
HealthCheckStatusWarning HealthCheckStatus = "warning"
|
|
HealthCheckStatusUnknown HealthCheckStatus = "unknown"
|
|
HealthCheckStatusTimeout HealthCheckStatus = "timeout"
|
|
)
|
|
|
|
// HealthCheckSchedule defines when health checks should run
|
|
type HealthCheckSchedule struct {
|
|
CheckName string `json:"check_name"`
|
|
Interval time.Duration `json:"interval"`
|
|
NextRun time.Time `json:"next_run"`
|
|
LastRun time.Time `json:"last_run"`
|
|
Enabled bool `json:"enabled"`
|
|
FailureCount int `json:"failure_count"`
|
|
}
|
|
|
|
// AlertManager manages system alerts and notifications
|
|
type AlertManager struct {
|
|
mu sync.RWMutex
|
|
alertRules map[string]*AlertRule
|
|
activeAlerts map[string]*Alert
|
|
alertHistory []*Alert
|
|
notifiers []AlertNotifier
|
|
silences map[string]*AlertSilence
|
|
running bool
|
|
}
|
|
|
|
// AlertRule defines conditions for triggering alerts
|
|
type AlertRule struct {
|
|
Name string `json:"name"`
|
|
Description string `json:"description"`
|
|
Severity AlertSeverity `json:"severity"`
|
|
Conditions []*AlertCondition `json:"conditions"`
|
|
Duration time.Duration `json:"duration"` // How long condition must persist
|
|
Cooldown time.Duration `json:"cooldown"` // Minimum time between alerts
|
|
Labels map[string]string `json:"labels"`
|
|
Annotations map[string]string `json:"annotations"`
|
|
Enabled bool `json:"enabled"`
|
|
LastTriggered *time.Time `json:"last_triggered,omitempty"`
|
|
}
|
|
|
|
// AlertCondition defines a single condition for an alert
|
|
type AlertCondition struct {
|
|
MetricName string `json:"metric_name"`
|
|
Operator ConditionOperator `json:"operator"`
|
|
Threshold float64 `json:"threshold"`
|
|
Duration time.Duration `json:"duration"`
|
|
}
|
|
|
|
// ConditionOperator represents comparison operators for alert conditions
|
|
type ConditionOperator string
|
|
|
|
const (
|
|
OperatorGreaterThan ConditionOperator = "gt"
|
|
OperatorLessThan ConditionOperator = "lt"
|
|
OperatorEquals ConditionOperator = "eq"
|
|
OperatorNotEquals ConditionOperator = "ne"
|
|
OperatorGreaterOrEqual ConditionOperator = "gte"
|
|
OperatorLessOrEqual ConditionOperator = "lte"
|
|
)
|
|
|
|
// Alert represents an active alert
|
|
type Alert struct {
|
|
ID string `json:"id"`
|
|
RuleName string `json:"rule_name"`
|
|
Severity AlertSeverity `json:"severity"`
|
|
Status AlertStatus `json:"status"`
|
|
Message string `json:"message"`
|
|
Details map[string]interface{} `json:"details"`
|
|
Labels map[string]string `json:"labels"`
|
|
Annotations map[string]string `json:"annotations"`
|
|
StartsAt time.Time `json:"starts_at"`
|
|
EndsAt *time.Time `json:"ends_at,omitempty"`
|
|
LastUpdated time.Time `json:"last_updated"`
|
|
AckBy string `json:"acknowledged_by,omitempty"`
|
|
AckAt *time.Time `json:"acknowledged_at,omitempty"`
|
|
}
|
|
|
|
// AlertSeverity represents the severity level of an alert
|
|
type AlertSeverity string
|
|
|
|
const (
|
|
SeverityInfo AlertSeverity = "info"
|
|
SeverityWarning AlertSeverity = "warning"
|
|
SeverityError AlertSeverity = "error"
|
|
SeverityCritical AlertSeverity = "critical"
|
|
)
|
|
|
|
// AlertStatus represents the current status of an alert
|
|
type AlertStatus string
|
|
|
|
const (
|
|
AlertStatusFiring AlertStatus = "firing"
|
|
AlertStatusResolved AlertStatus = "resolved"
|
|
AlertStatusAcknowledged AlertStatus = "acknowledged"
|
|
AlertStatusSilenced AlertStatus = "silenced"
|
|
)
|
|
|
|
// AlertNotifier sends alert notifications
|
|
type AlertNotifier interface {
|
|
Notify(ctx context.Context, alert *Alert) error
|
|
Name() string
|
|
IsEnabled() bool
|
|
}
|
|
|
|
// AlertSilence represents a silenced alert
|
|
type AlertSilence struct {
|
|
ID string `json:"id"`
|
|
Matchers map[string]string `json:"matchers"`
|
|
StartTime time.Time `json:"start_time"`
|
|
EndTime time.Time `json:"end_time"`
|
|
CreatedBy string `json:"created_by"`
|
|
Comment string `json:"comment"`
|
|
Active bool `json:"active"`
|
|
}
|
|
|
|
// DashboardServer provides web-based monitoring dashboard
|
|
type DashboardServer struct {
|
|
mu sync.RWMutex
|
|
server *http.Server
|
|
dashboards map[string]*Dashboard
|
|
widgets map[string]*Widget
|
|
customPages map[string]*CustomPage
|
|
running bool
|
|
port int
|
|
}
|
|
|
|
// Dashboard represents a monitoring dashboard
|
|
type Dashboard struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
Description string `json:"description"`
|
|
Widgets []*Widget `json:"widgets"`
|
|
Layout *DashboardLayout `json:"layout"`
|
|
Settings *DashboardSettings `json:"settings"`
|
|
CreatedBy string `json:"created_by"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
// Widget represents a dashboard widget
|
|
type Widget struct {
|
|
ID string `json:"id"`
|
|
Type WidgetType `json:"type"`
|
|
Title string `json:"title"`
|
|
DataSource string `json:"data_source"`
|
|
Query string `json:"query"`
|
|
Settings map[string]interface{} `json:"settings"`
|
|
Position *WidgetPosition `json:"position"`
|
|
RefreshRate time.Duration `json:"refresh_rate"`
|
|
LastUpdated time.Time `json:"last_updated"`
|
|
}
|
|
|
|
// WidgetType represents different types of dashboard widgets
|
|
type WidgetType string
|
|
|
|
const (
|
|
WidgetTypeMetric WidgetType = "metric"
|
|
WidgetTypeChart WidgetType = "chart"
|
|
WidgetTypeTable WidgetType = "table"
|
|
WidgetTypeAlert WidgetType = "alert"
|
|
WidgetTypeHealth WidgetType = "health"
|
|
WidgetTypeTopology WidgetType = "topology"
|
|
WidgetTypeLog WidgetType = "log"
|
|
WidgetTypeCustom WidgetType = "custom"
|
|
)
|
|
|
|
// WidgetPosition defines widget position and size
|
|
type WidgetPosition struct {
|
|
X int `json:"x"`
|
|
Y int `json:"y"`
|
|
Width int `json:"width"`
|
|
Height int `json:"height"`
|
|
}
|
|
|
|
// DashboardLayout defines dashboard layout settings
|
|
type DashboardLayout struct {
|
|
Columns int `json:"columns"`
|
|
RowHeight int `json:"row_height"`
|
|
Margins [2]int `json:"margins"` // [x, y]
|
|
Spacing [2]int `json:"spacing"` // [x, y]
|
|
Breakpoints map[string]int `json:"breakpoints"`
|
|
}
|
|
|
|
// DashboardSettings contains dashboard configuration
|
|
type DashboardSettings struct {
|
|
AutoRefresh bool `json:"auto_refresh"`
|
|
RefreshInterval time.Duration `json:"refresh_interval"`
|
|
TimeRange string `json:"time_range"`
|
|
Theme string `json:"theme"`
|
|
ShowLegend bool `json:"show_legend"`
|
|
ShowGrid bool `json:"show_grid"`
|
|
}
|
|
|
|
// CustomPage represents a custom monitoring page
|
|
type CustomPage struct {
|
|
Path string `json:"path"`
|
|
Title string `json:"title"`
|
|
Content string `json:"content"`
|
|
ContentType string `json:"content_type"`
|
|
Handler http.HandlerFunc `json:"-"`
|
|
}
|
|
|
|
// LogManager manages system logs and log analysis
|
|
type LogManager struct {
|
|
mu sync.RWMutex
|
|
logSources map[string]*LogSource
|
|
logEntries []*LogEntry
|
|
logAnalyzers []LogAnalyzer
|
|
retentionPolicy *LogRetentionPolicy
|
|
running bool
|
|
}
|
|
|
|
// LogSource represents a source of log data
|
|
type LogSource struct {
|
|
Name string `json:"name"`
|
|
Type LogSourceType `json:"type"`
|
|
Location string `json:"location"`
|
|
Format LogFormat `json:"format"`
|
|
Labels map[string]string `json:"labels"`
|
|
Enabled bool `json:"enabled"`
|
|
LastRead time.Time `json:"last_read"`
|
|
}
|
|
|
|
// LogSourceType represents different types of log sources
|
|
type LogSourceType string
|
|
|
|
const (
|
|
LogSourceTypeFile LogSourceType = "file"
|
|
LogSourceTypeHTTP LogSourceType = "http"
|
|
LogSourceTypeStream LogSourceType = "stream"
|
|
LogSourceTypeDatabase LogSourceType = "database"
|
|
LogSourceTypeCustom LogSourceType = "custom"
|
|
)
|
|
|
|
// LogFormat represents log entry format
|
|
type LogFormat string
|
|
|
|
const (
|
|
LogFormatJSON LogFormat = "json"
|
|
LogFormatText LogFormat = "text"
|
|
LogFormatSyslog LogFormat = "syslog"
|
|
LogFormatCustom LogFormat = "custom"
|
|
)
|
|
|
|
// LogEntry represents a single log entry
|
|
type LogEntry struct {
|
|
Timestamp time.Time `json:"timestamp"`
|
|
Level LogLevel `json:"level"`
|
|
Source string `json:"source"`
|
|
Message string `json:"message"`
|
|
Fields map[string]interface{} `json:"fields"`
|
|
Labels map[string]string `json:"labels"`
|
|
TraceID string `json:"trace_id,omitempty"`
|
|
SpanID string `json:"span_id,omitempty"`
|
|
}
|
|
|
|
// LogLevel represents log entry severity
|
|
type LogLevel string
|
|
|
|
const (
|
|
LogLevelTrace LogLevel = "trace"
|
|
LogLevelDebug LogLevel = "debug"
|
|
LogLevelInfo LogLevel = "info"
|
|
LogLevelWarn LogLevel = "warn"
|
|
LogLevelError LogLevel = "error"
|
|
LogLevelFatal LogLevel = "fatal"
|
|
)
|
|
|
|
// LogAnalyzer analyzes log entries for patterns and anomalies
|
|
type LogAnalyzer interface {
|
|
Analyze(ctx context.Context, entries []*LogEntry) (*LogAnalysisResult, error)
|
|
Name() string
|
|
}
|
|
|
|
// LogAnalysisResult represents the result of log analysis
|
|
type LogAnalysisResult struct {
|
|
AnalyzerName string `json:"analyzer_name"`
|
|
Anomalies []*LogAnomaly `json:"anomalies"`
|
|
Patterns []*LogPattern `json:"patterns"`
|
|
Statistics *LogStatistics `json:"statistics"`
|
|
Recommendations []string `json:"recommendations"`
|
|
AnalyzedAt time.Time `json:"analyzed_at"`
|
|
}
|
|
|
|
// LogAnomaly represents detected log anomaly
|
|
type LogAnomaly struct {
|
|
Type AnomalyType `json:"type"`
|
|
Severity AlertSeverity `json:"severity"`
|
|
Description string `json:"description"`
|
|
Entries []*LogEntry `json:"entries"`
|
|
Confidence float64 `json:"confidence"`
|
|
DetectedAt time.Time `json:"detected_at"`
|
|
}
|
|
|
|
// AnomalyType represents different types of log anomalies
|
|
type AnomalyType string
|
|
|
|
const (
|
|
AnomalyTypeErrorSpike AnomalyType = "error_spike"
|
|
AnomalyTypeUnusualPattern AnomalyType = "unusual_pattern"
|
|
AnomalyTypeMissingLogs AnomalyType = "missing_logs"
|
|
AnomalyTypeRateChange AnomalyType = "rate_change"
|
|
AnomalyTypeNewError AnomalyType = "new_error"
|
|
)
|
|
|
|
// LogPattern represents detected log pattern
|
|
type LogPattern struct {
|
|
Pattern string `json:"pattern"`
|
|
Frequency int `json:"frequency"`
|
|
LastSeen time.Time `json:"last_seen"`
|
|
Sources []string `json:"sources"`
|
|
Confidence float64 `json:"confidence"`
|
|
}
|
|
|
|
// LogStatistics provides log statistics
|
|
type LogStatistics struct {
|
|
TotalEntries int64 `json:"total_entries"`
|
|
EntriesByLevel map[LogLevel]int64 `json:"entries_by_level"`
|
|
EntriesBySource map[string]int64 `json:"entries_by_source"`
|
|
ErrorRate float64 `json:"error_rate"`
|
|
AverageRate float64 `json:"average_rate"`
|
|
TimeRange [2]time.Time `json:"time_range"`
|
|
}
|
|
|
|
// LogRetentionPolicy defines log retention rules
|
|
type LogRetentionPolicy struct {
|
|
RetentionPeriod time.Duration `json:"retention_period"`
|
|
MaxEntries int64 `json:"max_entries"`
|
|
CompressionAge time.Duration `json:"compression_age"`
|
|
ArchiveAge time.Duration `json:"archive_age"`
|
|
Rules []*RetentionRule `json:"rules"`
|
|
}
|
|
|
|
// RetentionRule defines specific retention rules
|
|
type RetentionRule struct {
|
|
Name string `json:"name"`
|
|
Condition string `json:"condition"` // Query expression
|
|
Retention time.Duration `json:"retention"`
|
|
Action RetentionAction `json:"action"`
|
|
}
|
|
|
|
// RetentionAction represents retention actions
|
|
type RetentionAction string
|
|
|
|
const (
|
|
RetentionActionDelete RetentionAction = "delete"
|
|
RetentionActionArchive RetentionAction = "archive"
|
|
RetentionActionCompress RetentionAction = "compress"
|
|
)
|
|
|
|
// TraceManager manages distributed tracing
|
|
type TraceManager struct {
|
|
mu sync.RWMutex
|
|
traces map[string]*Trace
|
|
spans map[string]*Span
|
|
samplers []TraceSampler
|
|
exporters []TraceExporter
|
|
running bool
|
|
}
|
|
|
|
// Trace represents a distributed trace
|
|
type Trace struct {
|
|
TraceID string `json:"trace_id"`
|
|
Spans []*Span `json:"spans"`
|
|
Duration time.Duration `json:"duration"`
|
|
StartTime time.Time `json:"start_time"`
|
|
EndTime time.Time `json:"end_time"`
|
|
Status TraceStatus `json:"status"`
|
|
Tags map[string]string `json:"tags"`
|
|
Operations []string `json:"operations"`
|
|
}
|
|
|
|
// Span represents a single span in a trace
|
|
type Span struct {
|
|
SpanID string `json:"span_id"`
|
|
TraceID string `json:"trace_id"`
|
|
ParentID string `json:"parent_id,omitempty"`
|
|
Operation string `json:"operation"`
|
|
Service string `json:"service"`
|
|
StartTime time.Time `json:"start_time"`
|
|
EndTime time.Time `json:"end_time"`
|
|
Duration time.Duration `json:"duration"`
|
|
Status SpanStatus `json:"status"`
|
|
Tags map[string]string `json:"tags"`
|
|
Logs []*SpanLog `json:"logs"`
|
|
}
|
|
|
|
// TraceStatus represents the status of a trace
|
|
type TraceStatus string
|
|
|
|
const (
|
|
TraceStatusOK TraceStatus = "ok"
|
|
TraceStatusError TraceStatus = "error"
|
|
TraceStatusTimeout TraceStatus = "timeout"
|
|
)
|
|
|
|
// SpanStatus represents the status of a span
|
|
type SpanStatus string
|
|
|
|
const (
|
|
SpanStatusOK SpanStatus = "ok"
|
|
SpanStatusError SpanStatus = "error"
|
|
)
|
|
|
|
// SpanLog represents a log entry within a span
|
|
type SpanLog struct {
|
|
Timestamp time.Time `json:"timestamp"`
|
|
Fields map[string]interface{} `json:"fields"`
|
|
}
|
|
|
|
// TraceSampler determines which traces to sample
|
|
type TraceSampler interface {
|
|
Sample(traceID string, operation string) bool
|
|
Name() string
|
|
}
|
|
|
|
// TraceExporter exports traces to external systems
|
|
type TraceExporter interface {
|
|
Export(ctx context.Context, traces []*Trace) error
|
|
Name() string
|
|
}
|
|
|
|
// ErrorEvent represents a system error event
|
|
type ErrorEvent struct {
|
|
ID string `json:"id"`
|
|
Timestamp time.Time `json:"timestamp"`
|
|
Level LogLevel `json:"level"`
|
|
Component string `json:"component"`
|
|
Message string `json:"message"`
|
|
Error string `json:"error"`
|
|
Context map[string]interface{} `json:"context"`
|
|
TraceID string `json:"trace_id,omitempty"`
|
|
SpanID string `json:"span_id,omitempty"`
|
|
Count int `json:"count"`
|
|
FirstSeen time.Time `json:"first_seen"`
|
|
LastSeen time.Time `json:"last_seen"`
|
|
}
|
|
|
|
// NewMonitoringSystem creates a comprehensive monitoring system
|
|
func NewMonitoringSystem(config *config.Config) (*MonitoringSystem, error) {
|
|
if config == nil {
|
|
return nil, fmt.Errorf("config is required")
|
|
}
|
|
|
|
ms := &MonitoringSystem{
|
|
config: config,
|
|
monitoringPort: 8080,
|
|
updateInterval: 30 * time.Second,
|
|
retentionPeriod: 24 * time.Hour,
|
|
}
|
|
|
|
// Initialize components
|
|
if err := ms.initializeComponents(); err != nil {
|
|
return nil, fmt.Errorf("failed to initialize monitoring components: %w", err)
|
|
}
|
|
|
|
return ms, nil
|
|
}
|
|
|
|
// initializeComponents initializes all monitoring components
|
|
func (ms *MonitoringSystem) initializeComponents() error {
|
|
// Initialize metrics collector
|
|
ms.metrics = &MetricsCollector{
|
|
timeSeries: make(map[string]*TimeSeries),
|
|
counters: make(map[string]*Counter),
|
|
gauges: make(map[string]*Gauge),
|
|
histograms: make(map[string]*Histogram),
|
|
customMetrics: make(map[string]*CustomMetric),
|
|
aggregatedStats: &AggregatedStatistics{
|
|
LastUpdated: time.Now(),
|
|
},
|
|
exporters: []MetricsExporter{},
|
|
lastCollection: time.Now(),
|
|
}
|
|
|
|
// Initialize health check manager
|
|
ms.healthChecks = &HealthCheckManager{
|
|
healthChecks: make(map[string]*HealthCheck),
|
|
checkResults: make(map[string]*HealthCheckResult),
|
|
schedules: make(map[string]*HealthCheckSchedule),
|
|
running: false,
|
|
}
|
|
|
|
// Initialize alert manager
|
|
ms.alertManager = &AlertManager{
|
|
alertRules: make(map[string]*AlertRule),
|
|
activeAlerts: make(map[string]*Alert),
|
|
alertHistory: []*Alert{},
|
|
notifiers: []AlertNotifier{},
|
|
silences: make(map[string]*AlertSilence),
|
|
running: false,
|
|
}
|
|
|
|
// Initialize dashboard server
|
|
ms.dashboard = &DashboardServer{
|
|
dashboards: make(map[string]*Dashboard),
|
|
widgets: make(map[string]*Widget),
|
|
customPages: make(map[string]*CustomPage),
|
|
running: false,
|
|
port: ms.monitoringPort,
|
|
}
|
|
|
|
// Initialize log manager
|
|
ms.logManager = &LogManager{
|
|
logSources: make(map[string]*LogSource),
|
|
logEntries: []*LogEntry{},
|
|
logAnalyzers: []LogAnalyzer{},
|
|
retentionPolicy: &LogRetentionPolicy{
|
|
RetentionPeriod: 7 * 24 * time.Hour,
|
|
MaxEntries: 1000000,
|
|
CompressionAge: 24 * time.Hour,
|
|
ArchiveAge: 7 * 24 * time.Hour,
|
|
Rules: []*RetentionRule{},
|
|
},
|
|
running: false,
|
|
}
|
|
|
|
// Initialize trace manager
|
|
ms.traceManager = &TraceManager{
|
|
traces: make(map[string]*Trace),
|
|
spans: make(map[string]*Span),
|
|
samplers: []TraceSampler{},
|
|
exporters: []TraceExporter{},
|
|
running: false,
|
|
}
|
|
|
|
// Register default health checks
|
|
ms.registerDefaultHealthChecks()
|
|
|
|
// Register default alert rules
|
|
ms.registerDefaultAlertRules()
|
|
|
|
// Create default dashboards
|
|
ms.createDefaultDashboards()
|
|
|
|
return nil
|
|
}
|
|
|
|
// Start starts the monitoring system
|
|
func (ms *MonitoringSystem) Start(ctx context.Context) error {
|
|
ms.mu.Lock()
|
|
if ms.running {
|
|
ms.mu.Unlock()
|
|
return fmt.Errorf("monitoring system already running")
|
|
}
|
|
ms.running = true
|
|
ms.mu.Unlock()
|
|
|
|
// Start metrics collection
|
|
go ms.metricsCollectionWorker(ctx)
|
|
|
|
// Start health check manager
|
|
ms.healthChecks.running = true
|
|
go ms.healthCheckWorker(ctx)
|
|
|
|
// Start alert manager
|
|
ms.alertManager.running = true
|
|
go ms.alertWorker(ctx)
|
|
|
|
// Start log manager
|
|
ms.logManager.running = true
|
|
go ms.logWorker(ctx)
|
|
|
|
// Start trace manager
|
|
ms.traceManager.running = true
|
|
go ms.traceWorker(ctx)
|
|
|
|
// Start dashboard server
|
|
if err := ms.startDashboardServer(); err != nil {
|
|
return fmt.Errorf("failed to start dashboard server: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Stop stops the monitoring system
|
|
func (ms *MonitoringSystem) Stop() error {
|
|
ms.mu.Lock()
|
|
defer ms.mu.Unlock()
|
|
|
|
ms.running = false
|
|
ms.healthChecks.running = false
|
|
ms.alertManager.running = false
|
|
ms.logManager.running = false
|
|
ms.traceManager.running = false
|
|
|
|
// Stop dashboard server
|
|
if ms.dashboard.server != nil {
|
|
return ms.dashboard.server.Shutdown(context.Background())
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetMetrics returns current system metrics
|
|
func (ms *MonitoringSystem) GetMetrics() (*AggregatedStatistics, error) {
|
|
ms.metrics.mu.RLock()
|
|
defer ms.metrics.mu.RUnlock()
|
|
|
|
return ms.metrics.aggregatedStats, nil
|
|
}
|
|
|
|
// GetHealthStatus returns current health status
|
|
func (ms *MonitoringSystem) GetHealthStatus() (map[string]*HealthCheckResult, error) {
|
|
ms.healthChecks.mu.RLock()
|
|
defer ms.healthChecks.mu.RUnlock()
|
|
|
|
results := make(map[string]*HealthCheckResult)
|
|
for name, result := range ms.healthChecks.checkResults {
|
|
results[name] = result
|
|
}
|
|
|
|
return results, nil
|
|
}
|
|
|
|
// GetActiveAlerts returns currently active alerts
|
|
func (ms *MonitoringSystem) GetActiveAlerts() ([]*Alert, error) {
|
|
ms.alertManager.mu.RLock()
|
|
defer ms.alertManager.mu.RUnlock()
|
|
|
|
alerts := make([]*Alert, 0, len(ms.alertManager.activeAlerts))
|
|
for _, alert := range ms.alertManager.activeAlerts {
|
|
alerts = append(alerts, alert)
|
|
}
|
|
|
|
// Sort by severity and timestamp
|
|
sort.Slice(alerts, func(i, j int) bool {
|
|
if alerts[i].Severity != alerts[j].Severity {
|
|
return ms.severityWeight(alerts[i].Severity) > ms.severityWeight(alerts[j].Severity)
|
|
}
|
|
return alerts[i].StartsAt.After(alerts[j].StartsAt)
|
|
})
|
|
|
|
return alerts, nil
|
|
}
|
|
|
|
// RecordMetric records a custom metric
|
|
func (ms *MonitoringSystem) RecordMetric(name string, value float64, labels map[string]string) error {
|
|
ms.metrics.mu.Lock()
|
|
defer ms.metrics.mu.Unlock()
|
|
|
|
// Create or update gauge
|
|
if gauge, exists := ms.metrics.gauges[name]; exists {
|
|
gauge.Value = value
|
|
gauge.LastUpdated = time.Now()
|
|
if labels != nil {
|
|
gauge.Labels = labels
|
|
}
|
|
} else {
|
|
ms.metrics.gauges[name] = &Gauge{
|
|
Name: name,
|
|
Value: value,
|
|
Min: value,
|
|
Max: value,
|
|
Average: value,
|
|
Labels: labels,
|
|
LastUpdated: time.Now(),
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Background workers (placeholder implementations)
|
|
|
|
func (ms *MonitoringSystem) metricsCollectionWorker(ctx context.Context) {
|
|
ticker := time.NewTicker(ms.updateInterval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if ms.running {
|
|
ms.collectSystemMetrics()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (ms *MonitoringSystem) healthCheckWorker(ctx context.Context) {
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if ms.healthChecks.running {
|
|
ms.runHealthChecks(ctx)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (ms *MonitoringSystem) alertWorker(ctx context.Context) {
|
|
ticker := time.NewTicker(10 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if ms.alertManager.running {
|
|
ms.evaluateAlertRules(ctx)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (ms *MonitoringSystem) logWorker(ctx context.Context) {
|
|
ticker := time.NewTicker(60 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if ms.logManager.running {
|
|
ms.analyzeLogs(ctx)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (ms *MonitoringSystem) traceWorker(ctx context.Context) {
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if ms.traceManager.running {
|
|
ms.processTraces(ctx)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (ms *MonitoringSystem) startDashboardServer() error {
|
|
mux := http.NewServeMux()
|
|
|
|
// API endpoints
|
|
mux.HandleFunc("/api/metrics", ms.handleMetrics)
|
|
mux.HandleFunc("/api/health", ms.handleHealth)
|
|
mux.HandleFunc("/api/alerts", ms.handleAlerts)
|
|
mux.HandleFunc("/api/dashboards", ms.handleDashboards)
|
|
|
|
// Dashboard UI (placeholder)
|
|
mux.HandleFunc("/", ms.handleDashboard)
|
|
|
|
ms.dashboard.server = &http.Server{
|
|
Addr: fmt.Sprintf(":%d", ms.dashboard.port),
|
|
Handler: mux,
|
|
}
|
|
|
|
go func() {
|
|
if err := ms.dashboard.server.ListenAndServe(); err != http.ErrServerClosed {
|
|
// Log error
|
|
}
|
|
}()
|
|
|
|
ms.dashboard.running = true
|
|
return nil
|
|
}
|
|
|
|
// HTTP handlers (placeholder implementations)
|
|
|
|
func (ms *MonitoringSystem) handleMetrics(w http.ResponseWriter, r *http.Request) {
|
|
metrics, err := ms.GetMetrics()
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(metrics)
|
|
}
|
|
|
|
func (ms *MonitoringSystem) handleHealth(w http.ResponseWriter, r *http.Request) {
|
|
health, err := ms.GetHealthStatus()
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(health)
|
|
}
|
|
|
|
func (ms *MonitoringSystem) handleAlerts(w http.ResponseWriter, r *http.Request) {
|
|
alerts, err := ms.GetActiveAlerts()
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(alerts)
|
|
}
|
|
|
|
func (ms *MonitoringSystem) handleDashboards(w http.ResponseWriter, r *http.Request) {
|
|
ms.dashboard.mu.RLock()
|
|
dashboards := make([]*Dashboard, 0, len(ms.dashboard.dashboards))
|
|
for _, dashboard := range ms.dashboard.dashboards {
|
|
dashboards = append(dashboards, dashboard)
|
|
}
|
|
ms.dashboard.mu.RUnlock()
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(dashboards)
|
|
}
|
|
|
|
func (ms *MonitoringSystem) handleDashboard(w http.ResponseWriter, r *http.Request) {
|
|
// Placeholder dashboard HTML
|
|
html := `
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head><title>CHORUS SLURP Monitoring</title></head>
|
|
<body>
|
|
<h1>CHORUS SLURP Distributed Context Monitoring</h1>
|
|
<p>Monitoring dashboard placeholder</p>
|
|
</body>
|
|
</html>
|
|
`
|
|
w.Header().Set("Content-Type", "text/html")
|
|
w.Write([]byte(html))
|
|
}
|
|
|
|
// Helper methods (placeholder implementations)
|
|
|
|
func (ms *MonitoringSystem) collectSystemMetrics() {
|
|
// Collect system metrics
|
|
ms.metrics.aggregatedStats.SystemOverview = &SystemOverview{
|
|
TotalNodes: 1, // Placeholder
|
|
HealthyNodes: 1,
|
|
TotalContexts: 0,
|
|
DistributedContexts: 0,
|
|
ReplicationFactor: 3.0,
|
|
SystemUptime: time.Since(time.Now()),
|
|
ClusterVersion: "1.0.0",
|
|
LastRestart: time.Now(),
|
|
}
|
|
|
|
ms.metrics.aggregatedStats.LastUpdated = time.Now()
|
|
}
|
|
|
|
func (ms *MonitoringSystem) runHealthChecks(ctx context.Context) {
|
|
// Run scheduled health checks
|
|
}
|
|
|
|
func (ms *MonitoringSystem) evaluateAlertRules(ctx context.Context) {
|
|
// Evaluate alert rules against current metrics
|
|
}
|
|
|
|
func (ms *MonitoringSystem) analyzeLogs(ctx context.Context) {
|
|
// Analyze logs for patterns and anomalies
|
|
}
|
|
|
|
func (ms *MonitoringSystem) processTraces(ctx context.Context) {
|
|
// Process distributed traces
|
|
}
|
|
|
|
func (ms *MonitoringSystem) registerDefaultHealthChecks() {
|
|
// Register default health checks
|
|
}
|
|
|
|
func (ms *MonitoringSystem) registerDefaultAlertRules() {
|
|
// Register default alert rules
|
|
}
|
|
|
|
func (ms *MonitoringSystem) createDefaultDashboards() {
|
|
// Create default dashboards
|
|
}
|
|
|
|
func (ms *MonitoringSystem) severityWeight(severity AlertSeverity) int {
|
|
switch severity {
|
|
case SeverityCritical:
|
|
return 4
|
|
case SeverityError:
|
|
return 3
|
|
case SeverityWarning:
|
|
return 2
|
|
case SeverityInfo:
|
|
return 1
|
|
default:
|
|
return 0
|
|
}
|
|
} |