FuseRule ⚡

FuseRule is a high-performance, developer-first rule engine built for the cloud-native ecosystem. It leverages Apache Arrow and DataFusion to provide a lightning-fast, SQL-expressive core for real-time data auditing and event processing.

Designed as an Infrastructure Primitive, FuseRule decouples its deterministic core from pluggable "edges" like persistence, evaluation engines, and notification agents.

🚀 Features

⚡ Arrow-Native: Zero-copy columnar data processing for maximum performance
🔍 SQL-Powered Rules: Write complex predicates using standard SQL expressions
🏗️ Infrastructure-First: Decoupled core with swappable traits for StateStore, RuleEvaluator, and Agent
📊 Real-Time Observability: Returns machine-readable EvaluationTrace logs for every ingestion
🔄 Zero-Downtime Reloading: Hot-swap rules and agents via SIGHUP without restarting the daemon
📈 Cloud-Native Metrics: Built-in Prometheus-formatted telemetry
🔔 Stateful Transitions: Built-in state management for Activated and Deactivated transitions
🪟 Time Windows: Sliding time windows for aggregate functions (AVG, COUNT, SUM)
🔐 API Key Authentication: Secure endpoints with configurable API keys
🚦 Rate Limiting: Built-in rate limiting for ingestion endpoints
📡 Multiple Ingestion Sources: HTTP, Kafka, and WebSocket support
🎨 Action Templates: Handlebars templating for custom webhook payloads
🐛 Interactive Debugging: REPL and step-through debugger for rule development

📦 Installation

As a Library

Add to your Cargo.toml:

[dependencies]
fuse-rule = "0.1.0"

As a Binary

cargo install fuse-rule

🎯 Quickstart

1. Create a Configuration File

Create fuse_rule_config.yaml:

engine:
  persistence_path: "fuserule_state"
  ingest_rate_limit: 1000  # requests per second
  api_keys:
    - "sk_live_abc123..."

schema:
  - name: "price"
    data_type: "float64"
  - name: "symbol"
    data_type: "utf8"
  - name: "volume"
    data_type: "int32"

agents:
  - name: "logger"
    type: "logger"
  - name: "slack-webhook"
    type: "webhook"
    url: "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
    template: |
      {
        "text": "🚨 {{rule_name}} triggered!",
        "symbol": "{{matched_data.0.symbol}}",
        "price": "{{matched_data.0.price}}"
      }

rules:
  - id: "high_price_alert"
    name: "High Price Alert"
    predicate: "price > 1000"
    action: "slack-webhook"
    version: 1
    enabled: true
    state_ttl_seconds: 3600  # Expire state after 1 hour

  - id: "volume_spike"
    name: "Volume Spike"
    predicate: "AVG(volume) > 10000"
    action: "logger"
    window_seconds: 60  # 60-second sliding window
    version: 1
    enabled: true

2. Start the Server

fuserule run --config fuse_rule_config.yaml --port 3030

3. Ingest Data

curl -X POST http://localhost:3030/ingest \
     -H "Content-Type: application/json" \
     -H "X-API-Key: sk_live_abc123..." \
     -d '[{"price": 1500, "symbol": "AAPL", "volume": 5000}]'

4. Check Rule States

# Get all rule states
curl http://localhost:3030/api/v1/state \
     -H "X-API-Key: sk_live_abc123..."

# Get specific rule state
curl http://localhost:3030/api/v1/state/high_price_alert \
     -H "X-API-Key: sk_live_abc123..."

📚 Usage as a Library

Basic Example

use arrow_rule_agent::{RuleEngine, config::FuseRuleConfig};
use arrow::array::{Float64Array, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use std::sync::Arc;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Load configuration
    let config = FuseRuleConfig::from_file("fuse_rule_config.yaml")?;
    
    // Create engine from config
    let mut engine = RuleEngine::from_config(config).await?;
    
    // Create a test batch
    let schema = Schema::new(vec![
        Field::new("price", DataType::Float64, true),
        Field::new("symbol", DataType::Utf8, true),
    ]);
    
    let price_array = Arc::new(Float64Array::from(vec![1500.0, 500.0]));
    let symbol_array = Arc::new(StringArray::from(vec!["AAPL", "GOOGL"]));
    
    let batch = RecordBatch::try_new(
        Arc::new(schema),
        vec![price_array, symbol_array],
    )?;
    
    // Process batch and get evaluation traces
    let traces = engine.process_batch(&batch).await?;
    
    for trace in traces {
        if trace.action_fired {
            println!("Rule '{}' activated!", trace.rule_name);
        }
    }
    
    Ok(())
}

Programmatic Rule Management

use arrow_rule_agent::{RuleEngine, rule::Rule};

// Add a rule programmatically
let rule = Rule {
    id: "custom_rule".to_string(),
    name: "Custom Rule".to_string(),
    predicate: "price > 100 AND volume < 50".to_string(),
    action: "logger".to_string(),
    window_seconds: None,
    version: 1,
    enabled: true,
};

engine.add_rule(rule).await?;

// Update a rule
engine.update_rule("custom_rule", updated_rule).await?;

// Toggle rule
engine.toggle_rule("custom_rule", false).await?;

🏗️ Architecture

┌─────────────────────────────────────────────────────────────┐
│                      FuseRule Engine                         │
├─────────────────────────────────────────────────────────────┤
│                                                               │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐ │
│  │   Ingest     │───▶│   Evaluate   │───▶│   Activate   │ │
│  │   Sources    │    │   Rules      │    │   Agents     │ │
│  └──────────────┘    └──────────────┘    └──────────────┘ │
│         │                   │                    │           │
│         │                   │                    │           │
│         ▼                   ▼                    ▼           │
│  ┌──────────────────────────────────────────────────────┐   │
│  │          Arrow RecordBatch (Zero-Copy)              │   │
│  └──────────────────────────────────────────────────────┘   │
│                                                               │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐ │
│  │   State      │    │   Windows     │    │   Metrics    │ │
│  │   Store      │    │   Buffers     │    │   (Prom)     │ │
│  └──────────────┘    └──────────────┘    └──────────────┘ │
└─────────────────────────────────────────────────────────────┘

Design Philosophy: "Hard Core, Soft Edges"

FuseRule is built on the philosophy that the core logic of a rule engine should be a "boring," deterministic primitive, while the integration points (Ingress, Persistence, Notifications) should be flexible and pluggable.

Core (Hard):

Rule evaluation logic
State transitions
Window management
Metrics collection

Edges (Soft):

StateStore trait (Sled, Redis, etc.)
RuleEvaluator trait (DataFusion, custom SQL engines)
Agent trait (Webhooks, Loggers, custom actions)
Ingestion sources (HTTP, Kafka, WebSocket)

🔧 CLI Commands

Run Server

fuserule run --config fuse_rule_config.yaml --port 3030

Validate Rules

# Validate all rules in config
fuserule validate --config fuse_rule_config.yaml

# Validate specific predicate
fuserule validate --config fuse_rule_config.yaml --predicate "price > 100"

Interactive REPL

fuserule repl --config fuse_rule_config.yaml

Rule Debugger

fuserule debug --config fuse_rule_config.yaml

📡 API Endpoints

Public Endpoints

GET /status - Server status
GET /health - Health check with engine stats
GET /metrics - Prometheus metrics

Protected Endpoints (Require `X-API-Key` header)

Rule Management

GET /rules - List all rules
POST /api/v1/rules - Create new rule
PUT /api/v1/rules/:id - Update rule
PATCH /api/v1/rules/:id - Partial update (e.g., enable/disable)
DELETE /api/v1/rules/:id - Delete rule
POST /api/v1/rules/validate - Validate rule predicate

State Management

GET /api/v1/state - Get all rule states
GET /api/v1/state/:rule_id - Get specific rule state

Data Ingestion

POST /ingest - Ingest JSON data (rate-limited)

📊 Monitoring

FuseRule exposes Prometheus metrics at /metrics:

fuserule_batches_processed_total - Total batches ingested
fuserule_activations_total - Total rule activations
fuserule_agent_failures_total - Total agent failures
fuserule_evaluation_duration_seconds - Evaluation latency histogram
fuserule_rule_evaluations_total{rule_id} - Per-rule evaluation count
fuserule_rule_activations_total{rule_id} - Per-rule activation count

🔌 Ingestion Sources

HTTP (Default)

curl -X POST http://localhost:3030/ingest \
     -H "Content-Type: application/json" \
     -d '[{"price": 150, "symbol": "AAPL"}]'

Kafka

Configure in fuse_rule_config.yaml:

sources:
  - type: "kafka"
    brokers: ["localhost:9092"]
    topic: "events"
    group_id: "fuserule"
    auto_commit: true

WebSocket

Configure in fuse_rule_config.yaml:

sources:
  - type: "websocket"
    bind: "0.0.0.0:3031"
    max_connections: 1000

Connect and send JSON:

const ws = new WebSocket('ws://localhost:3031/ws');
ws.send(JSON.stringify([{"price": 150, "symbol": "AAPL"}]));

🎨 Action Templates

Use Handlebars templates for custom webhook payloads:

agents:
  - name: "custom-webhook"
    type: "webhook"
    url: "https://api.example.com/webhook"
    template: |
      {
        "alert": "{{rule_name}}",
        "timestamp": "{{timestamp}}",
        "data": {{#each matched_data}}
          {
            "price": {{price}},
            "symbol": "{{symbol}}"
          }{{#unless @last}},{{/unless}}
        {{/each}},
        "count": {{count}}
      }

🧪 Testing

Unit Tests

cargo test --test unit_test

Integration Tests

cargo test --test integration_test

Property-Based Tests

cargo test --test property_test

📖 Documentation

Architecture Guide - Deep dive into design
API Documentation - Full API reference (when published)
Local API Documentation: Generate and view locally with:
```
cargo doc --no-deps --open
```
This will build and open the documentation in your browser at target/doc/arrow_rule_agent/index.html
Examples - Code examples

🤝 Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

📜 License

Licensed under the Apache License, Version 2.0. See LICENSE for details.

🙏 Acknowledgments

Built on Apache Arrow for zero-copy data processing
Powered by DataFusion for SQL evaluation
Inspired by the "Hard Core, Soft Edges" philosophy

Name		Name	Last commit message	Last commit date
Latest commit History 24 Commits
.github/workflows		.github/workflows
docs		docs
examples		examples
src		src
test_state		test_state
test_state2		test_state2
tests		tests
.fuserule_history		.fuserule_history
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
CONTRIBUTING.md		CONTRIBUTING.md
Cargo.lock		Cargo.lock
Cargo.toml		Cargo.toml
Dockerfile		Dockerfile
LICENSE		LICENSE
README.md		README.md
RELEASE.md		RELEASE.md
docker-compose.yaml		docker-compose.yaml
fuse_rule_config.yaml		fuse_rule_config.yaml

License

hamzzy/FuseRule

Folders and files

Latest commit

History

Repository files navigation