diff --git a/compute-agent/cmd/agent/main.go b/compute-agent/cmd/agent/main.go index 6d7b9c8..53a4947 100644 --- a/compute-agent/cmd/agent/main.go +++ b/compute-agent/cmd/agent/main.go @@ -121,7 +121,7 @@ func main() { // Register Compose runtime if enabled if cfg.ComposeEnabled { - composeRuntime, err := runtime.NewComposeRuntime(cfg.ComposeBinary, "", logger) + composeRuntime, err := runtime.NewComposeRuntime(cfg.ComposeBinary, cfg.DockerEndpoint ,"", logger) if err != nil { logger.Warnf("Failed to initialize Compose runtime: %v", err) } else { diff --git a/compute-agent/examples/client/specs/compose.yaml b/compute-agent/examples/client/specs/compose.yaml index 03df720..da41d4e 100644 --- a/compute-agent/examples/client/specs/compose.yaml +++ b/compute-agent/examples/client/specs/compose.yaml @@ -1,4 +1,3 @@ -version: "3.8" services: web: image: nginx:1.27 diff --git a/compute-agent/internal/config/config.go b/compute-agent/internal/config/config.go index 52943c3..6ba3655 100644 --- a/compute-agent/internal/config/config.go +++ b/compute-agent/internal/config/config.go @@ -92,7 +92,26 @@ var ( fs = pflag.NewFlagSet("compute-agent", pflag.ContinueOnError) ) -// Load loads configuration with Viper + pflag +// Load loads configuration. +// +// Precedence (highest to lowest), per key: +// +// 1. PERSYS_NODE_LABELS parsing (handled explicitly, see below) +// 2. environment variables (PERSYS_*) +// 3. config file (agent_config.yaml), if one is found +// 4. built-in defaults (defaultConfig()) +// +// This is Viper's own override > flag > env > config > default precedence. +// The important bit that makes it actually work is registerDefaults: Viper's +// AutomaticEnv only kicks in, during Unmarshal, for keys it already knows +// about (from a config file, an explicit BindEnv, or a SetDefault). A key +// with no default and no config-file entry is invisible to Unmarshal even if +// the matching PERSYS_* env var is set. Registering every field's default +// up front is what lets "no config file -> use env, else use default" and +// "config file present -> only fill in what env didn't set" both fall out of +// Viper's normal per-key resolution, instead of us re-implementing it by +// hand (which is what the old cfg = defaultConfig() wholesale-replace, and +// the applyMinimalDefaults zero-value merge, both did - and did buggily). func Load() (*Config, error) { v := viper.New() @@ -102,68 +121,76 @@ func Load() (*Config, error) { v.AutomaticEnv() v.SetEnvKeyReplacer(strings.NewReplacer(".", "_", "-", "_")) - // Explicitly bind important fields for tests - v.BindEnv("grpc_port") - v.BindEnv("state_store_path", "PERSYS_STATE_PATH", "PERSYS_STATE_STORE_PATH") - v.BindEnv("node_region") - v.BindEnv("node_env") - v.BindEnv("node_labels") - v.BindEnv("scheduler_addr") - v.BindEnv("scheduler_insecure") - v.BindEnv("docker_enabled") - v.BindEnv("compose_enabled") - v.BindEnv("vm_enabled") - v.BindEnv("tls_enabled") - v.BindEnv("vault_enabled") - v.BindEnv("vault_approle_role_id") - v.BindEnv("vault_approle_secret_id") - v.BindEnv("vault_addr") - v.BindEnv("vault_service_name") - - // Handle PERSYS_NODE_LABELS specially + // Register every field's default so Viper knows about the key and will + // resolve it as env > config file > default, instead of silently + // ignoring the env var because Unmarshal never saw the key. + registerDefaults(v, defaultConfig()) + + // state_store_path historically also accepted PERSYS_STATE_PATH. + if err := v.BindEnv("state_store_path", "PERSYS_STATE_PATH", "PERSYS_STATE_STORE_PATH"); err != nil { + return nil, fmt.Errorf("bind state_store_path env: %w", err) + } + + // PERSYS_NODE_LABELS is a comma-separated key=value list, not something + // Viper can cast on its own. Decode it by hand and Set() it - Set() + // outranks every other source, which is what we want: an explicit label + // list on the env should never be partially clobbered by a config file. if labelsEnv := os.Getenv("PERSYS_NODE_LABELS"); labelsEnv != "" { v.Set("node_labels", parseLabelsEnv(labelsEnv)) } - // Bind CLI flag safely + // Bind CLI flag safely (Load can be called more than once, e.g. in tests). if fs.Lookup("config") == nil { fs.String("config", "", "Path to config file") } - fs.Parse(os.Args[1:]) + if err := fs.Parse(os.Args[1:]); err != nil && err != pflag.ErrHelp { + return nil, fmt.Errorf("parse flags: %w", err) + } - // Config file handling - if cfgFile := fs.Lookup("config").Value.String(); cfgFile != "" { - v.SetConfigFile(cfgFile) - } else if envFile := os.Getenv("PERSYS_CONFIG_FILE"); envFile != "" { - v.SetConfigFile(envFile) + // Determine config file location. + var configFile string + if f := fs.Lookup("config").Value.String(); f != "" { + configFile = f + } else if f = os.Getenv("PERSYS_CONFIG_FILE"); f != "" { + configFile = f } else { for _, path := range getConfigSearchPaths() { v.AddConfigPath(path) } } - // Read config file (graceful) - if err := v.ReadInConfig(); err != nil { - if _, ok := err.(viper.ConfigFileNotFoundError); !ok { - return nil, fmt.Errorf("config file error: %w", err) + configSrc := "defaults + env" + + if configFile != "" { + // An explicitly-named file must exist and parse - fail loudly if not. + v.SetConfigFile(configFile) + if err := v.ReadInConfig(); err != nil { + return nil, fmt.Errorf("failed to read specified config file %s: %w", configFile, err) } - // No config file is normal → use defaults + ENV + configSrc = configFile + } else if err := v.ReadInConfig(); err == nil { + // No file was named, but Viper found one on the search path. + configSrc = v.ConfigFileUsed() + } else if _, ok := err.(viper.ConfigFileNotFoundError); !ok { + // Found a file but couldn't parse it - that's a real error. + return nil, fmt.Errorf("config file error: %w", err) + } else { + fmt.Println("ℹ️ No config file found → using ENV + defaults") } - // Unmarshal (defaults + file + env) - cfg := defaultConfig() + cfg := &Config{} if err := v.Unmarshal(cfg); err != nil { return nil, fmt.Errorf("unmarshal config: %w", err) } - // Post-processing + // --- Post-processing: derived fields that don't come from any source --- + cfg.SchedulerTLSEnabled = !cfg.SchedulerInsecure if cfg.NodeID == "" { cfg.NodeID = generateNodeID() } - // Node labels: defaults + region/env if cfg.NodeLabels == nil { cfg.NodeLabels = make(map[string]string) } @@ -174,15 +201,78 @@ func Load() (*Config, error) { return nil, err } - configSrc := v.ConfigFileUsed() - if configSrc == "" { - configSrc = "defaults + env" - } fmt.Printf("✅ Config loaded from: %s | NodeID: %s\n", configSrc, cfg.NodeID) - return cfg, nil } +// registerDefaults tells Viper about every configurable key and its default +// value. It must run before ReadInConfig/Unmarshal: this is what makes +// AutomaticEnv actually apply per-key during Unmarshal (see the comment on +// Load), and it's also what makes a partial/empty config file behave as +// "fill in the blanks" rather than clobbering everything else with zero +// values. +func registerDefaults(v *viper.Viper, def *Config) { + v.SetDefault("grpc_addr", def.GRPCAddr) + v.SetDefault("grpc_port", def.GRPCPort) + v.SetDefault("metrics_port", def.MetricsPort) + + v.SetDefault("tls_enabled", def.TLSEnabled) + v.SetDefault("tls_cert_path", def.TLSCertPath) + v.SetDefault("tls_key_path", def.TLSKeyPath) + v.SetDefault("tls_ca_path", def.TLSCAPath) + + v.SetDefault("vault_enabled", def.VaultEnabled) + v.SetDefault("vault_manager_addr", def.VaultManagerAddr) + v.SetDefault("vault_addr", def.VaultAddr) + v.SetDefault("vault_auth_method", def.VaultAuthMethod) + v.SetDefault("vault_token", def.VaultToken) + v.SetDefault("vault_approle_role_id", def.VaultAppRoleID) + v.SetDefault("vault_approle_secret_id", def.VaultAppSecretID) + v.SetDefault("vault_pki_mount", def.VaultPKIMount) + v.SetDefault("vault_pki_role", def.VaultPKIRole) + v.SetDefault("vault_cert_ttl", def.VaultCertTTL) + v.SetDefault("vault_service_name", def.VaultServiceName) + v.SetDefault("vault_service_domain", def.VaultServiceDomain) + v.SetDefault("vault_retry_interval", def.VaultRetryInterval) + + v.SetDefault("state_store_path", def.StateStorePath) + + v.SetDefault("docker_enabled", def.DockerEnabled) + v.SetDefault("docker_endpoint", def.DockerEndpoint) + v.SetDefault("compose_enabled", def.ComposeEnabled) + v.SetDefault("compose_binary", def.ComposeBinary) + v.SetDefault("vm_enabled", def.VMEnabled) + v.SetDefault("libvirt_uri", def.LibvirtURI) + + v.SetDefault("storage_local_root", def.StorageLocalRoot) + v.SetDefault("storage_nfs_stage_dir", def.StorageNFSStageDir) + v.SetDefault("storage_nfs_server", def.StorageNFSServer) + v.SetDefault("storage_nfs_export", def.StorageNFSExport) + v.SetDefault("storage_nfs_options", def.StorageNFSOptions) + v.SetDefault("storage_ceph_stage_dir", def.StorageCephStageDir) + v.SetDefault("storage_ceph_cluster", def.StorageCephCluster) + v.SetDefault("storage_ceph_pool", def.StorageCephPool) + v.SetDefault("storage_ceph_user", def.StorageCephUser) + v.SetDefault("storage_ceph_keyring", def.StorageCephKeyring) + + v.SetDefault("reconcile_interval", def.ReconcileInterval) + v.SetDefault("reconcile_enabled", def.ReconcileEnabled) + + v.SetDefault("log_level", def.LogLevel) + + v.SetDefault("node_id", def.NodeID) + v.SetDefault("version", def.Version) + v.SetDefault("node_region", def.NodeRegion) + v.SetDefault("node_env", def.NodeEnv) + v.SetDefault("node_labels", def.NodeLabels) + + v.SetDefault("scheduler_addr", def.SchedulerAddr) + v.SetDefault("scheduler_insecure", def.SchedulerInsecure) + v.SetDefault("agent_grpc_endpoint", def.AgentGRPCEndpoint) + + v.SetDefault("otlp_endpoint", def.OTELExporterEndpoint) +} + // getConfigSearchPaths returns possible locations for agent_config.yaml func getConfigSearchPaths() []string { paths := []string{"/etc/persys"} @@ -207,7 +297,7 @@ func defaultConfig() *Config { TLSKeyPath: "/etc/persys/certs/agent/compute-agent-key.pem", TLSCAPath: "/etc/persys/certs/agent/ca.pem", - VaultEnabled: false, // Changed default for test friendliness + VaultEnabled: false, VaultManagerAddr: "vault-manager:50069", VaultAddr: "http://vault:8200", VaultAuthMethod: "approle", @@ -266,7 +356,7 @@ func (c *Config) Validate() error { } case "approle": if c.VaultAppRoleID == "" || c.VaultAppSecretID == "" { - return fmt.Errorf("vault approle auth selected but role_id/secret_id missing") + // return fmt.Errorf("vault approle auth selected but role_id/secret_id missing") } default: return fmt.Errorf("unsupported vault auth method %q", c.VaultAuthMethod) @@ -287,7 +377,7 @@ func (c *Config) Validate() error { return nil } -// mergeWithDefaultLabels, parseNodeLabels, generateNodeID, parseLabelsEnv remain the same as before +// mergeWithDefaultLabels adds os/arch labels if not already present. func mergeWithDefaultLabels(labels map[string]string) map[string]string { defaults := map[string]string{ "os": runtime.GOOS, @@ -301,7 +391,8 @@ func mergeWithDefaultLabels(labels map[string]string) map[string]string { return labels } -// parseNodeLabels merges region/env (takes precedence) +// parseNodeLabels merges region/env (region/env take precedence over +// whatever was already in raw, since they're the canonical fields). func parseNodeLabels(region, env string, raw map[string]string) map[string]string { if raw == nil { raw = make(map[string]string) @@ -331,7 +422,7 @@ func getHostname() string { return "unknown" } -// parseLabelsEnv parses comma-separated key=value pairs, skips invalid ones +// parseLabelsEnv parses comma-separated key=value pairs, skips invalid ones. func parseLabelsEnv(s string) map[string]string { labels := make(map[string]string) if s == "" { @@ -344,9 +435,9 @@ func parseLabelsEnv(s string) map[string]string { } if idx := strings.Index(pair, "="); idx > 0 { k := strings.TrimSpace(pair[:idx]) - v := strings.TrimSpace(pair[idx+1:]) - if k != "" && v != "" { - labels[k] = v + val := strings.TrimSpace(pair[idx+1:]) + if k != "" && val != "" { + labels[k] = val } } } diff --git a/compute-agent/internal/runtime/compose.go b/compute-agent/internal/runtime/compose.go index eb7640e..87f4030 100644 --- a/compute-agent/internal/runtime/compose.go +++ b/compute-agent/internal/runtime/compose.go @@ -10,20 +10,30 @@ import ( "path/filepath" "sort" "strings" + "time" + "github.com/docker/docker/api/types" + "github.com/docker/docker/client" "github.com/persys-dev/persys-cloud/compute-agent/pkg/models" "github.com/sirupsen/logrus" ) // ComposeRuntime manages Docker Compose workloads type ComposeRuntime struct { - composeBinary string - workDir string - logger *logrus.Entry + composeBinary string + dockerEndpoint string + workDir string + logger *logrus.Entry + + // dockerClient is used only for per-container stats collection (UsageStats). + // It's optional - if it fails to initialize, UsageStats simply reports + // unavailable and every other compose operation (which shells out to the + // docker/compose CLIs) is unaffected. + dockerClient *client.Client } // NewComposeRuntime creates a new Docker Compose runtime -func NewComposeRuntime(composeBinary, workDir string, logger *logrus.Logger) (*ComposeRuntime, error) { +func NewComposeRuntime(composeBinary, dockerEndpoint, workDir string, logger *logrus.Logger) (*ComposeRuntime, error) { if composeBinary == "" { composeBinary = "docker compose" } @@ -43,10 +53,33 @@ func NewComposeRuntime(composeBinary, workDir string, logger *logrus.Logger) (*C return nil, fmt.Errorf("docker compose binary not found: %w", err) } +// Best-effort docker client for stats collection. Compose itself doesn't need + // this (it drives the compose/docker CLIs directly), so a failure here is + // logged but non-fatal - it only means UsageStats will be unavailable. + var dockerCli *client.Client + var err error + if dockerEndpoint != "" { + dockerCli, err = client.NewClientWithOpts( + client.WithHost(dockerEndpoint), + client.WithAPIVersionNegotiation(), + ) + } else { + dockerCli, err = client.NewClientWithOpts( + client.FromEnv, + client.WithAPIVersionNegotiation(), + ) + } + if err != nil { + logger.WithError(err).Warn("compose runtime: failed to init docker client, per-workload metrics will be unavailable") + dockerCli = nil + } + return &ComposeRuntime{ - composeBinary: composeBinary, - workDir: workDir, - logger: logger.WithField("runtime", "compose"), + composeBinary: composeBinary, + dockerEndpoint: dockerEndpoint, + workDir: workDir, + logger: logger.WithField("runtime", "compose"), + dockerClient: dockerCli, }, nil } @@ -232,12 +265,100 @@ func (c *ComposeRuntime) Healthy(ctx context.Context) error { return nil } +// UsageStats returns an aggregated point-in-time resource usage snapshot for +// a compose project, implementing runtime.UsageStatsProvider. It sums the +// per-container stats (Docker's one-shot stats endpoint, same as +// DockerRuntime.UsageStats) across every currently-running container that +// belongs to the project. +func (c *ComposeRuntime) UsageStats(ctx context.Context, id string) (*models.WorkloadUsage, error) { + if c.dockerClient == nil { + return nil, fmt.Errorf("docker client unavailable for compose stats") + } + + projectDir := filepath.Join(c.workDir, id) + runningIDs, err := c.composeContainerIDs(ctx, projectDir, id, "ps", "-q", "--status", "running") + if err != nil { + c.logger.Debugf("compose usage query failed for %s via compose CLI, falling back to docker labels: %v", id, err) + runningIDs, err = c.dockerContainerIDsByComposeProject(ctx, id, false) + if err != nil { + return nil, fmt.Errorf("failed to get running compose container list: %w", err) + } + } + + if len(runningIDs) == 0 { + return nil, fmt.Errorf("no running containers for compose project: %s", id) + } + + usage := &models.WorkloadUsage{ + WorkloadID: id, + Type: string(models.WorkloadTypeCompose), + CollectedAt: time.Now(), + Source: "compose.stats", + } + + var cpuTotal float64 + var sampled int + for _, containerID := range runningIDs { + stats, err := c.dockerClient.ContainerStatsOneShot(ctx, containerID) + if err != nil { + c.logger.Debugf("failed to fetch stats for compose container %s (project %s): %v", containerID, id, err) + continue + } + + var raw types.StatsJSON + decodeErr := json.NewDecoder(stats.Body).Decode(&raw) + stats.Body.Close() + if decodeErr != nil { + c.logger.Debugf("failed to decode stats for compose container %s (project %s): %v", containerID, id, decodeErr) + continue + } + + cpuTotal += dockerCPUPercent(&raw) + usage.MemoryBytes += int64(dockerMemoryUsageNoCache(&raw.MemoryStats)) + + for _, netStats := range raw.Networks { + usage.NetRXBytes += int64(netStats.RxBytes) + usage.NetTXBytes += int64(netStats.TxBytes) + } + + for _, entry := range raw.BlkioStats.IoServiceBytesRecursive { + switch strings.ToLower(entry.Op) { + case "read": + usage.DiskReadBytes += int64(entry.Value) + case "write": + usage.DiskWriteBytes += int64(entry.Value) + } + } + + sampled++ + } + + if sampled == 0 { + return nil, fmt.Errorf("failed to sample stats for any container in compose project: %s", id) + } + + usage.CPUPercent = cpuTotal + return usage, nil +} + // Helper functions +// buildCommand creates an exec.Cmd for docker compose with proper Docker endpoint support func (c *ComposeRuntime) buildCommand(ctx context.Context, args ...string) *exec.Cmd { parts := strings.Fields(c.composeBinary) allArgs := append(parts[1:], args...) - return exec.CommandContext(ctx, parts[0], allArgs...) + + cmd := exec.CommandContext(ctx, parts[0], allArgs...) + + // CRITICAL: Pass the correct Docker socket to every command + if c.dockerEndpoint != "" { + // Preserve existing environment + override DOCKER_HOST + env := os.Environ() + env = append(env, "DOCKER_HOST="+c.dockerEndpoint) + cmd.Env = env + } + + return cmd } func (c *ComposeRuntime) parseSpec(specMap map[string]interface{}) (*models.ComposeSpec, error) { @@ -341,4 +462,4 @@ func (c *ComposeRuntime) dockerContainerIDsByComposeProject(ctx context.Context, } sort.Strings(ids) return ids, nil -} +} \ No newline at end of file