Develop WebAssembly (WASM) modules and graph definitions for data flow graphs (preview)

• Rust toolchain: Install with:

  curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
  

• WASM target: Add with:

  rustup target add wasm32-wasip2
  

• Build tools: Install with:

  cargo install wasm-tools --version '=1.201.0' --locked
  

• Python 3.8 or later
• componentize-py: Install with:

  pip install "componentize-py==0.14"
  

The WASM Rust SDK is available through a custom Azure DevOps registry. Configure access by setting these environment variables:

export CARGO_REGISTRIES_AZURE_VSCODE_TINYKUBE_INDEX="sparse+https://pkgs.dev.azure.com/azure-iot-sdks/iot-operations/_packaging/preview/Cargo/index/"
export CARGO_NET_GIT_FETCH_WITH_CLI=true

Add the following environment variables to your shell profile for persistent access:

echo 'export CARGO_REGISTRIES_AZURE_VSCODE_TINYKUBE_INDEX="sparse+https://pkgs.dev.azure.com/azure-iot-sdks/iot-operations/_packaging/preview/Cargo/index/"' >> ~/.bashrc
echo 'export CARGO_NET_GIT_FETCH_WITH_CLI=true' >> ~/.bashrc
source ~/.bashrc

Python development uses componentize-py with WebAssembly Interface Types (WIT) for code generation. The WIT schemas define the interfaces between your Python code and the WASM runtime.

Get the WIT schemas: The required schemas are available in the Azure IoT Operations samples repository. Clone or download these schemas to your development environment:

# Clone the repository to access WIT schemas
git clone https://github.com/Azure-Samples/explore-iot-operations.git

The schemas are located at explore-iot-operations/samples/wasm/python/schema/ and include interface definitions for all supported operator types (map, filter, branch, etc.).

You don't need any other environment configuration beyond installing the prerequisites and obtaining the WIT schemas.

cargo new --lib temperature-converter
cd temperature-converter

Configure Cargo.toml

Edit the Cargo.toml file to include dependencies for the WASM SDK and other libraries:

[package]
name = "temperature-converter"
version = "0.1.0"
edition = "2021"

[dependencies]
# WebAssembly Interface Types (WIT) code generation
wit-bindgen = "0.22"

# Azure IoT Operations WASM SDK - provides operator macros and host APIs
tinykube_wasm_sdk = { version = "0.2.0", registry = "azure-vscode-tinykube" }

# JSON serialization/deserialization for data processing
serde = { version = "1", default-features = false, features = ["derive"] }
serde_json = { version = "1", default-features = false, features = ["alloc"] }

[lib]
# Required for WASM module compilation
crate-type = ["cdylib"]

Key dependencies explained:

• wit-bindgen: Generates Rust bindings from WebAssembly Interface Types (WIT) definitions, enabling your code to interface with the WASM runtime
• tinykube_wasm_sdk: Azure IoT Operations SDK providing operator macros (#[map_operator], #[filter_operator], etc.) and host APIs for logging, metrics, and state management
• serde + serde_json: JSON processing libraries for parsing and generating data payloads; default-features = false optimizes for WASM size constraints
• crate-type = ["cdylib"]: Compiles the Rust library as a C-compatible dynamic library, which is required for WASM module generation

Create a Python file for your operator. The filename should match your module name:

# Create your operator file
touch temperature_converter.py

Python WASM modules don't require other project configuration files. The Python class that implements the operator interface defines the module structure.

// src/lib.rs
use tinykube_wasm_sdk::logger::{self, Level};
use tinykube_wasm_sdk::macros::map_operator;
use serde_json::{json, Value};

// Import the generated types from wit-bindgen
use crate::tinykube_graph::processor::types::{DataModel, ModuleConfiguration, BufferOrBytes};

fn temperature_converter_init(_configuration: ModuleConfiguration) -> bool {
    logger::log(Level::Info, "temperature-converter", "Init invoked");
    true
}

#[map_operator(init = "temperature_converter_init")]
fn temperature_converter(input: DataModel) -> DataModel {
    let DataModel::Message(mut result) = input else {
        return input;
    };

    let payload = &result.payload.read();
    if let Ok(data_str) = std::str::from_utf8(payload) {
        if let Ok(mut data) = serde_json::from_str::<Value>(data_str) {
            if let Some(temp) = data["value"]["temperature"].as_f64() {
                let fahrenheit = (temp * 9.0 / 5.0) + 32.0;
                data["value"] = json!({
                    "temperature_fahrenheit": fahrenheit,
                    "original_celsius": temp
                });
                
                if let Ok(output_str) = serde_json::to_string(&data) {
                    result.payload = BufferOrBytes::Bytes(output_str.into_bytes());
                }
            }
        }
    }

    DataModel::Message(result)
}

# temperature_converter.py
import json
from map_impl import exports
from map_impl import imports
from map_impl.imports import types

class Map(exports.Map):
    def init(self, configuration) -> bool:
        imports.logger.log(imports.logger.Level.INFO, "temperature-converter", "Init invoked")
        return True

    def process(self, message: types.DataModel) -> types.DataModel:
        # Ensure the input is of the expected type  
        if not isinstance(message, types.DataModel_Message):
            imports.logger.log(imports.logger.Level.ERROR, "temperature-converter", "Unexpected input type")
            return message

        # Extract and decode the payload
        buffer = message.value.payload.value
        payload = buffer.read()
        data_str = payload.decode('utf-8')
        
        try:
            data = json.loads(data_str) 
            # Process temperature conversion logic
            if 'value' in data and 'temperature' in data['value']:
                celsius = float(data['value']['temperature'])
                fahrenheit = (celsius * 9/5) + 32
                
                output = {
                    'value': {
                        'temperature_fahrenheit': fahrenheit,
                        'original_celsius': celsius
                    }
                }
                
                output_str = json.dumps(output)
                output_bytes = output_str.encode('utf-8')
                
                # Update the message payload
                message.value.payload = types.BufferOrBytes_Bytes(value=output_bytes)
            
            return message  # Return the modified message
            
        except Exception as e:
            imports.logger.log(imports.logger.Level.ERROR, "temperature-converter", f"Error: {e}")
            return message

# Build WASM module
cargo build --release --target wasm32-wasip2

# Find your module  
ls target/wasm32-wasip2/release/*.wasm

# Navigate to the WIT schemas directory (from the Configure development environment section)
cd explore-iot-operations/samples/wasm/python/schema

# Generate Python bindings from schema
componentize-py -d ./schema/ -w map-impl bindings ./

# Build WASM module
componentize-py -d ./schema/ -w map-impl componentize temperature_converter -o temperature_converter.wasm

# Verify build
file temperature_converter.wasm  # Should show: WebAssembly (wasm) binary module

The Rust Docker builder is maintained in the Azure IoT Operations samples repository and includes all necessary dependencies. For detailed documentation, see Rust Docker builder usage.

# Build release version (optimized for production)
docker run --rm -v "$(pwd):/workspace" ghcr.io/azure-samples/explore-iot-operations/rust-wasm-builder --app-name temperature-converter

# Build debug version (includes debugging symbols and less optimization)
docker run --rm -v "$(pwd):/workspace" ghcr.io/azure-samples/explore-iot-operations/rust-wasm-builder --app-name temperature-converter --build-mode debug

Docker build options:

• --app-name: Must match your Rust crate name from Cargo.toml
• --build-mode: Choose release (default) for optimized builds or debug for development builds with symbols

The Python Docker builder is maintained in the Azure IoT Operations samples repository and includes all necessary dependencies and schemas. For detailed documentation, see Python Docker builder usage.

# Build release version (optimized for production)
docker run --rm -v "$(pwd):/workspace" ghcr.io/azure-samples/explore-iot-operations/python-wasm-builder --app-name temperature_converter --app-type map

# Build debug version (includes debugging symbols and less optimization)
docker run --rm -v "$(pwd):/workspace" ghcr.io/azure-samples/explore-iot-operations/python-wasm-builder --app-name temperature_converter --app-type map --build-mode debug

Docker build options:

• --app-name: Must match your Python filename without the .py extension
• --app-type: Specify the operator type (map, filter, branch, etc.)
• --build-mode: Choose release (default) for optimized builds or debug for development builds with symbols

For comprehensive examples, see the Rust examples in the samples repository. Complete implementations include:

• Map operators: Data transformation and conversion logic
• Filter operators: Conditional data processing and validation
• Branch operators: Multi-path routing based on data content
• Accumulate operators: Time-windowed aggregation and statistical processing
• Delay operators: Time-based processing control

The examples demonstrate working implementations that show the complete structure for each operator type, including proper error handling and logging patterns.

For comprehensive examples, see the Python examples in the samples repository. Complete implementations include:

• Map operators: Data transformation and conversion logic
• Filter operators: Conditional data processing and validation
• Branch operators: Multi-path routing based on data content

The WASM Rust SDK provides comprehensive development tools:

Operator macros

use tinykube_wasm_sdk::macros::{map_operator, filter_operator, branch_operator};
use tinykube_wasm_sdk::{DataModel, HybridLogicalClock};

// Map operator - transforms each data item
#[map_operator(init = "my_init_function")]
fn my_map(input: DataModel) -> DataModel {
    // Transform logic here
}

// Filter operator - allows/rejects data based on predicate  
#[filter_operator(init = "my_init_function")]
fn my_filter(input: DataModel) -> bool {
    // Return true to pass data through, false to filter out
}

// Branch operator - routes data to different arms
#[branch_operator(init = "my_init_function")]
fn my_branch(input: DataModel, timestamp: HybridLogicalClock) -> bool {
    // Return true for "True" arm, false for "False" arm
}

Module configuration parameters

Your WASM operators can receive runtime configuration parameters through the ModuleConfiguration struct passed to the init function. These parameters are defined in the graph definition and allow runtime customization without rebuilding modules.

use tinykube_wasm_sdk::logger::{self, Level};
use tinykube_wasm_sdk::ModuleConfiguration;

fn my_operator_init(configuration: ModuleConfiguration) -> bool {
    // Access required parameters
    if let Some(threshold_param) = configuration.parameters.get("temperature_threshold") {
        let threshold: f64 = threshold_param.parse().unwrap_or(25.0);
        logger::log(Level::Info, "my-operator", &format!("Using threshold: {}", threshold));
    }
    
    // Access optional parameters with defaults
    let unit = configuration.parameters
        .get("output_unit")
        .map(|s| s.as_str())
        .unwrap_or("celsius");
    
    logger::log(Level::Info, "my-operator", &format!("Output unit: {}", unit));
    true
}

For detailed information about defining configuration parameters in graph definitions, see Module configuration parameters.

Host APIs

Use the SDK to work with distributed services:

State store for persistent data:

use tinykube_wasm_sdk::state_store;

// Set value
state_store::set(key.as_bytes(), value.as_bytes(), None, None, options)?;

// Get value  
let response = state_store::get(key.as_bytes(), None)?;

// Delete key
state_store::del(key.as_bytes(), None, None)?;

Structured logging:

use tinykube_wasm_sdk::logger::{self, Level};

logger::log(Level::Info, "my-operator", "Processing started");
logger::log(Level::Error, "my-operator", &format!("Error: {}", error));

OpenTelemetry-compatible metrics:

use tinykube_wasm_sdk::metrics;

// Increment counter
metrics::add_to_counter("requests_total", 1.0, Some(labels))?;

// Record histogram value
metrics::record_to_histogram("processing_duration", duration_ms, Some(labels))?;

Python WASM development doesn't use a traditional SDK. Instead, you use generated bindings from WebAssembly Interface Types (WIT). The WIT schemas are available from the Azure IoT Operations samples repository.

These bindings give you:

Typed interfaces for operators:

from map_impl import exports, imports
from map_impl.imports import types

# Implement the operator interface
class Map(exports.Map):
    def init(self, configuration) -> bool:
        # Access configuration parameters
        threshold = configuration.get_parameter("temperature_threshold")
        unit = configuration.get_parameter("output_unit", default="celsius")
        
        imports.logger.log(imports.logger.Level.INFO, "my-operator", 
                          f"Initialized with threshold={threshold}, unit={unit}")
        return True
    
    def process(self, message: types.DataModel) -> types.DataModel:
        # Your processing logic here
        return message

For detailed information about defining configuration parameters in graph definitions, see Module configuration parameters.

Logging through imports:

# Access to structured logging
imports.logger.log(imports.logger.Level.INFO, "my-operator", "Processing started")
imports.logger.log(imports.logger.Level.ERROR, "my-operator", f"Error: {error}")

Error handling:

try:
    # Processing logic
    pass
except Exception as e:
    imports.logger.log(imports.logger.Level.ERROR, "my-operator", f"Error: {e}")
    return message  # Return original message on error