diff --git a/internal/alerting/alert_manager.go b/internal/alerting/alert_manager.go new file mode 100644 index 0000000..5605bd5 --- /dev/null +++ b/internal/alerting/alert_manager.go @@ -0,0 +1,217 @@ +package alerting + +import ( + "fmt" + "sync" + "time" +) + +// AlertLevel represents alert severity +type AlertLevel string + +const ( + AlertWarning AlertLevel = "warning" + AlertError AlertLevel = "error" + AlertCritical AlertLevel = "critical" +) + +// Alert represents an alert notification +type Alert struct { + ID string + Level AlertLevel + Title string + Message string + Timestamp time.Time + Resolved bool + Source string +} + +// AlertRule represents a rule that triggers alerts +type AlertRule struct { + ID string + Name string + Threshold float64 + Metric string + Level AlertLevel +} + +// AlertManager manages alert rules and notifications +type AlertManager struct { + mu sync.RWMutex + rules map[string]*AlertRule + alerts map[string]*Alert + history []*Alert +} + +// NewAlertManager creates a new alert manager +func NewAlertManager() *AlertManager { + return &AlertManager{ + rules: make(map[string]*AlertRule), + alerts: make(map[string]*Alert), + history: make([]*Alert, 0), + } +} + +// AddRule adds an alert rule +func (am *AlertManager) AddRule(rule *AlertRule) error { + if rule.ID == "" || rule.Name == "" { + return fmt.Errorf("rule ID and name required") + } + + am.mu.Lock() + defer am.mu.Unlock() + + am.rules[rule.ID] = rule + return nil +} + +// RemoveRule removes an alert rule +func (am *AlertManager) RemoveRule(ruleID string) error { + am.mu.Lock() + defer am.mu.Unlock() + + if _, exists := am.rules[ruleID]; !exists { + return fmt.Errorf("rule not found: %s", ruleID) + } + + delete(am.rules, ruleID) + return nil +} + +// TriggerAlert triggers a new alert +func (am *AlertManager) TriggerAlert(title, message string, level AlertLevel) (*Alert, error) { + if title == "" { + return nil, fmt.Errorf("alert title required") + } + + am.mu.Lock() + defer am.mu.Unlock() + + alert := &Alert{ + ID: fmt.Sprintf("alert-%d", len(am.alerts)), + Level: level, + Title: title, + Message: message, + Timestamp: time.Now(), + Resolved: false, + } + + am.alerts[alert.ID] = alert + am.history = append(am.history, alert) + + return alert, nil +} + +// ResolveAlert marks an alert as resolved +func (am *AlertManager) ResolveAlert(alertID string) error { + am.mu.Lock() + defer am.mu.Unlock() + + alert, exists := am.alerts[alertID] + if !exists { + return fmt.Errorf("alert not found: %s", alertID) + } + + alert.Resolved = true + return nil +} + +// GetActiveAlerts returns all unresolved alerts +func (am *AlertManager) GetActiveAlerts() []*Alert { + am.mu.RLock() + defer am.mu.RUnlock() + + result := make([]*Alert, 0) + for _, alert := range am.alerts { + if !alert.Resolved { + result = append(result, alert) + } + } + + return result +} + +// GetAlertsByLevel returns alerts by severity level +func (am *AlertManager) GetAlertsByLevel(level AlertLevel) []*Alert { + am.mu.RLock() + defer am.mu.RUnlock() + + result := make([]*Alert, 0) + for _, alert := range am.alerts { + if alert.Level == level { + result = append(result, alert) + } + } + + return result +} + +// GetHistory returns alert history +func (am *AlertManager) GetHistory() []*Alert { + am.mu.RLock() + defer am.mu.RUnlock() + + result := make([]*Alert, len(am.history)) + copy(result, am.history) + + return result +} + +// GetRules returns all alert rules +func (am *AlertManager) GetRules() map[string]*AlertRule { + am.mu.RLock() + defer am.mu.RUnlock() + + result := make(map[string]*AlertRule) + for id, rule := range am.rules { + result[id] = rule + } + + return result +} + +// GetAlertCount returns total active alert count +func (am *AlertManager) GetAlertCount() int { + am.mu.RLock() + defer am.mu.RUnlock() + + return len(am.alerts) +} + +// Clear clears all alerts +func (am *AlertManager) Clear() { + am.mu.Lock() + defer am.mu.Unlock() + + am.alerts = make(map[string]*Alert) +} + +// EvaluateRule checks if a metric triggers an alert rule +func (am *AlertManager) EvaluateRule(ruleID string, metricValue float64) (*Alert, error) { + am.mu.Lock() + defer am.mu.Unlock() + + rule, exists := am.rules[ruleID] + if !exists { + return nil, fmt.Errorf("rule not found: %s", ruleID) + } + + if metricValue >= rule.Threshold { + alert := &Alert{ + ID: fmt.Sprintf("alert-%d", len(am.alerts)), + Level: rule.Level, + Title: rule.Name, + Message: fmt.Sprintf("Threshold %.2f exceeded: %.2f", rule.Threshold, metricValue), + Timestamp: time.Now(), + Resolved: false, + Source: ruleID, + } + + am.alerts[alert.ID] = alert + am.history = append(am.history, alert) + + return alert, nil + } + + return nil, nil +} diff --git a/internal/alerting/alert_manager_test.go b/internal/alerting/alert_manager_test.go new file mode 100644 index 0000000..446ef88 --- /dev/null +++ b/internal/alerting/alert_manager_test.go @@ -0,0 +1,157 @@ +package alerting + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestAddRule(t *testing.T) { + am := NewAlertManager() + rule := &AlertRule{ + ID: "rule-1", + Name: "High Error Rate", + Threshold: 0.1, + Metric: "error_rate", + Level: AlertError, + } + + err := am.AddRule(rule) + assert.NoError(t, err) + + rules := am.GetRules() + assert.Equal(t, 1, len(rules)) +} + +func TestTriggerAlert(t *testing.T) { + am := NewAlertManager() + + alert, err := am.TriggerAlert("Database Down", "PostgreSQL unavailable", AlertCritical) + assert.NoError(t, err) + assert.NotNil(t, alert) + assert.Equal(t, AlertCritical, alert.Level) +} + +func TestResolveAlert(t *testing.T) { + am := NewAlertManager() + + alert, _ := am.TriggerAlert("Test Alert", "Test", AlertWarning) + err := am.ResolveAlert(alert.ID) + + assert.NoError(t, err) + assert.True(t, alert.Resolved) +} + +func TestGetActiveAlerts(t *testing.T) { + am := NewAlertManager() + + alert1, _ := am.TriggerAlert("Alert 1", "Test", AlertWarning) + alert2, _ := am.TriggerAlert("Alert 2", "Test", AlertError) + + am.ResolveAlert(alert1.ID) + + active := am.GetActiveAlerts() + assert.Equal(t, 1, len(active)) + assert.Equal(t, alert2.ID, active[0].ID) +} + +func TestGetAlertsByLevel(t *testing.T) { + am := NewAlertManager() + + am.TriggerAlert("Alert 1", "Test", AlertWarning) + am.TriggerAlert("Alert 2", "Test", AlertError) + am.TriggerAlert("Alert 3", "Test", AlertError) + + errors := am.GetAlertsByLevel(AlertError) + assert.Equal(t, 2, len(errors)) +} + +func TestGetHistory(t *testing.T) { + am := NewAlertManager() + + am.TriggerAlert("Alert 1", "Test", AlertWarning) + am.TriggerAlert("Alert 2", "Test", AlertError) + + history := am.GetHistory() + assert.Equal(t, 2, len(history)) +} + +func TestRemoveRule(t *testing.T) { + am := NewAlertManager() + + rule := &AlertRule{ + ID: "rule-1", + Name: "Test Rule", + Level: AlertWarning, + } + + am.AddRule(rule) + err := am.RemoveRule("rule-1") + + assert.NoError(t, err) + rules := am.GetRules() + assert.Equal(t, 0, len(rules)) +} + +func TestClear(t *testing.T) { + am := NewAlertManager() + + am.TriggerAlert("Alert 1", "Test", AlertWarning) + am.TriggerAlert("Alert 2", "Test", AlertError) + + am.Clear() + + assert.Equal(t, 0, am.GetAlertCount()) +} + +func TestEvaluateRule(t *testing.T) { + am := NewAlertManager() + + rule := &AlertRule{ + ID: "rule-1", + Name: "High Error Rate", + Threshold: 0.1, + Level: AlertError, + } + + am.AddRule(rule) + + alert, _ := am.EvaluateRule("rule-1", 0.15) + assert.NotNil(t, alert) +} + +func TestEvaluateRuleBelowThreshold(t *testing.T) { + am := NewAlertManager() + + rule := &AlertRule{ + ID: "rule-1", + Name: "High Error Rate", + Threshold: 0.1, + Level: AlertError, + } + + am.AddRule(rule) + + alert, _ := am.EvaluateRule("rule-1", 0.05) + assert.Nil(t, alert) +} + +func TestGetAlertCount(t *testing.T) { + am := NewAlertManager() + + am.TriggerAlert("Alert 1", "Test", AlertWarning) + am.TriggerAlert("Alert 2", "Test", AlertError) + + assert.Equal(t, 2, am.GetAlertCount()) +} + +func TestAddRuleError(t *testing.T) { + am := NewAlertManager() + + rule := &AlertRule{ + Name: "No ID", + } + + err := am.AddRule(rule) + assert.Error(t, err) +} diff --git a/internal/clusters/cluster_manager.go b/internal/clusters/cluster_manager.go new file mode 100644 index 0000000..d539527 --- /dev/null +++ b/internal/clusters/cluster_manager.go @@ -0,0 +1,223 @@ +package clusters + +import ( + "fmt" + "sync" + "time" +) + +// ClusterInfo represents a Kubernetes cluster +type ClusterInfo struct { + Name string + APIServer string + Healthy bool + LastCheck time.Time + Capacity int // Max concurrent tasks + Usage int // Current task count +} + +// ClusterManager manages multiple K8s clusters +type ClusterManager struct { + mu sync.RWMutex + clusters map[string]*ClusterInfo +} + +// NewClusterManager creates a new cluster manager +func NewClusterManager() *ClusterManager { + return &ClusterManager{ + clusters: make(map[string]*ClusterInfo), + } +} + +// RegisterCluster registers a new cluster +func (cm *ClusterManager) RegisterCluster(name, apiServer string, capacity int) error { + if name == "" || apiServer == "" { + return fmt.Errorf("cluster name and API server required") + } + + cm.mu.Lock() + defer cm.mu.Unlock() + + if _, exists := cm.clusters[name]; exists { + return fmt.Errorf("cluster already registered: %s", name) + } + + cm.clusters[name] = &ClusterInfo{ + Name: name, + APIServer: apiServer, + Healthy: true, + LastCheck: time.Now(), + Capacity: capacity, + Usage: 0, + } + + return nil +} + +// UnregisterCluster removes a cluster +func (cm *ClusterManager) UnregisterCluster(name string) error { + cm.mu.Lock() + defer cm.mu.Unlock() + + if _, exists := cm.clusters[name]; !exists { + return fmt.Errorf("cluster not found: %s", name) + } + + delete(cm.clusters, name) + return nil +} + +// GetCluster retrieves cluster info +func (cm *ClusterManager) GetCluster(name string) (*ClusterInfo, bool) { + cm.mu.RLock() + defer cm.mu.RUnlock() + + cluster, exists := cm.clusters[name] + return cluster, exists +} + +// ListClusters returns all registered clusters +func (cm *ClusterManager) ListClusters() map[string]*ClusterInfo { + cm.mu.RLock() + defer cm.mu.RUnlock() + + result := make(map[string]*ClusterInfo) + for name, cluster := range cm.clusters { + result[name] = cluster + } + + return result +} + +// HealthCheck checks cluster health +func (cm *ClusterManager) HealthCheck(name string) error { + cm.mu.Lock() + defer cm.mu.Unlock() + + cluster, exists := cm.clusters[name] + if !exists { + return fmt.Errorf("cluster not found: %s", name) + } + + // Simulate health check (in production, query API server) + cluster.Healthy = true + cluster.LastCheck = time.Now() + + return nil +} + +// MarkUnhealthy marks a cluster as unhealthy +func (cm *ClusterManager) MarkUnhealthy(name string) error { + cm.mu.Lock() + defer cm.mu.Unlock() + + cluster, exists := cm.clusters[name] + if !exists { + return fmt.Errorf("cluster not found: %s", name) + } + + cluster.Healthy = false + return nil +} + +// AllocateTask allocates a task to a cluster +func (cm *ClusterManager) AllocateTask(name string) error { + cm.mu.Lock() + defer cm.mu.Unlock() + + cluster, exists := cm.clusters[name] + if !exists { + return fmt.Errorf("cluster not found: %s", name) + } + + if !cluster.Healthy { + return fmt.Errorf("cluster not healthy: %s", name) + } + + if cluster.Usage >= cluster.Capacity { + return fmt.Errorf("cluster at capacity: %s", name) + } + + cluster.Usage++ + return nil +} + +// ReleaseTask releases a task from a cluster +func (cm *ClusterManager) ReleaseTask(name string) error { + cm.mu.Lock() + defer cm.mu.Unlock() + + cluster, exists := cm.clusters[name] + if !exists { + return fmt.Errorf("cluster not found: %s", name) + } + + if cluster.Usage > 0 { + cluster.Usage-- + } + + return nil +} + +// FindBestCluster finds the cluster with most available capacity +func (cm *ClusterManager) FindBestCluster() (string, error) { + cm.mu.RLock() + defer cm.mu.RUnlock() + + var bestCluster string + maxCapacity := 0 + + for name, cluster := range cm.clusters { + if cluster.Healthy { + available := cluster.Capacity - cluster.Usage + if available > maxCapacity { + bestCluster = name + maxCapacity = available + } + } + } + + if bestCluster == "" { + return "", fmt.Errorf("no healthy clusters available") + } + + return bestCluster, nil +} + +// GetCapacitySummary returns capacity summary +func (cm *ClusterManager) GetCapacitySummary() map[string]interface{} { + cm.mu.RLock() + defer cm.mu.RUnlock() + + totalCapacity := 0 + totalUsage := 0 + healthyCount := 0 + + for _, cluster := range cm.clusters { + totalCapacity += cluster.Capacity + totalUsage += cluster.Usage + if cluster.Healthy { + healthyCount++ + } + } + + return map[string]interface{}{ + "total_capacity": totalCapacity, + "total_usage": totalUsage, + "healthy_clusters": healthyCount, + "total_clusters": len(cm.clusters), + } +} + +// GetHealthStatus returns health status for all clusters +func (cm *ClusterManager) GetHealthStatus() map[string]bool { + cm.mu.RLock() + defer cm.mu.RUnlock() + + result := make(map[string]bool) + for name, cluster := range cm.clusters { + result[name] = cluster.Healthy + } + + return result +} diff --git a/internal/clusters/cluster_manager_test.go b/internal/clusters/cluster_manager_test.go new file mode 100644 index 0000000..cc158c3 --- /dev/null +++ b/internal/clusters/cluster_manager_test.go @@ -0,0 +1,147 @@ +package clusters + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestRegisterCluster(t *testing.T) { + cm := NewClusterManager() + + err := cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + assert.NoError(t, err) + + cluster, exists := cm.GetCluster("prod") + assert.True(t, exists) + assert.Equal(t, "prod", cluster.Name) +} + +func TestUnregisterCluster(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + + err := cm.UnregisterCluster("prod") + assert.NoError(t, err) + + _, exists := cm.GetCluster("prod") + assert.False(t, exists) +} + +func TestListClusters(t *testing.T) { + cm := NewClusterManager() + + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + cm.RegisterCluster("staging", "https://k8s-staging.com", 50) + + clusters := cm.ListClusters() + assert.Equal(t, 2, len(clusters)) +} + +func TestHealthCheck(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + + err := cm.HealthCheck("prod") + assert.NoError(t, err) + + cluster, _ := cm.GetCluster("prod") + assert.True(t, cluster.Healthy) +} + +func TestMarkUnhealthy(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + + cm.MarkUnhealthy("prod") + + cluster, _ := cm.GetCluster("prod") + assert.False(t, cluster.Healthy) +} + +func TestAllocateTask(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + + err := cm.AllocateTask("prod") + assert.NoError(t, err) + + cluster, _ := cm.GetCluster("prod") + assert.Equal(t, 1, cluster.Usage) +} + +func TestAllocateTaskUnhealthy(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + cm.MarkUnhealthy("prod") + + err := cm.AllocateTask("prod") + assert.Error(t, err) +} + +func TestAllocateTaskAtCapacity(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 1) + + cm.AllocateTask("prod") + err := cm.AllocateTask("prod") + + assert.Error(t, err) +} + +func TestReleaseTask(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + + cm.AllocateTask("prod") + cm.ReleaseTask("prod") + + cluster, _ := cm.GetCluster("prod") + assert.Equal(t, 0, cluster.Usage) +} + +func TestFindBestCluster(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + cm.RegisterCluster("staging", "https://k8s-staging.com", 50) + + cm.AllocateTask("staging") + cm.AllocateTask("staging") + + best, err := cm.FindBestCluster() + assert.NoError(t, err) + assert.Equal(t, "prod", best) +} + +func TestGetCapacitySummary(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + cm.RegisterCluster("staging", "https://k8s-staging.com", 50) + + cm.AllocateTask("prod") + + summary := cm.GetCapacitySummary() + assert.Equal(t, 150, summary["total_capacity"]) + assert.Equal(t, 1, summary["total_usage"]) + assert.Equal(t, 2, summary["healthy_clusters"]) +} + +func TestGetHealthStatus(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + cm.RegisterCluster("staging", "https://k8s-staging.com", 50) + + cm.MarkUnhealthy("staging") + + status := cm.GetHealthStatus() + assert.True(t, status["prod"]) + assert.False(t, status["staging"]) +} + +func TestRegisterClusterError(t *testing.T) { + cm := NewClusterManager() + cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + + err := cm.RegisterCluster("prod", "https://k8s-prod.com", 100) + assert.Error(t, err) +} diff --git a/internal/deployment/self_deployer.go b/internal/deployment/self_deployer.go new file mode 100644 index 0000000..615f78a --- /dev/null +++ b/internal/deployment/self_deployer.go @@ -0,0 +1,235 @@ +package deployment + +import ( + "fmt" + "sync" + "time" +) + +// DeploymentStatus represents deployment status +type DeploymentStatus string + +const ( + StatusPending DeploymentStatus = "pending" + StatusBuilding DeploymentStatus = "building" + StatusPushing DeploymentStatus = "pushing" + StatusApplying DeploymentStatus = "applying" + StatusSuccess DeploymentStatus = "success" + StatusFailed DeploymentStatus = "failed" +) + +// DeploymentInfo represents a deployment attempt +type DeploymentInfo struct { + ID string + Version string + Status DeploymentStatus + StartedAt time.Time + CompletedAt time.Time + Container string + Registry string + Manifest string +} + +// SelfDeployer handles orchestrator self-deployment +type SelfDeployer struct { + mu sync.RWMutex + deployments map[string]*DeploymentInfo + currentVersion string + registry string + kubeConfig string +} + +// NewSelfDeployer creates a new self deployer +func NewSelfDeployer(registry, kubeConfig string) *SelfDeployer { + return &SelfDeployer{ + deployments: make(map[string]*DeploymentInfo), + currentVersion: "1.0.0", + registry: registry, + kubeConfig: kubeConfig, + } +} + +// BuildContainer builds a Docker container image +func (sd *SelfDeployer) BuildContainer(version string) (string, error) { + if version == "" { + return "", fmt.Errorf("version required") + } + + sd.mu.Lock() + defer sd.mu.Unlock() + + deploymentID := fmt.Sprintf("deploy-%s-%d", version, len(sd.deployments)) + + deployment := &DeploymentInfo{ + ID: deploymentID, + Version: version, + Status: StatusBuilding, + StartedAt: time.Now(), + Container: fmt.Sprintf("%s/orchestrator:%s", sd.registry, version), + Registry: sd.registry, + } + + sd.deployments[deploymentID] = deployment + + // Simulate build + deployment.Status = StatusPushing + + return deploymentID, nil +} + +// PushImage pushes the container image to registry +func (sd *SelfDeployer) PushImage(deploymentID string) error { + sd.mu.Lock() + defer sd.mu.Unlock() + + deployment, exists := sd.deployments[deploymentID] + if !exists { + return fmt.Errorf("deployment not found: %s", deploymentID) + } + + if deployment.Status != StatusPushing { + return fmt.Errorf("invalid status for push: %s", deployment.Status) + } + + // Simulate push + deployment.Status = StatusApplying + + return nil +} + +// GenerateManifest generates K8s manifests +func (sd *SelfDeployer) GenerateManifest(deploymentID string, replicas int) (string, error) { + sd.mu.Lock() + defer sd.mu.Unlock() + + deployment, exists := sd.deployments[deploymentID] + if !exists { + return "", fmt.Errorf("deployment not found: %s", deploymentID) + } + + manifest := fmt.Sprintf(` +apiVersion: apps/v1 +kind: Deployment +metadata: + name: poimen-orchestrator +spec: + replicas: %d + selector: + matchLabels: + app: poimen-orchestrator + template: + metadata: + labels: + app: poimen-orchestrator + spec: + containers: + - name: orchestrator + image: %s + ports: + - containerPort: 7233 + - containerPort: 8081 +`, replicas, deployment.Container) + + deployment.Manifest = manifest + + return manifest, nil +} + +// Deploy applies the deployment to K8s +func (sd *SelfDeployer) Deploy(deploymentID string) error { + sd.mu.Lock() + defer sd.mu.Unlock() + + deployment, exists := sd.deployments[deploymentID] + if !exists { + return fmt.Errorf("deployment not found: %s", deploymentID) + } + + if deployment.Status != StatusApplying { + return fmt.Errorf("invalid status for deploy: %s", deployment.Status) + } + + // Simulate deployment + deployment.Status = StatusSuccess + deployment.CompletedAt = time.Now() + + // Update current version + sd.currentVersion = deployment.Version + + return nil +} + +// Rollback rolls back to previous version +func (sd *SelfDeployer) Rollback(previousVersion string) error { + sd.mu.Lock() + defer sd.mu.Unlock() + + // Create a new deployment for rollback + deploymentID := fmt.Sprintf("rollback-%s-%d", previousVersion, len(sd.deployments)) + + deployment := &DeploymentInfo{ + ID: deploymentID, + Version: previousVersion, + Status: StatusSuccess, + StartedAt: time.Now(), + CompletedAt: time.Now(), + Container: fmt.Sprintf("%s/orchestrator:%s", sd.registry, previousVersion), + } + + sd.deployments[deploymentID] = deployment + sd.currentVersion = previousVersion + + return nil +} + +// GetDeploymentInfo retrieves deployment info +func (sd *SelfDeployer) GetDeploymentInfo(deploymentID string) (*DeploymentInfo, bool) { + sd.mu.RLock() + defer sd.mu.RUnlock() + + deployment, exists := sd.deployments[deploymentID] + return deployment, exists +} + +// GetCurrentVersion returns the current orchestrator version +func (sd *SelfDeployer) GetCurrentVersion() string { + sd.mu.RLock() + defer sd.mu.RUnlock() + + return sd.currentVersion +} + +// ListDeployments returns all deployments +func (sd *SelfDeployer) ListDeployments() map[string]*DeploymentInfo { + sd.mu.RLock() + defer sd.mu.RUnlock() + + result := make(map[string]*DeploymentInfo) + for id, deployment := range sd.deployments { + result[id] = deployment + } + + return result +} + +// HealthCheck checks if the deployed orchestrator is healthy +func (sd *SelfDeployer) HealthCheck(deploymentID string) (bool, error) { + sd.mu.RLock() + defer sd.mu.RUnlock() + + deployment, exists := sd.deployments[deploymentID] + if !exists { + return false, fmt.Errorf("deployment not found: %s", deploymentID) + } + + // Simulate health check + return deployment.Status == StatusSuccess, nil +} + +// SetVersion sets the target version +func (sd *SelfDeployer) SetVersion(version string) { + sd.mu.Lock() + defer sd.mu.Unlock() + + sd.currentVersion = version +} diff --git a/internal/deployment/self_deployer_test.go b/internal/deployment/self_deployer_test.go new file mode 100644 index 0000000..8f8be21 --- /dev/null +++ b/internal/deployment/self_deployer_test.go @@ -0,0 +1,136 @@ +package deployment + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNewSelfDeployer(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + assert.NotNil(t, deployer) + assert.Equal(t, "1.0.0", deployer.GetCurrentVersion()) +} + +func TestBuildContainer(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + deploymentID, err := deployer.BuildContainer("2.0.0") + assert.NoError(t, err) + assert.NotEmpty(t, deploymentID) + + deployment, exists := deployer.GetDeploymentInfo(deploymentID) + assert.True(t, exists) + assert.Equal(t, "2.0.0", deployment.Version) +} + +func TestPushImage(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + deploymentID, _ := deployer.BuildContainer("2.0.0") + err := deployer.PushImage(deploymentID) + + assert.NoError(t, err) +} + +func TestGenerateManifest(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + deploymentID, _ := deployer.BuildContainer("2.0.0") + manifest, err := deployer.GenerateManifest(deploymentID, 3) + + assert.NoError(t, err) + assert.NotEmpty(t, manifest) + assert.Contains(t, manifest, "replicas: 3") +} + +func TestDeploy(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + deploymentID, _ := deployer.BuildContainer("2.0.0") + deployer.PushImage(deploymentID) + deployer.GenerateManifest(deploymentID, 3) + err := deployer.Deploy(deploymentID) + + assert.NoError(t, err) + assert.Equal(t, "2.0.0", deployer.GetCurrentVersion()) +} + +func TestRollback(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + deployer.SetVersion("2.0.0") + err := deployer.Rollback("1.0.0") + + assert.NoError(t, err) + assert.Equal(t, "1.0.0", deployer.GetCurrentVersion()) +} + +func TestHealthCheck(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + deploymentID, _ := deployer.BuildContainer("2.0.0") + deployer.PushImage(deploymentID) + deployer.GenerateManifest(deploymentID, 3) + deployer.Deploy(deploymentID) + + healthy, err := deployer.HealthCheck(deploymentID) + assert.NoError(t, err) + assert.True(t, healthy) +} + +func TestListDeployments(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + deployer.BuildContainer("2.0.0") + deployer.BuildContainer("2.0.1") + + deployments := deployer.ListDeployments() + assert.Equal(t, 2, len(deployments)) +} + +func TestBuildContainerError(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + _, err := deployer.BuildContainer("") + assert.Error(t, err) +} + +func TestPushImageError(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + err := deployer.PushImage("nonexistent") + assert.Error(t, err) +} + +func TestFullDeploymentCycle(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + // Build + deploymentID, _ := deployer.BuildContainer("2.0.0") + + // Push + deployer.PushImage(deploymentID) + + // Generate manifest + deployer.GenerateManifest(deploymentID, 3) + + // Deploy + err := deployer.Deploy(deploymentID) + assert.NoError(t, err) + + // Verify + assert.Equal(t, "2.0.0", deployer.GetCurrentVersion()) + + // Health check + healthy, _ := deployer.HealthCheck(deploymentID) + assert.True(t, healthy) +} + +func TestSetVersion(t *testing.T) { + deployer := NewSelfDeployer("docker.io", "/etc/kubernetes/config") + + deployer.SetVersion("3.0.0") + assert.Equal(t, "3.0.0", deployer.GetCurrentVersion()) +} diff --git a/internal/profiling/workflow_profiler.go b/internal/profiling/workflow_profiler.go new file mode 100644 index 0000000..8ef48f4 --- /dev/null +++ b/internal/profiling/workflow_profiler.go @@ -0,0 +1,208 @@ +package profiling + +import ( + "fmt" + "sort" + "sync" +) + +// TaskProfile represents profiling data for a task +type TaskProfile struct { + TaskID string + Duration float64 + CPUUsage float64 + MemUsage float64 + Throughput float64 +} + +// WorkflowProfile represents profiling for an entire workflow +type WorkflowProfile struct { + WorkflowID string + Tasks map[string]*TaskProfile + TotalDuration float64 + CriticalPath []string +} + +// WorkflowProfiler profiles workflow execution +type WorkflowProfiler struct { + mu sync.RWMutex + profiles map[string]*WorkflowProfile +} + +// NewWorkflowProfiler creates a new workflow profiler +func NewWorkflowProfiler() *WorkflowProfiler { + return &WorkflowProfiler{ + profiles: make(map[string]*WorkflowProfile), + } +} + +// RecordTaskExecution records task execution metrics +func (wp *WorkflowProfiler) RecordTaskExecution(workflowID, taskID string, duration, cpu, mem float64) { + wp.mu.Lock() + defer wp.mu.Unlock() + + if _, exists := wp.profiles[workflowID]; !exists { + wp.profiles[workflowID] = &WorkflowProfile{ + WorkflowID: workflowID, + Tasks: make(map[string]*TaskProfile), + CriticalPath: make([]string, 0), + } + } + + profile := wp.profiles[workflowID] + profile.Tasks[taskID] = &TaskProfile{ + TaskID: taskID, + Duration: duration, + CPUUsage: cpu, + MemUsage: mem, + Throughput: 1000.0 / duration, // Tasks per second + } + + // Recalculate total duration + total := 0.0 + for _, tp := range profile.Tasks { + if tp.Duration > total { + total = tp.Duration + } + } + profile.TotalDuration = total +} + +// GetSlowTasks returns tasks sorted by duration (slowest first) +func (wp *WorkflowProfiler) GetSlowTasks(workflowID string, limit int) []string { + wp.mu.RLock() + defer wp.mu.RUnlock() + + profile, exists := wp.profiles[workflowID] + if !exists { + return []string{} + } + + // Sort tasks by duration + type taskDuration struct { + taskID string + duration float64 + } + + tasks := make([]taskDuration, 0) + for taskID, tp := range profile.Tasks { + tasks = append(tasks, taskDuration{taskID, tp.Duration}) + } + + sort.Slice(tasks, func(i, j int) bool { + return tasks[i].duration > tasks[j].duration + }) + + result := make([]string, 0) + for i, t := range tasks { + if i >= limit { + break + } + result = append(result, t.taskID) + } + + return result +} + +// GetHighCPUTasks returns tasks with high CPU usage +func (wp *WorkflowProfiler) GetHighCPUTasks(workflowID string, threshold float64) []string { + wp.mu.RLock() + defer wp.mu.RUnlock() + + profile, exists := wp.profiles[workflowID] + if !exists { + return []string{} + } + + result := make([]string, 0) + for taskID, tp := range profile.Tasks { + if tp.CPUUsage > threshold { + result = append(result, taskID) + } + } + + return result +} + +// GetHighMemTasks returns tasks with high memory usage +func (wp *WorkflowProfiler) GetHighMemTasks(workflowID string, threshold float64) []string { + wp.mu.RLock() + defer wp.mu.RUnlock() + + profile, exists := wp.profiles[workflowID] + if !exists { + return []string{} + } + + result := make([]string, 0) + for taskID, tp := range profile.Tasks { + if tp.MemUsage > threshold { + result = append(result, taskID) + } + } + + return result +} + +// GetOptimizationSuggestions returns optimization recommendations +func (wp *WorkflowProfiler) GetOptimizationSuggestions(workflowID string) []string { + wp.mu.RLock() + defer wp.mu.RUnlock() + + profile, exists := wp.profiles[workflowID] + if !exists { + return []string{} + } + + suggestions := make([]string, 0) + + // Check for slow tasks + for taskID, tp := range profile.Tasks { + if tp.Duration > profile.TotalDuration*0.5 { + suggestions = append(suggestions, fmt.Sprintf("Task %s takes 50%% of total time, consider optimizing", taskID)) + } + if tp.CPUUsage > 0.8 { + suggestions = append(suggestions, fmt.Sprintf("Task %s has high CPU usage (%.2f), consider parallelizing", taskID, tp.CPUUsage)) + } + if tp.MemUsage > 0.8 { + suggestions = append(suggestions, fmt.Sprintf("Task %s has high memory usage (%.2f), consider reducing payload", taskID, tp.MemUsage)) + } + } + + return suggestions +} + +// GetProfile retrieves profiling data for a workflow +func (wp *WorkflowProfiler) GetProfile(workflowID string) (*WorkflowProfile, bool) { + wp.mu.RLock() + defer wp.mu.RUnlock() + + profile, exists := wp.profiles[workflowID] + return profile, exists +} + +// GetTaskProfile retrieves profiling data for a specific task +func (wp *WorkflowProfiler) GetTaskProfile(workflowID, taskID string) (*TaskProfile, error) { + wp.mu.RLock() + defer wp.mu.RUnlock() + + profile, exists := wp.profiles[workflowID] + if !exists { + return nil, fmt.Errorf("workflow not found: %s", workflowID) + } + + taskProfile, exists := profile.Tasks[taskID] + if !exists { + return nil, fmt.Errorf("task not found: %s", taskID) + } + + return taskProfile, nil +} + +// Clear clears all profiles +func (wp *WorkflowProfiler) Clear() { + wp.mu.Lock() + defer wp.mu.Unlock() + + wp.profiles = make(map[string]*WorkflowProfile) +} diff --git a/internal/profiling/workflow_profiler_test.go b/internal/profiling/workflow_profiler_test.go new file mode 100644 index 0000000..7dd2a87 --- /dev/null +++ b/internal/profiling/workflow_profiler_test.go @@ -0,0 +1,125 @@ +package profiling + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestRecordTaskExecution(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + + profile, exists := profiler.GetProfile("wf-1") + assert.True(t, exists) + assert.Equal(t, 1, len(profile.Tasks)) +} + +func TestGetSlowTasks(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + profiler.RecordTaskExecution("wf-1", "task-2", 500, 0.8, 0.6) + profiler.RecordTaskExecution("wf-1", "task-3", 200, 0.4, 0.2) + + slow := profiler.GetSlowTasks("wf-1", 2) + assert.Equal(t, 2, len(slow)) + assert.Equal(t, "task-2", slow[0]) +} + +func TestGetHighCPUTasks(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + profiler.RecordTaskExecution("wf-1", "task-2", 200, 0.9, 0.6) + + highCPU := profiler.GetHighCPUTasks("wf-1", 0.7) + assert.Equal(t, 1, len(highCPU)) + assert.Equal(t, "task-2", highCPU[0]) +} + +func TestGetHighMemTasks(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + profiler.RecordTaskExecution("wf-1", "task-2", 200, 0.6, 0.9) + + highMem := profiler.GetHighMemTasks("wf-1", 0.7) + assert.Equal(t, 1, len(highMem)) + assert.Equal(t, "task-2", highMem[0]) +} + +func TestGetOptimizationSuggestions(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + profiler.RecordTaskExecution("wf-1", "task-2", 200, 0.9, 0.9) + + suggestions := profiler.GetOptimizationSuggestions("wf-1") + assert.Greater(t, len(suggestions), 0) +} + +func TestGetProfile(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + + profile, exists := profiler.GetProfile("wf-1") + assert.True(t, exists) + assert.Equal(t, "wf-1", profile.WorkflowID) +} + +func TestGetTaskProfile(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + + taskProfile, err := profiler.GetTaskProfile("wf-1", "task-1") + assert.NoError(t, err) + assert.Equal(t, 100.0, taskProfile.Duration) +} + +func TestTaskNotFound(t *testing.T) { + profiler := NewWorkflowProfiler() + + _, err := profiler.GetTaskProfile("wf-1", "task-999") + assert.Error(t, err) +} + +func TestClear(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + profiler.Clear() + + profile, exists := profiler.GetProfile("wf-1") + assert.False(t, exists) + assert.Nil(t, profile) +} + +func TestThroughputCalculation(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 1000, 0.5, 0.3) + + profile, _ := profiler.GetProfile("wf-1") + taskProfile := profile.Tasks["task-1"] + + assert.Equal(t, 1.0, taskProfile.Throughput) // 1000ms = 1 task per second +} + +func TestMultipleWorkflows(t *testing.T) { + profiler := NewWorkflowProfiler() + + profiler.RecordTaskExecution("wf-1", "task-1", 100, 0.5, 0.3) + profiler.RecordTaskExecution("wf-2", "task-1", 200, 0.6, 0.4) + + profile1, exists1 := profiler.GetProfile("wf-1") + profile2, exists2 := profiler.GetProfile("wf-2") + + assert.True(t, exists1) + assert.True(t, exists2) + assert.Equal(t, 100.0, profile1.TotalDuration) + assert.Equal(t, 200.0, profile2.TotalDuration) +}