Fixed memory and CPU to be more verbose, and also to gather them through docker stats

.env

@@ -55,6 +55,7 @@ SHUFFLE_PASS_APP_PROXY=FALSE
 TZ=Europe/Amsterdam 					# Timezone-handler in Orborus, Worker and Apps
 ORBORUS_CONTAINER_NAME=				# Used to FIND the containername. cgroup v2: issue 501
 SHUFFLE_ORBORUS_STARTUP_DELAY= 		# Used for setting up a startup delay for Orborus
+IS_KUBERNETES=false 				# Used for controlling if the environment should run in kubernetes or not
 
 SHUFFLE_BASE_IMAGE_NAME=shuffle
 SHUFFLE_BASE_IMAGE_REGISTRY=ghcr.io
@@ -67,6 +68,7 @@ SHUFFLE_SWARM_BRIDGE_DEFAULT_INTERFACE=eth0
 SHUFFLE_SWARM_BRIDGE_DEFAULT_MTU=1500 # 1500 by default
 
 # Used for auto-cleanup of containers. REALLY important at scale. Set to false to see all container info.
+SHUFFLE_MEMCACHED=
 SHUFFLE_CONTAINER_AUTO_CLEANUP=true
 SHUFFLE_ORBORUS_EXECUTION_CONCURRENCY=3 # The amount of concurrent executions Orborus can handle. This is a soft limit, but it's recommended to keep it low.
 SHUFFLE_HEALTHCHECK_DISABLED=false 

backend/go-app/go.mod

@@ -1,6 +1,6 @@
 module shuffle-shared
 
-//replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
+replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
 
 go 1.19
 

functions/onprem/orborus/go.mod

@@ -2,7 +2,7 @@ module orborus
 
 go 1.19
 
-//replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared
+replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared
 
 require (
 	github.com/docker/docker v23.0.3+incompatible

functions/onprem/orborus/orborus.go

@@ -27,6 +27,8 @@ import (
 	"strconv"
 	"strings"
 	"time"
+	"sync"
+	//"math/rand"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
@@ -39,8 +41,8 @@ import (
 	uuid "github.com/satori/go.uuid"
 
 	//"github.com/mackerelio/go-osstat/disk"
-	"github.com/mackerelio/go-osstat/memory"
+	//"github.com/mackerelio/go-osstat/memory"
-	"github.com/shirou/gopsutil/cpu"
+	//"github.com/shirou/gopsutil/cpu"
 
 	//k8s deps
 	"k8s.io/client-go/kubernetes"
@@ -68,6 +70,7 @@ var workerVersion = os.Getenv("SHUFFLE_WORKER_VERSION")
 var newWorkerImage = os.Getenv("SHUFFLE_WORKER_IMAGE")
 var dockerSwarmBridgeMTU = os.Getenv("SHUFFLE_SWARM_BRIDGE_DEFAULT_MTU")
 var dockerSwarmBridgeInterface = os.Getenv("SHUFFLE_SWARM_BRIDGE_DEFAULT_INTERFACE")
+var isKubernetes = os.Getenv("IS_KUBERNETES")
 
 // var baseimagename = "docker.pkg.github.com/shuffle/shuffle"
 // var baseimagename = "ghcr.io/frikky"
@@ -229,16 +232,18 @@ func deployServiceWorkers(image string) {
 			log.Printf("[ERROR] Failed to get network interfaces: %s", err)
 		}
 
-		mtu, err := strconv.Atoi(dockerSwarmBridgeMTU) // by default
+		mtu := 1500
-		bridgeName := dockerSwarmBridgeInterface
+		if len(dockerSwarmBridgeMTU) == 0 {
-
+			mtu, err = strconv.Atoi(dockerSwarmBridgeMTU) // by default
-		if bridgeName == "" {
+			if err != nil {
-			bridgeName = "eth0"
+				log.Printf("[DEBUG] Failed to convert the default MTU to int: %s. Using 1500 instead. Input: %s", err, dockerSwarmBridgeMTU)
+				mtu = 1500
+			}
 		}
 
-		if err != nil {
+		bridgeName := dockerSwarmBridgeInterface
-			log.Printf("[DEBUG] Failed to convert the default MTU to int: %s. Using 1500 instead", err)
+		if bridgeName == "" {
-			mtu = 1500
+			bridgeName = "eth0"
 		}
 
 		// Check if there is at least one interface
@@ -583,10 +588,7 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar {
 
 func deployWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) error {
 
-	if os.Getenv("IS_KUBERNETES") == "true" {
+	if isKubernetes == "true" {
-		// log.Printf("IS_KUBERNETS", os.Getenv("IS_KUBERNETES"))
-		// log.Printf("REGISTRY_URL", os.Getenv("REGISTRY_URL"))
-
 		if len(os.Getenv("REGISTRY_URL")) > 0 && os.Getenv("REGISTRY_URL") != "" {
 			env = append(env, fmt.Sprintf("REGISTRY_URL=%s", os.Getenv("REGISTRY_URL")))
 			env = append(env, fmt.Sprintf("IS_KUBERNETES=%s", os.Getenv("IS_KUBERNETES")))
@@ -944,6 +946,49 @@ func checkSwarmService(ctx context.Context) {
 	log.Printf("[DEBUG] Swarm info: %s\n\n", ret)
 }
 
+func getContainerResourceUsage(ctx context.Context, cli *dockerclient.Client, containerID string) (float64, float64, error) {
+	// Get container stats
+	stats, err := cli.ContainerStats(ctx, containerID, false)
+	if err != nil {
+		return 0, 0, err
+	}
+	defer stats.Body.Close()
+
+	// Parse and return CPU and memory utilization
+	cpuUsage, memoryUsage, err := parseResourceUsage(stats.Body)
+	if err != nil {
+		return 0, 0, err
+	}
+
+	return cpuUsage, memoryUsage, nil
+}
+
+func parseResourceUsage(body io.Reader) (float64, float64, error) {
+	var stats types.StatsJSON
+
+	// Decode the stream of stats as JSON
+	decoder := json.NewDecoder(body)
+	if err := decoder.Decode(&stats); err != nil {
+		return 0, 0, err
+	}
+
+	//log.Printf("CPU : %d", stats.CPUStats.CPUUsage.TotalUsage)
+	//log.Printf("CPU2: %d", stats.PreCPUStats.CPUUsage.TotalUsage)
+
+	// Calculate time difference between current and previous stats in nanoseconds
+	timeDelta := float64(stats.Read.Sub(stats.PreRead).Nanoseconds())
+
+	// Calculate CPU usage percentage
+	cpuDelta := float64(stats.CPUStats.CPUUsage.TotalUsage - stats.PreCPUStats.CPUUsage.TotalUsage)
+	cpuUsage := (cpuDelta / timeDelta) * 100.0
+
+	// Calculate memory usage percentage
+	memoryUsage := float64(stats.MemoryStats.Usage) / float64(stats.MemoryStats.Limit) * 100.0
+
+	return cpuUsage, memoryUsage, nil
+
+}
+
 func getOrborusStats() shuffle.OrborusStats {
 	newStats := shuffle.OrborusStats{
 		OrgId:        org,
@@ -956,25 +1001,97 @@ func getOrborusStats() shuffle.OrborusStats {
 		newStats.Swarm = true
 	}
 
-	if runningMode == "kubernetes" || runningMode == "k8s" {
+
-		newStats.Kubernetes = true
+	// Run this 1/10 times
-	}
+	//if rand.Intn(10) != 1 {
+	//	return newStats
+	//}
 
 	newStats.PollTime = sleepTime
 	newStats.MaxQueue = maxConcurrency
 	newStats.Queue = executionCount
 
-	// Get CPU usage and max CPU
+	if isKubernetes == "true" || runningMode == "kubernetes" || runningMode == "k8s"  {
-	/*
+		newStats.Kubernetes = true
-		before, err := cpu.Get()
+		return newStats
-		if err != nil {
+	}
-			log.Printf("[ERROR] Failed getting CPU stats: %s", err)
-		} else {
-			newStats.CPU = int(before.User)
-			newStats.MaxCPU = int(before.Total)
-		}
-	*/
 
+	// Use the docker API to get the CPU usage of the docker engine machine
+	ctx := context.Background()
+	pers, err := dockercli.Info(ctx)
+	if err != nil {
+		log.Printf("[ERROR] Failed getting docker info: %s", err)
+	} else {
+		newStats.TotalContainers = pers.Containers
+		newStats.StoppedContainers = pers.ContainersStopped
+
+		// Calculate the amount of CPU utilization on the host
+		newStats.CPU = int(pers.NCPU)
+		newStats.MaxCPU = int(pers.NCPU)
+		newStats.Memory = int(pers.MemTotal)
+		newStats.MaxMemory = int(pers.MemTotal)
+	}
+
+		// Get list of all running containers
+	containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{})
+	if err != nil {
+		log.Printf("[ERROR] Failed getting container list: %s", err)
+		return newStats
+	}
+
+	// Iterate through containers and get CPU usage
+	totalCPU := 0.0
+	memUsage := 0.0
+
+	// Use a WaitGroup to wait for all goroutines to finish
+	var wg sync.WaitGroup
+
+	// Channel to collect results
+	resultCh := make(chan struct {
+		containerID string
+		cpuUsage    float64
+		memoryUsage float64
+	})
+
+	// Iterate through containers and start a goroutine for each container
+	for _, container := range containers {
+		wg.Add(1)
+		go func(container types.Container) {
+			defer wg.Done()
+
+			// Get CPU and memory usage for the container
+			cpuUsage, memoryUsage, err := getContainerResourceUsage(ctx, dockercli, container.ID)
+			if err != nil {
+				fmt.Printf("Error getting resource usage for container %s: %v\n", container.ID, err)
+			}
+
+			// Send the result to the channel
+			resultCh <- struct {
+				containerID string
+				cpuUsage    float64
+				memoryUsage float64
+			}{container.ID, cpuUsage, memoryUsage}
+		}(container)
+	}
+
+	// Close the result channel after all goroutines are done
+	go func() {
+		wg.Wait()
+		close(resultCh)
+	}()
+
+	// Collect results from the channel
+	for result := range resultCh {
+		//fmt.Printf("Container %s CPU utilization: %.2f%%, Memory utilization: %.2f%%\n", result.containerID, result.cpuUsage, result.memoryUsage)
+		totalCPU += result.cpuUsage
+		memUsage += result.memoryUsage
+	}
+
+	newStats.CPUPercent = totalCPU
+	newStats.Memory = int(memUsage)
+	log.Printf("[DEBUG] CPU: %f, Memory: %f", totalCPU, memUsage)
+
+	/*
 	cpuPercent, err := cpu.Percent(250*time.Millisecond, false)
 	if err == nil && len(cpuPercent) > 0 {
 		newStats.CPUPercent = cpuPercent[0]
@@ -989,6 +1106,7 @@ func getOrborusStats() shuffle.OrborusStats {
 		newStats.Memory = int(memory.Used)
 		newStats.MaxMemory = int(memory.Total)
 	}
+	*/
 
 	// Get disk usage
 	/*
@@ -1048,12 +1166,10 @@ func getKubernetesClient() (*kubernetes.Clientset, error) {
 
 // Initial loop etc
 func main() {
-
 	if isRunningInCluster() {
 		log.Printf("[INFO] Running inside k8s cluster")
 	}
 
-
 	startupDelay := os.Getenv("SHUFFLE_ORBORUS_STARTUP_DELAY")
 	if len(startupDelay) > 0 {
 		log.Printf("[DEBUG] Setting startup delay to %#v", startupDelay)
@@ -1469,7 +1585,7 @@ func main() {
 func getRunningWorkers(ctx context.Context, workerTimeout int) int {
 	//log.Printf("[DEBUG] Getting running workers with API version %s", dockerApiVersion)
 	counter := 0
-	if os.Getenv("IS_KUBERNETES") == "true" {
+	if isKubernetes  == "true" {
 		log.Printf("[INFO] getting running workers in kubernetes")
 
 		thresholdTime := time.Now().Add(time.Duration(-workerTimeout) * time.Second)
@@ -1554,7 +1670,7 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int {
 // FIXME - add this to remove exited workers
 // Should it check what happened to the execution? idk
 func zombiecheck(ctx context.Context, workerTimeout int) error {
-	isK8s := os.Getenv("IS_KUBERNETES") == "true"
+	isK8s := isKubernetes == "true"
 
 	executionIds = []string{}
 	if swarmConfig == "run" || swarmConfig == "swarm" || isK8s {