feat : satellite heartbeat

ground-control/internal/server/routes.go

@@ -37,6 +37,6 @@ func (s *Server) RegisterRoutes() http.Handler {
 	r.HandleFunc("/satellites/{satellite}", s.GetSatelliteByName).Methods("GET")       // Get specific satellite
 	r.HandleFunc("/satellites/{satellite}", s.DeleteSatelliteByName).Methods("DELETE") // Delete specific satellite
 	// r.HandleFunc("/satellites/{satellite}/images", s.GetImagesForSatellite).Methods("GET") // Get satellite images
-
+	r.HandleFunc("/satellites/status",s.statusReportHandler).Methods("POST")
 	return r
 }

ground-control/internal/server/satellite_handlers.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"os"
 	"strconv"
+	"time"
 
 	"github.com/container-registry/harbor-satellite/ground-control/internal/database"
 	"github.com/container-registry/harbor-satellite/ground-control/internal/utils"
@@ -25,6 +26,18 @@ type RegisterSatelliteParams struct {
 	ConfigName string    `json:"config_name"`
 }
 
+type SatelliteStatusParams struct {
+	Name                string    `json:"name"`                  // Satellite identifier
+	Activity            string    `json:"activity"`              // Current activity satellite is doing
+	StateReportInterval string    `json:"state_report_interval"` // Interval between status reports
+	LatestStateDigest   string    `json:"latest_state_digest"`   // Digest of latest state artifact
+	LatestConfigDigest  string    `json:"latest_config_digest"`  // Digest of latest config artifact
+	MemoryUsedBytes     uint64    `json:"memory_used_bytes"`     // Memory currently used by satellite
+	StorageUsedBytes    uint64    `json:"storage_used_bytes"`    // Storage currently used by satellite
+	CPUPercent          float64   `json:"cpu_percent"`           // CPU usage percentage
+	RequestCreatedTime  time.Time `json:"request_created_time"`  // Timestamp of request creation
+}
+
 type RegisterSatelliteResponse struct {
 	Token string `json:"token"`
 }
@@ -340,6 +353,18 @@ func (s *Server) listSatelliteHandler(w http.ResponseWriter, r *http.Request) {
 	WriteJSONResponse(w, http.StatusOK, result)
 }
 
+func (s *Server) statusReportHandler(w http.ResponseWriter, r *http.Request) {
+	var req SatelliteStatusParams
+	if err := DecodeRequestBody(r, &req); err != nil {
+		log.Println(err)
+		HandleAppError(w, err)
+		return
+	}
+	// todo : process the heartbeat. eg:- save latest state in db
+	fmt.Printf("satellite reported status : %v", req)
+
+}
+
 func (s *Server) GetSatelliteByName(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	satellite := vars["satellite"]

internal/satellite/satellite.go

@@ -2,7 +2,6 @@ package satellite
 
 import (
 	"context"
-
 	"github.com/container-registry/harbor-satellite/internal/logger"
 	"github.com/container-registry/harbor-satellite/internal/scheduler"
 	"github.com/container-registry/harbor-satellite/internal/state"
@@ -24,16 +23,19 @@ func NewSatellite(cm *config.ConfigManager) *Satellite {
 func (s *Satellite) Run(ctx context.Context) error {
 	log := logger.FromContext(ctx)
 	log.Info().Msg("Starting Satellite")
+	var heartbeatPayload scheduler.UpstreamInfo
 
 	fetchAndReplicateStateProcess := state.NewFetchAndReplicateStateProcess(s.cm)
 	ztrProcess := state.NewZtrProcess(s.cm)
+	statusReportingProcess := state.NewStatusReportingProcess(s.cm)
 
 	// Create schedulers instead of using ScheduleFunc
 	if !s.cm.IsZTRDone() {
 		ztrScheduler, err := scheduler.NewSchedulerWithInterval(
 			s.cm.GetRegistrationInterval(),
 			ztrProcess,
 			log,
+			&heartbeatPayload,
 		)
 		if err != nil {
 			log.Error().Err(err).Msg("Failed to create ZTR scheduler")
@@ -48,14 +50,35 @@ func (s *Satellite) Run(ctx context.Context) error {
 		s.cm.GetStateReplicationInterval(),
 		fetchAndReplicateStateProcess,
 		log,
+		&heartbeatPayload,
 	)
+
 	if err != nil {
 		log.Error().Err(err).Msg("Failed to create state replication scheduler")
 		return err
 	}
 	s.schedulers = append(s.schedulers, stateScheduler)
+
+	// Create state reporting
+	statusReportingScheduler, err := scheduler.NewSchedulerWithInterval(
+		s.cm.GetStateReportingInterval(),
+		statusReportingProcess,
+		log,
+		&heartbeatPayload,
+	)
+
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to create status reporting scheduler")
+		return err
+	}
+	s.schedulers = append(s.schedulers, statusReportingScheduler)
+
 	go stateScheduler.Run(ctx)
 
+	if !(s.cm.IsHeartbeatDisabled()) {
+		go statusReportingScheduler.Run(ctx)
+	}
+
 	return ctx.Err()
 }
 

internal/scheduler/process.go

@@ -10,7 +10,7 @@ type Process interface {
 	Name() string
 
 	// Execute runs the process
-	Execute(ctx context.Context) error
+	Execute(ctx context.Context, upstreamPayload *UpstreamInfo) error
 
 	// IsRunning returns true if the process is running
 	IsRunning() bool

internal/scheduler/scheduler.go

@@ -10,30 +10,45 @@ import (
 	"github.com/rs/zerolog"
 )
 
+type UpstreamInfo struct {
+	LatestStateDigest  string
+	LatestConfigDigest string
+	CurrentActivity    string
+	StateURL           string
+}
+
 // Scheduler manages the execution of processes with configurable intervals
 type Scheduler struct {
-	name     string
-	ticker   *time.Ticker
-	process  Process
-	log      *zerolog.Logger
-	interval time.Duration
-	mu       sync.Mutex
+	name            string
+	ticker          *time.Ticker
+	process         Process
+	log             *zerolog.Logger
+	interval        time.Duration
+	mu              sync.Mutex
+	upstreamPayload *UpstreamInfo
 }
 
+const (
+	ActivityStateSynced      = "state synced successfully"
+	ActivityEncounteredError = "encountered error"
+	ActivityReconcilingState = "reconciling state"
+)
+
 // NewSchedulerWithInterval creates a new scheduler with a parsed interval string
-func NewSchedulerWithInterval(intervalExpr string, process Process, log *zerolog.Logger) (*Scheduler, error) {
-	duration, err := parseEveryExpr(intervalExpr)
+func NewSchedulerWithInterval(intervalExpr string, process Process, log *zerolog.Logger, upstreamPayload *UpstreamInfo) (*Scheduler, error) {
+	duration, err := ParseEveryExpr(intervalExpr)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse interval: %w", err)
 	}
 
 	ticker := time.NewTicker(duration)
 	scheduler := &Scheduler{
-		name:     process.Name(),
-		ticker:   ticker,
-		process:  process,
-		log:      log,
-		interval: duration,
+		name:            process.Name(),
+		ticker:          ticker,
+		process:         process,
+		log:             log,
+		interval:        duration,
+		upstreamPayload: upstreamPayload,
 	}
 
 	return scheduler, nil
@@ -83,7 +98,7 @@ func (s *Scheduler) ResetInterval(newInterval time.Duration) {
 
 // ResetIntervalFromExpr changes the ticker interval using an expression string
 func (s *Scheduler) ResetIntervalFromExpr(intervalExpr string) error {
-	duration, err := parseEveryExpr(intervalExpr)
+	duration, err := ParseEveryExpr(intervalExpr)
 	if err != nil {
 		return fmt.Errorf("failed to parse interval: %w", err)
 	}
@@ -119,7 +134,7 @@ func (s *Scheduler) launchProcess(ctx context.Context) {
 			Msg("Scheduler triggering task execution")
 
 		go func() {
-			if err := s.process.Execute(ctx); err != nil {
+			if err := s.process.Execute(ctx, s.upstreamPayload); err != nil {
 				s.log.Warn().
 					Str("Process", s.process.Name()).
 					Err(err).
@@ -133,7 +148,7 @@ func (s *Scheduler) launchProcess(ctx context.Context) {
 	}
 }
 
-func parseEveryExpr(expr string) (time.Duration, error) {
+func ParseEveryExpr(expr string) (time.Duration, error) {
 	const prefix = "@every "
 	if expr == "" {
 		return 0, fmt.Errorf("empty expression provided")

internal/state/registration_process.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 
 	"github.com/container-registry/harbor-satellite/internal/logger"
+	"github.com/container-registry/harbor-satellite/internal/scheduler"
 	"github.com/container-registry/harbor-satellite/pkg/config"
 	"github.com/rs/zerolog"
 )
@@ -38,7 +39,7 @@ func NewZtrProcess(cm *config.ConfigManager) *ZtrProcess {
 	}
 }
 
-func (z *ZtrProcess) Execute(ctx context.Context) error {
+func (z *ZtrProcess) Execute(ctx context.Context, upstreamPayload *scheduler.UpstreamInfo) error {
 	z.start()
 	defer z.stop()
 
@@ -72,6 +73,7 @@ func (z *ZtrProcess) Execute(ctx context.Context) error {
 
 	// Close the z.Done channel on successful ZTR alone.
 	close(z.Done)
+
 	return nil
 }
 

internal/state/report.go

@@ -0,0 +1,112 @@
+package state
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/shirou/gopsutil/v4/cpu"
+	"github.com/shirou/gopsutil/v4/disk"
+	"github.com/shirou/gopsutil/v4/mem"
+)
+
+type StatusReportParams struct {
+	Name                string    `json:"name"`                  // Satellite identifier
+	Activity            string    `json:"activity"`              // Current activity satellite is doing
+	StateReportInterval string    `json:"state_report_interval"` // Interval between status reports
+	LatestStateDigest   string    `json:"latest_state_digest"`   // Digest of latest state artifact
+	LatestConfigDigest  string    `json:"latest_config_digest"`  // Digest of latest config artifact
+	MemoryUsedBytes     uint64    `json:"memory_used_bytes"`     // Memory currently used by satellite
+	StorageUsedBytes    uint64    `json:"storage_used_bytes"`    // Storage currently used by satellite
+	CPUPercent          float64   `json:"cpu_percent"`           // CPU usage percentage
+	RequestCreatedTime  time.Time `json:"request_created_time"`  // Timestamp of request creation
+}
+
+func collectStatusReportParams(ctx context.Context, duration time.Duration, req *StatusReportParams) error {
+	cpuPercent, err := getAvgCpuUsage(ctx, 1*time.Second, duration)
+	if err != nil {
+		return err
+	}
+
+	memUsed, err := getMemoryUsedBytes(ctx)
+	if err != nil {
+		return err
+	}
+
+	storUsed, err := getStorageUsedBytes(ctx, "/")
+	if err != nil {
+		return err
+	}
+
+	req.CPUPercent = cpuPercent
+	req.MemoryUsedBytes = memUsed
+	req.StorageUsedBytes = storUsed
+	req.RequestCreatedTime = time.Now()
+
+	return nil
+}
+
+func getAvgCpuUsage(ctx context.Context, sampleInterval time.Duration, totalDuration time.Duration) (float64, error) {
+	var sum float64
+	var count int
+
+	ticker := time.NewTicker(sampleInterval)
+	defer ticker.Stop()
+
+	timeout := time.After(totalDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return 0, ctx.Err()
+		case <-timeout:
+			if count == 0 {
+				return 0, fmt.Errorf("no samples collected")
+			}
+			avg := sum / float64(count)
+			return math.Round(avg*100) / 100, nil
+		case <-ticker.C:
+			percent, err := cpu.PercentWithContext(ctx, 0, false)
+			if err != nil {
+				continue
+			}
+			if len(percent) > 0 {
+				sum += percent[0]
+				count++
+			}
+		}
+	}
+}
+
+func getStorageUsedBytes(ctx context.Context, path string) (uint64, error) {
+	usageStat, err := disk.UsageWithContext(ctx, path)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get storage used: %w", err)
+	}
+	return usageStat.Used, nil
+}
+
+func getMemoryUsedBytes(ctx context.Context) (uint64, error) {
+	vmStat, err := mem.VirtualMemoryWithContext(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get memory used: %w", err)
+	}
+	return vmStat.Used, nil
+}
+
+func extractSatelliteNameFromURL(stateURL string) (string, error) {
+	u, err := url.Parse(stateURL)
+	if err != nil {
+		return "", fmt.Errorf("invalid state URL %q: %w", stateURL, err)
+	}
+
+	parts := strings.FieldsFunc(u.Path, func(r rune) bool { return r == '/' })
+	if len(parts) < 4 {
+		return "", fmt.Errorf("state URL %q does not have enough path segments to extract satellite name", stateURL)
+	}
+
+	return parts[2], nil
+}