v0.12.0

agent/cmd/agent/main.go

@@ -20,6 +20,7 @@ import (
 	"techulus/cloud-agent/internal/dns"
 	agenthttp "techulus/cloud-agent/internal/http"
 	"techulus/cloud-agent/internal/logs"
+	"techulus/cloud-agent/internal/metrics"
 	"techulus/cloud-agent/internal/network"
 	"techulus/cloud-agent/internal/paths"
 	"techulus/cloud-agent/internal/reconcile"
@@ -34,17 +35,19 @@ var (
 
 func main() {
 	var (
-		controlPlaneURL  string
-		token            string
-		isProxy          bool
-		logsEndpointFlag string
-		disableDNS       bool
+		controlPlaneURL     string
+		token               string
+		isProxy             bool
+		logsEndpointFlag    string
+		metricsEndpointFlag string
+		disableDNS          bool
 	)
 
 	flag.StringVar(&controlPlaneURL, "url", "", "Control plane URL (required)")
 	flag.StringVar(&token, "token", "", "Registration token (required for first run)")
 	flag.BoolVar(&isProxy, "proxy", false, "Run as proxy node (handles TLS and public traffic)")
 	flag.StringVar(&logsEndpointFlag, "logs-endpoint", "", "Override logs endpoint URL (optional)")
+	flag.StringVar(&metricsEndpointFlag, "metrics-endpoint", "", "Override metrics endpoint URL (optional)")
 	flag.BoolVar(&disableDNS, "no-dns", false, "Disable local DNS server")
 	flag.Parse()
 
@@ -53,6 +56,7 @@ func main() {
 	}
 
 	var logsEndpoint string
+	var metricsEndpoint string
 
 	if isProxy && runtime.GOOS != "linux" {
 		log.Fatal("--proxy flag is only supported on Linux")
@@ -112,6 +116,11 @@ func main() {
 		} else if config.LoggingEndpoint != "" {
 			logsEndpoint = config.LoggingEndpoint
 		}
+		if metricsEndpointFlag != "" {
+			metricsEndpoint = metricsEndpointFlag
+		} else if config.MetricsEndpoint != "" {
+			metricsEndpoint = config.MetricsEndpoint
+		}
 
 		if err = container.EnsureNetwork(config.SubnetID); err != nil {
 			log.Printf("Warning: Failed to ensure container network: %v", err)
@@ -162,6 +171,11 @@ func main() {
 			respLoggingEndpoint = *resp.LoggingEndpoint
 		}
 
+		var respMetricsEndpoint string
+		if resp.MetricsEndpoint != nil {
+			respMetricsEndpoint = *resp.MetricsEndpoint
+		}
+
 		var registryURL, registryUsername, registryPassword string
 		if resp.RegistryURL != nil {
 			registryURL = *resp.RegistryURL
@@ -180,6 +194,7 @@ func main() {
 			EncryptionKey:    resp.EncryptionKey,
 			IsProxy:          isProxy,
 			LoggingEndpoint:  respLoggingEndpoint,
+			MetricsEndpoint:  respMetricsEndpoint,
 			RegistryURL:      registryURL,
 			RegistryUsername: registryUsername,
 			RegistryPassword: registryPassword,
@@ -191,6 +206,11 @@ func main() {
 		} else if respLoggingEndpoint != "" {
 			logsEndpoint = respLoggingEndpoint
 		}
+		if metricsEndpointFlag != "" {
+			metricsEndpoint = metricsEndpointFlag
+		} else if respMetricsEndpoint != "" {
+			metricsEndpoint = respMetricsEndpoint
+		}
 
 		if err = configuration.Save(config); err != nil {
 			log.Fatalf("Failed to save config: %v", err)
@@ -251,6 +271,7 @@ func main() {
 	var traefikLogCollector *logs.TraefikCollector
 	var logsSender *logs.VictoriaLogsSender
 	var agentLogWriter *logs.AgentLogWriter
+	var metricsSender agent.MetricsSender
 
 	if logsEndpoint != "" {
 		log.Println("[logs] log collection enabled, endpoint:", logsEndpoint)
@@ -265,6 +286,11 @@ func main() {
 		log.Println("[agent-logs] Agent log collection enabled")
 	}
 
+	if metricsEndpoint != "" {
+		log.Println("[metrics] metrics collection enabled, endpoint:", metricsEndpoint)
+		metricsSender = metrics.NewVictoriaMetricsSender(metricsEndpoint, config.ServerID)
+	}
+
 	var builder *build.Builder
 	if logsSender != nil {
 		builder = build.NewBuilder(dataDir, logsSender)
@@ -294,7 +320,7 @@ func main() {
 	privateIP := network.PrivateIP()
 	log.Printf("Agent v%s started. Public IP: %s, Private IP: %s. Tick interval: %v", agent.Version, publicIP, privateIP, agent.TickInterval)
 
-	agentInstance := agent.NewAgent(client, reconciler, config, publicIP, privateIP, dataDir, logCollector, traefikLogCollector, builder, config.IsProxy, disableDNS)
+	agentInstance := agent.NewAgent(client, reconciler, config, publicIP, privateIP, dataDir, logCollector, traefikLogCollector, metricsSender, builder, config.IsProxy, disableDNS)
 	agentInstance.Run(ctx)
 
 	if agentLogFlusherDone != nil {

agent/internal/agent/agent.go

@@ -6,6 +6,7 @@ import (
 
 	"techulus/cloud-agent/internal/build"
 	"techulus/cloud-agent/internal/container"
+	"techulus/cloud-agent/internal/health"
 	agenthttp "techulus/cloud-agent/internal/http"
 	"techulus/cloud-agent/internal/logs"
 	"techulus/cloud-agent/internal/reconcile"
@@ -42,6 +43,7 @@ type Config struct {
 	EncryptionKey    string `json:"encryptionKey"`
 	IsProxy          bool   `json:"isProxy"`
 	LoggingEndpoint  string `json:"loggingEndpoint,omitempty"`
+	MetricsEndpoint  string `json:"metricsEndpoint,omitempty"`
 	RegistryURL      string `json:"registryUrl,omitempty"`
 	RegistryUsername string `json:"registryUsername,omitempty"`
 	RegistryPassword string `json:"registryPassword,omitempty"`
@@ -78,6 +80,7 @@ type Agent struct {
 	processingStart             time.Time
 	LogCollector                *logs.Collector
 	TraefikLogCollector         *logs.TraefikCollector
+	MetricsSender               MetricsSender
 	Builder                     *build.Builder
 	isBuilding                  bool
 	buildMutex                  sync.Mutex
@@ -94,6 +97,7 @@ func NewAgent(
 	publicIP, privateIP, dataDir string,
 	logCollector *logs.Collector,
 	traefikLogCollector *logs.TraefikCollector,
+	metricsSender MetricsSender,
 	builder *build.Builder,
 	isProxy bool,
 	disableDNS bool,
@@ -110,12 +114,17 @@ func NewAgent(
 		DataDir:               dataDir,
 		LogCollector:          logCollector,
 		TraefikLogCollector:   traefikLogCollector,
+		MetricsSender:         metricsSender,
 		Builder:               builder,
 		IsProxy:               isProxy,
 		DisableDNS:            disableDNS,
 	}
 }
 
+type MetricsSender interface {
+	SendSystemStats(stats *health.SystemStats, collectedAt time.Time) error
+}
+
 func (a *Agent) GetState() AgentState {
 	a.stateMutex.RLock()
 	defer a.stateMutex.RUnlock()

agent/internal/agent/backup.go

@@ -7,12 +7,14 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log"
 	"os"
 	"path/filepath"
 	"strings"
+	"syscall"
 
 	"techulus/cloud-agent/internal/container"
 	agenthttp "techulus/cloud-agent/internal/http"
@@ -240,7 +242,7 @@ func (a *Agent) processVolumeRestore(backupID, serviceID, containerID, volumeNam
 		return reportFailure(fmt.Errorf("failed to create volume parent directory: %w", err))
 	}
 
-	if err := os.Rename(tempExtractPath, volumePath); err != nil {
+	if err := moveDir(tempExtractPath, volumePath); err != nil {
 		startContainerWithRetry()
 		return reportFailure(fmt.Errorf("failed to move restored data to volume path: %w", err))
 	}
@@ -256,6 +258,96 @@ func (a *Agent) processVolumeRestore(backupID, serviceID, containerID, volumeNam
 	return nil
 }
 
+func moveDir(src, dst string) error {
+	if err := os.Rename(src, dst); err == nil {
+		return nil
+	} else if !errors.Is(err, syscall.EXDEV) {
+		return err
+	}
+
+	if err := copyDir(src, dst); err != nil {
+		return err
+	}
+
+	return os.RemoveAll(src)
+}
+
+func copyDir(src, dst string) error {
+	info, err := os.Stat(src)
+	if err != nil {
+		return err
+	}
+	if !info.IsDir() {
+		return fmt.Errorf("source is not a directory: %s", src)
+	}
+
+	if err := os.MkdirAll(dst, info.Mode()); err != nil {
+		return err
+	}
+
+	entries, err := os.ReadDir(src)
+	if err != nil {
+		return err
+	}
+
+	for _, entry := range entries {
+		srcPath := filepath.Join(src, entry.Name())
+		dstPath := filepath.Join(dst, entry.Name())
+		entryInfo, err := os.Lstat(srcPath)
+		if err != nil {
+			return err
+		}
+
+		if entryInfo.IsDir() {
+			if err := copyDir(srcPath, dstPath); err != nil {
+				return err
+			}
+			continue
+		}
+
+		if entryInfo.Mode()&os.ModeSymlink != 0 {
+			linkTarget, err := os.Readlink(srcPath)
+			if err != nil {
+				return err
+			}
+			if err := os.Symlink(linkTarget, dstPath); err != nil {
+				return err
+			}
+			continue
+		}
+
+		if !entryInfo.Mode().IsRegular() {
+			return fmt.Errorf("unsupported file type in restore archive: %s", srcPath)
+		}
+
+		if err := copyFile(srcPath, dstPath, entryInfo.Mode()); err != nil {
+			return err
+		}
+	}
+
+	return os.Chmod(dst, info.Mode())
+}
+
+func copyFile(src, dst string, mode os.FileMode) error {
+	in, err := os.Open(src)
+	if err != nil {
+		return err
+	}
+	defer in.Close()
+
+	out, err := os.OpenFile(dst, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, mode)
+	if err != nil {
+		return err
+	}
+
+	_, copyErr := io.Copy(out, in)
+	closeErr := out.Close()
+	if copyErr != nil {
+		return copyErr
+	}
+	return closeErr
+}
+
 func createS3Client(cfg StorageConfig) (*s3.Client, error) {
 	awsCfg, err := config.LoadDefaultConfig(context.Background(),
 		config.WithRegion(cfg.Region),

agent/internal/agent/reporting.go

@@ -39,7 +39,15 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport
 
 	healthCollectMu.Lock()
 	if time.Since(lastHealthCollect) >= 60*time.Second {
-		report.HealthStats = health.CollectSystemStats()
+		collectedAt := time.Now()
+		systemStats := health.CollectSystemStats()
+		if a.MetricsSender != nil {
+			go func() {
+				if err := a.MetricsSender.SendSystemStats(systemStats, collectedAt); err != nil {
+					log.Printf("[metrics] failed to send system stats: %v", err)
+				}
+			}()
+		}
 		report.NetworkHealth = health.CollectNetworkHealth("wg0")
 		report.ContainerHealth = health.CollectContainerHealth()
 		report.AgentHealth = &agenthttp.AgentHealth{
@@ -48,8 +56,8 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport
 		}
 		lastHealthCollect = time.Now()
 		log.Printf("[health] collected: cpu=%.1f%%, mem=%.1f%%, disk=%.1f%%, network=%v, containers=%d running",
-			report.HealthStats.CpuUsagePercent, report.HealthStats.MemoryUsagePercent,
-			report.HealthStats.DiskUsagePercent, report.NetworkHealth.TunnelUp,
+			systemStats.CpuUsagePercent, systemStats.MemoryUsagePercent,
+			systemStats.DiskUsagePercent, report.NetworkHealth.TunnelUp,
 			report.ContainerHealth.RunningContainers)
 	}
 	healthCollectMu.Unlock()

agent/internal/api/client.go

@@ -38,6 +38,7 @@ type RegisterResponse struct {
 	WireGuardIP      string  `json:"wireguardIp"`
 	EncryptionKey    string  `json:"encryptionKey"`
 	LoggingEndpoint  *string `json:"loggingEndpoint"`
+	MetricsEndpoint  *string `json:"metricsEndpoint"`
 	RegistryURL      *string `json:"registryUrl"`
 	RegistryUsername *string `json:"registryUsername"`
 	RegistryPassword *string `json:"registryPassword"`

agent/internal/http/client.go

@@ -251,7 +251,6 @@ type StatusReport struct {
 	Meta            map[string]string       `json:"meta,omitempty"`
 	Containers      []ContainerStatus       `json:"containers"`
 	DnsInSync       bool                    `json:"dnsInSync,omitempty"`
-	HealthStats     *health.SystemStats     `json:"healthStats,omitempty"`
 	NetworkHealth   *health.NetworkHealth   `json:"networkHealth,omitempty"`
 	ContainerHealth *health.ContainerHealth `json:"containerHealth,omitempty"`
 	AgentHealth     *AgentHealth            `json:"agentHealth,omitempty"`