Merge pull request #1 from yanmxa/patch_sidecar

patch sidecar

Author: mrrr3d

federated-learning-controller/deploy/obs/hack/certs/generate-certs.sh

@@ -37,8 +37,16 @@ kubectl --context kind-hub create secret tls otel-signer -n open-cluster-managem
 
 # replace root-ca.crt in deploy/resources/addon-template.yaml
 awk '{print "              " $0}' root-ca.crt > root-ca.crt.tmp
-sed -i "/PROM_WEB_ROOT_CA/{
+# Cross-platform sed -i compatibility
+if [[ "$OSTYPE" == "darwin"* ]]; then
+    sed -i '' "/PROM_WEB_ROOT_CA/{
         r root-ca.crt.tmp
         d
     }" ../../otel-addon/resources/addon-template.yaml
+else
+    sed -i "/PROM_WEB_ROOT_CA/{
+        r root-ca.crt.tmp
+        d
+    }" ../../otel-addon/resources/addon-template.yaml
+fi
 rm root-ca.crt.tmp

federated-learning-controller/examples/flower/app-torch/app_torch/client_app.py

@@ -56,14 +56,14 @@ def evaluate(self, parameters, config):
         loss, accuracy = test(self.net, self.device, self.valloader)
         print(f"Evaluation Loss: {loss}, Accuracy: {accuracy}")
         metrics = {
-            "evaluation_loss": loss,
+            "loss": loss,
             "accuracy": accuracy,
         }
         labels = {
             "round": config["server_round"],
         }
         write_metrics(metrics, labels)
-        return loss, len(self.valloader.dataset), {"accuracy": accuracy}
+        return loss, len(self.valloader.dataset), {"accuracy": accuracy, "loss": loss}
 
 def client_fn(context: Context):
     # Load model and data

federated-learning-controller/examples/flower/app-torch/app_torch/server_app.py

@@ -40,8 +40,9 @@ def parse_arguments():
 def client_weighted_average(metrics: List[Tuple[int, Metrics]]) -> Metrics:
     """Compute weighted average of client accuracy."""
     accuracies = [num_examples * metric["accuracy"] for num_examples, metric in metrics]
+    losses = [num_examples * metric["loss"] for num_examples, metric in metrics]
     examples = [num_examples for num_examples, _ in metrics]
-    return {"accuracy": sum(accuracies) / sum(examples)}
+    return {"accuracy": sum(accuracies) / sum(examples), "loss": sum(losses) / sum(examples)}
 
 def fit_evaluate_config(server_round: int):
     config = {

federated-learning-controller/internal/controller/manifests/client/manifestwork.yaml

@@ -39,6 +39,7 @@ spec:
               labels:
                 job-name: {{ .ClientJobName }} # Ensure labels match the selector
             spec:
+              shareProcessNamespace: true
               volumes:
                 - name: metric-data
                   emptyDir: {}

federated-learning-controller/internal/controller/manifests/server/server-job.yaml

@@ -12,6 +12,7 @@ spec:
       labels:
         job-name: {{ .Name }} # Ensure labels match the selector
     spec:
+      shareProcessNamespace: true
       containers:
       - name: flower-server
         image: {{ .Image }}

federated-learning-controller/internal/sidecar/main.go

@@ -2,14 +2,17 @@ package main
 
 import (
 	"context"
-	"fl_sidecar/exporter"
-	"fl_sidecar/watcher"
 	"flag"
 	"fmt"
 	"log"
 	"os"
+	"os/exec"
 	"os/signal"
 	"syscall"
+	"time"
+
+	"fl_sidecar/exporter"
+	"fl_sidecar/watcher"
 )
 
 // Command-line flags
@@ -65,12 +68,32 @@ func main() {
 
 	log.Printf("Start watching file %s", metricFile)
 
+	// Check if main container process is still running periodically
+	go func() {
+		for {
+			// Check if main container is still running (when shareProcessNamespace is enabled)
+			if checkMainContainerExited() {
+				log.Println("Main container exited, exiting sidecar...")
+				cancel()
+				return
+			}
+
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				// Check every 5 seconds
+				time.Sleep(5 * time.Second)
+			}
+		}
+	}()
+
 	// Main loop to process file updates and handle shutdown
 	for {
 		select {
 		case content, ok := <-updateChan:
 			if !ok {
-				log.Fatalf("Watcher channel closed")
+				log.Println("Watcher channel closed")
 				return
 			}
 
@@ -84,6 +107,30 @@ func main() {
 	}
 }
 
+// checkMainContainerExited checks if the main container (flower-server or flower-client) process has exited
+// This works when shareProcessNamespace is enabled in the pod spec
+func checkMainContainerExited() bool {
+	// Check if the main flower server or client process is still running
+	// Try different patterns for server and client containers
+	patterns := []string{
+		".*server.*--num-rounds",     // Server process pattern
+		".*client.*--data-config",    // Client process pattern
+		".*client.*--server-address", // Alternative client pattern
+	}
+
+	for _, pattern := range patterns {
+		cmd := exec.Command("pgrep", "-f", pattern)
+		err := cmd.Run()
+		if err == nil {
+			// Found a matching process, main container is still running
+			return false
+		}
+	}
+
+	// No matching processes found, main container has likely exited
+	return true
+}
+
 // parseAndPushMtrics parses the content of the metric file and pushes the metrics to the reporter.
 func parseAndPushMtrics(reporter *exporter.Reporter, content []byte) {
 	metrics, labels, err := exporter.ParseContetnt(content)