PROP-96 - Add Federated ML to Propeller

.gitignore

@@ -14,3 +14,7 @@ embed-proplet/bootloader
 embed-proplet/modules
 embed-proplet/tools
 embed-proplet/zephyr
+
+# ML-WASM example generated files
+examples/ml-wasm/mymodel.pkl
+examples/ml-wasm/mymodel.wasm

Makefile

@@ -120,6 +120,16 @@ lint:
 	golangci-lint run  --config .golangci.yaml
 	cd proplet && cargo check --release && cargo fmt --all -- --check && cargo clippy -- -D warnings
 
+test:
+	go test -v ./manager -run TestAggregateJSONF64
+	go test -v ./manager -run TestAggregateConcat
+	go test -v ./manager -run TestAggregateRound
+	go test -v ./manager -run TestFLWorkflowIntegration
+
+test-all:
+	go test -v ./...
+	cd proplet && cargo test --release
+
 start-supermq:
 	docker compose -f docker/compose.yaml --env-file docker/.env up -d
 
@@ -143,6 +153,8 @@ help:
 	@echo "  install:          install the binary i.e copies to GOBIN"
 	@echo "  clean:            clean the build directory and Rust target"
 	@echo "  lint:             run golangci-lint"
+	@echo "  test:             run FL unit and integration tests"
+	@echo "  test-all:         run all tests (Go and Rust)"
 	@echo "  dockers:          build and push all Docker images (Go and Rust services)"
 	@echo "  dockers_dev:      build all Go service dev Docker images"
 	@echo "  dockers_rust:     build all Rust service Docker images"

README.md

@@ -11,6 +11,7 @@
 - 🔧 **WAMR on Zephyr RTOS**: Deploy lightweight Wasm workloads on constrained devices running Zephyr RTOS via the WebAssembly Micro Runtime (WAMR).
 - 🛠️ **Powerful Service Mesh**: Integrates with **[SuperMQ](https://github.com/absmach)** for secure, efficient IoT device communication.
 - 🔒 **Security at the Core**: Propeller ensures secure workload execution and communication for IoT environments.
+- 🤖 **Federated Learning**: Built-in support for federated machine learning workflows with FedAvg aggregation, enabling privacy-preserving distributed training across edge devices.
 
 ## 🛠️ How It Works
 
@@ -32,6 +33,13 @@ For setup instructions, API references, and usage examples, see the documentatio
 - 🛡️ **Secure Workloads**: Run isolated, portable workloads securely on cloud or edge devices.
 - 🌎 **Smart Cities**: Power scalable IoT networks with efficient communication and dynamic workloads.
 - ☁️ **Serverless Applications**: Deploy FaaS applications leveraging Propeller's Wasm orchestration capabilities.
+- 🧠 **Federated Machine Learning**: Train machine learning models across distributed edge devices without exposing raw data, perfect for privacy-sensitive applications.
+
+### Architecture Notes
+
+- **Rust Proplet Only**: Propeller now uses only the Rust proplet implementation (Wasmtime runtime) for executing FL workloads
+- **MQTT Communication**: FL coordination uses MQTT topics under `m/{domain}/c/{channel}/control/...`
+- **Chunked Transport**: Large model artifacts are automatically chunked for efficient MQTT transport
 
 ## 🤝 Contributing
 

cli/fl.go

@@ -0,0 +1,104 @@
+package cli
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/spf13/cobra"
+)
+
+var flCmd = []cobra.Command{
+	{
+		Use:   "round-start",
+		Short: "Start a federated learning round",
+		Long:  `Publish a round start message to trigger FL training. This is a sample FML application command.`,
+		Run: func(cmd *cobra.Command, args []string) {
+			roundID, _ := cmd.Flags().GetString("round-id")
+			modelURI, _ := cmd.Flags().GetString("model-uri")
+			taskWasmImage, _ := cmd.Flags().GetString("task-wasm-image")
+			participants, _ := cmd.Flags().GetStringSlice("participants")
+			kOfN, _ := cmd.Flags().GetInt("k-of-n")
+			timeoutS, _ := cmd.Flags().GetInt("timeout-s")
+
+			// Build hyperparams
+			hyperparams := make(map[string]interface{})
+			if epochs, _ := cmd.Flags().GetInt("epochs"); epochs > 0 {
+				hyperparams["epochs"] = epochs
+			}
+			if lr, _ := cmd.Flags().GetFloat64("learning-rate"); lr > 0 {
+				hyperparams["lr"] = lr
+			}
+			if batchSize, _ := cmd.Flags().GetInt("batch-size"); batchSize > 0 {
+				hyperparams["batch_size"] = batchSize
+			}
+
+			// Validate required fields
+			if roundID == "" {
+				logErrorCmd(*cmd, fmt.Errorf("round-id is required"))
+				return
+			}
+			if modelURI == "" {
+				logErrorCmd(*cmd, fmt.Errorf("model-uri is required"))
+				return
+			}
+			if taskWasmImage == "" {
+				logErrorCmd(*cmd, fmt.Errorf("task-wasm-image is required"))
+				return
+			}
+			if len(participants) == 0 {
+				logErrorCmd(*cmd, fmt.Errorf("at least one participant is required"))
+				return
+			}
+
+			// Build round start message
+			roundStart := map[string]interface{}{
+				"round_id":        roundID,
+				"model_uri":       modelURI,
+				"task_wasm_image": taskWasmImage,
+				"participants":    participants,
+				"k_of_n":          kOfN,
+				"timeout_s":       timeoutS,
+			}
+
+			if len(hyperparams) > 0 {
+				roundStart["hyperparams"] = hyperparams
+			}
+
+			roundStartJSON, err := json.MarshalIndent(roundStart, "", "  ")
+			if err != nil {
+				logErrorCmd(*cmd, err)
+				return
+			}
+
+			fmt.Println("Round start message (publish to fl/rounds/start):")
+			fmt.Println(string(roundStartJSON))
+			fmt.Println("\nNote: This is a sample FML application. Use an MQTT client to publish this message.")
+		},
+	},
+}
+
+func NewFLCmd() *cobra.Command {
+	cmd := cobra.Command{
+		Use:   "fl",
+		Short: "Federated Learning (sample FML application)",
+		Long:  `Sample FML application commands. FL is implemented as an external application, not a core Propeller feature.`,
+	}
+
+	for i := range flCmd {
+		cmd.AddCommand(&flCmd[i])
+	}
+
+	// Flags for round-start command
+	roundStartCmd := &flCmd[0]
+	roundStartCmd.Flags().StringP("round-id", "r", "", "Round identifier (required)")
+	roundStartCmd.Flags().StringP("model-uri", "m", "", "Model URI (required)")
+	roundStartCmd.Flags().StringP("task-wasm-image", "i", "", "Task Wasm image OCI ref (required)")
+	roundStartCmd.Flags().StringSliceP("participants", "p", []string{}, "List of proplet IDs (required)")
+	roundStartCmd.Flags().IntP("k-of-n", "k", 3, "Minimum participants required for aggregation")
+	roundStartCmd.Flags().IntP("timeout-s", "t", 30, "Round timeout in seconds")
+	roundStartCmd.Flags().IntP("epochs", "e", 1, "Local training epochs")
+	roundStartCmd.Flags().Float64P("learning-rate", "l", 0.01, "Learning rate")
+	roundStartCmd.Flags().IntP("batch-size", "b", 16, "Batch size")
+
+	return &cmd
+}

cmd/cli/main.go

@@ -50,8 +50,9 @@ func main() {
 
 	tasksCmd := cli.NewTasksCmd()
 	provisionCmd := cli.NewProvisionCmd()
+	flCmd := cli.NewFLCmd()
 
-	rootCmd.AddCommand(tasksCmd, provisionCmd)
+	rootCmd.AddCommand(tasksCmd, provisionCmd, flCmd)
 
 	rootCmd.PersistentFlags().StringVarP(
 		&managerURL,

docker/Dockerfile.proplet

@@ -8,6 +8,9 @@ WORKDIR /build
 
 RUN apk add --no-cache \
     musl-dev \
+    openssl-dev \
+    openssl-libs-static \
+    pkgconfig \
     upx
 
 COPY proplet/ ./

embed-proplet/prj.conf

@@ -39,8 +39,11 @@ CONFIG_NET_MAX_CONTEXTS=10
 
 # MQTT Client
 CONFIG_MQTT_LIB=y
-# Enable Sockets (used by MQTT lib)
+# Enable Sockets (used by MQTT lib and HTTP client)
 CONFIG_NET_SOCKETS=y
+# HTTP Client support
+CONFIG_HTTP_CLIENT=y
+CONFIG_HTTP_CLIENT_BUFFER_SIZE=4096
 
 # Uncomment to enable TLS support for socket MQTT Library.
 # CONFIG_MQTT_LIB_TLS=y