✨ improve FL metrics collection and add mac environment setup (open-cluster-management-io#72)

* hello

Signed-off-by: Meng Yan <myan@redhat.com>

* test

Signed-off-by: Meng Yan <myan@redhat.com>

* print the value

Signed-off-by: Meng Yan <myan@redhat.com>

* test

Signed-off-by: Meng Yan <myan@redhat.com>

* update metrics

Signed-off-by: Meng Yan <myan@redhat.com>

---------

Signed-off-by: Meng Yan <myan@redhat.com>

Author: yanmxa

federated-learning-controller/deploy/ocm/mac-env.sh

@@ -0,0 +1,43 @@
+#!/bin/bash
+
+cd $(dirname ${BASH_SOURCE})
+
+set -e
+
+hub=${HUB:-hub}
+c1=${CLUSTER1:-cluster1}
+c2=${CLUSTER2:-cluster2}
+
+hubctx="kind-${hub}"
+c1ctx="kind-${c1}"
+c2ctx="kind-${c2}"
+
+cat <<EOF | kind create cluster --name "${hub}" --config=-
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+nodes:
+- role: control-plane
+  extraPortMappings:
+  - containerPort: 30090
+    hostPort: 30090
+    protocol: TCP
+EOF
+
+kind create cluster --name "${c1}"
+kind create cluster --name "${c2}"
+
+echo "Initialize the ocm hub cluster"
+clusteradm init --wait --context ${hubctx}
+joincmd=$(clusteradm get token --context ${hubctx} | grep clusteradm)
+
+echo "Join clusters to hub"
+eval "${joincmd//<cluster_name>/local-cluster} --force-internal-endpoint-lookup --wait --context ${hubctx}"
+eval "${joincmd//<cluster_name>/${c1}} --force-internal-endpoint-lookup --wait --context ${c1ctx}"
+eval "${joincmd//<cluster_name>/${c2}} --force-internal-endpoint-lookup --wait --context ${c2ctx}"
+
+echo "Accept join of clusters"
+clusteradm accept --context ${hubctx} --clusters ${c1},${c2},local-cluster --wait
+
+# label local-cluster
+kubectl label managedclusters local-cluster local-cluster=true --context ${hubctx}
+kubectl get managedclusters --all-namespaces --context ${hubctx}

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

@@ -24,45 +24,49 @@ def __init__(self, net, trainloader, valloader, local_epochs):
 
     def fit(self, parameters, config):
         set_weights(self.net, parameters)
-      
+        
+        # the aggregated model from the previous round
+        
         optimizer = optim.Adadelta(self.net.parameters(), lr=1.0)
         scheduler = StepLR(optimizer, step_size=1, gamma=0.7)
 
-        epoch_loss = 0.0
+        train_loss = 0.0
         for epoch in range(1, self.local_epochs + 1):
             # test_loss = test(model, device, test_loader)
             # test_losses.append(test_loss)
             loss = train(self.net, self.device, self.trainloader, optimizer, epoch)
-            metrics = {
-                "train_loss": loss,
-            }
-            labels = {
-                "round": config["server_round"],
-                "epoch": epoch,
-            }
-            write_metrics(metrics, labels)
-            epoch_loss += loss
+            train_loss += loss
             scheduler.step()
-            
+        
+        # client loss and accuracy   
+        loss, accuracy = test(self.net, self.device, self.valloader)
+        metrics = {
+            "train_loss": loss, 
+            "train_accuracy": accuracy, 
+        }
+        labels = {
+            "round": config["server_round"],
+        }
+        write_metrics(metrics, labels) 
+        
         return (
             get_weights(self.net),
             len(self.trainloader.dataset),
-            {"train_loss": epoch_loss / self.local_epochs},
+            {"train_loss": train_loss / self.local_epochs},
         )
 
     def evaluate(self, parameters, config):
+        # the aggregated model from the current round
         set_weights(self.net, parameters)
-        
         loss, accuracy = test(self.net, self.device, self.valloader)
-        print(f"Evaluation Loss: {loss}, Accuracy: {accuracy}")
         metrics = {
+            "accuracy": accuracy, 
             "loss": loss,
-            "accuracy": accuracy,
         }
         labels = {
             "round": config["server_round"],
         }
-        write_metrics(metrics, labels)
+        write_metrics(metrics, labels) 
         return loss, len(self.valloader.dataset), {"accuracy": accuracy, "loss": loss}
 
 def client_fn(context: Context):

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

@@ -110,14 +110,59 @@ func (r *Reporter) UpdateMetrics(newMetrics map[string]float64, newLabels map[st
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
+	// Check if metrics values have actually changed (excluding timestamp)
+	metricsChanged := false
+	
+	// Compare metrics count (excluding timestamp which we'll add/update later)
+	currentMetricsCount := len(r.metrics)
+	if _, hasTimestamp := r.metrics["timestamp"]; hasTimestamp {
+		currentMetricsCount--
+	}
+	
+	if len(newMetrics) != currentMetricsCount {
+		metricsChanged = true
+	} else {
+		for name, newValue := range newMetrics {
+			// Skip timestamp comparison since we always update it
+			if name == "timestamp" {
+				continue
+			}
+			if oldValue, exists := r.metrics[name]; !exists || oldValue != newValue {
+				metricsChanged = true
+				break
+			}
+		}
+	}
+
+	// Check if labels have changed
+	labelsChanged := false
+	if len(newLabels) != len(r.labels) {
+		labelsChanged = true
+	} else {
+		for name, newValue := range newLabels {
+			if oldValue, exists := r.labels[name]; !exists || oldValue != newValue {
+				labelsChanged = true
+				break
+			}
+		}
+	}
+
+	// Only update if values have actually changed
+	if !metricsChanged && !labelsChanged {
+		log.Println("No metric or label changes detected, skipping update")
+		return
+	}
+
 	// Update the metrics and labels
 	r.metrics = newMetrics
 	r.labels = newLabels
 
 	// Always overwrite "timestamp" metric with the current Unix time.
-	// This ensures we have a fresh timestamp regardless of input.
+	// This ensures we have a fresh timestamp only when other metrics change.
 	r.metrics["timestamp"] = float64(time.Now().Unix())
 
+	log.Printf("Metrics updated: %d metrics, %d labels", len(r.metrics), len(r.labels))
+
 	// Step 1: Detect if the set of metrics has changed
 	// (different number of metrics or different names)
 	needsUpdate := false