多租户部署实战

本教程演示如何在单一 EnerOS 集群中承载多个电力运营商租户，实现拓扑、数据与配额的物理隔离。

多租户架构概览

EnerOS 的多租户体系由 eneros-tenant crate 提供，核心抽象为「租户（Tenant）— 成员（Member）— 配额（Quota）— 用量（Usage）— API Key 绑定」五元组。所有租户元数据持久化于 SQLite（SqliteTenantStorage），上层服务通过 TenantManager 与 TenantUsageTracker 两个对象完成业务操作。

租户状态机：

准备工作

• 已部署 EnerOS 集群（3 节点起，参考 高可用集群部署）
• 已完成内部 CA 与 mTLS 证书签发流程
• SQLite 3.34+（集群使用共享块存储挂载 /var/lib/eneros/tenant.db，或 Raft 复制引擎代理）
• 已安装 eneros-cli，且具备 tenant:admin 角色

步骤 1：规划租户

为每个租户分配唯一 tenant_id、独立网络命名空间与证书 SAN。建议采用「业务层级 + 短代码」命名，避免与 UUID v4 主键冲突。

配额等级参考表：

步骤 2：配置租户隔离

在 eneros.toml 中声明租户：

# /etc/eneros/eneros.toml
[tenant]
db_path = "/var/lib/eneros/tenant.db"
default_quota = { max_devices = 1000, max_agents = 50, max_api_calls_per_day = 100000, max_storage_mb = 10240 }

[[tenants]]
id = "gs"
display_name = "国调中心"
network = "gs-net"
database = "eneros_gs"
cert_san = ["*.gs.eneros.internal"]
quota = { max_devices = 10000, max_agents = 200, max_api_calls_per_day = 1000000, max_storage_mb = 102400 }

[[tenants]]
id = "ps"
display_name = "省调中心"
network = "ps-net"
database = "eneros_ps"
cert_san = ["*.ps.eneros.internal"]
quota = { max_devices = 2000, max_agents = 50, max_api_calls_per_day = 200000, max_storage_mb = 20480 }

[[tenants]]
id = "ds"
display_name = "地调中心"
network = "ds-net"
database = "eneros_ds"
cert_san = ["*.ds.eneros.internal"]
quota = { max_devices = 500, max_agents = 20, max_api_calls_per_day = 50000, max_storage_mb = 5120 }

字段说明：

步骤 3：使用 TenantManager 创建租户

通过 TenantManager 编程化创建租户。create_tenant 会自动生成 UUID v4 作为主键，并以 Active 状态初始化全 0 用量行。

use std::sync::Arc;
use eneros_tenant::{
    SqliteTenantStorage, TenantManager, TenantQuota, TenantStatus,
};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // 1. 打开 SQLite 存储（自动建表 + WAL 模式 + foreign_keys = ON）
    let storage = Arc::new(SqliteTenantStorage::new("/var/lib/eneros/tenant.db")?);

    // 2. 构造 Manager
    let manager = TenantManager::new(storage.clone());

    // 3. 幂等确保 default 租户存在（首次部署调用）
    let default = manager.ensure_default_tenant()?;
    println!("default tenant: id={}, status={:?}", default.id, default.status);

    // 4. 创建国调租户（L3 配额）
    let gs_quota = TenantQuota {
        max_devices: 10_000,
        max_agents: 200,
        max_api_calls_per_day: 1_000_000,
        max_storage_mb: 102_400,
    };
    let gs = manager.create_tenant("国调中心", gs_quota)?;
    println!("created tenant: id={}, name={}", gs.id, gs.name);

    // 5. 创建省调租户（L2 配额）
    let ps_quota = TenantQuota {
        max_devices: 2_000,
        max_agents: 50,
        max_api_calls_per_day: 200_000,
        max_storage_mb: 20_480,
    };
    let ps = manager.create_tenant("省调中心", ps_quota)?;
    println!("created tenant: id={}, name={}", ps.id, ps.name);

    // 6. 列出所有活跃租户
    let active = manager.list_tenants(false)?;
    println!("active tenants: {}", active.len());
    for t in &active {
        println!("  - {} ({}) quotas={:?}", t.name, t.id, t.quotas);
    }

    Ok(())
}

输出示例：

default tenant: id=default, status=Active
created tenant: id=8a7b3c2d-1e2f-4a3b-9c8d-7e6f5a4b3c2d, name=国调中心
created tenant: id=2d3e4f5a-6b7c-8d9e-0f1a-2b3c4d5e6f7a, name=省调中心
active tenants: 3
  - Default Tenant (default) quotas=TenantQuota { max_devices: 1000, max_agents: 50, max_api_calls_per_day: 100000, max_storage_mb: 10240 }
  - 国调中心 (8a7b3c2d-1e2f-4a3b-9c8d-7e6f5a4b3c2d) quotas=TenantQuota { max_devices: 10000, max_agents: 200, max_api_calls_per_day: 1000000, max_storage_mb: 102400 }
  - 省调中心 (2d3e4f5a-6b7c-8d9e-0f1a-2b3c4d5e6f7a) quotas=TenantQuota { max_devices: 2000, max_agents: 50, max_api_calls_per_day: 200000, max_storage_mb: 20480 }

注意：create_tenant 自动生成的 UUID 与 eneros.toml 中的短代码（如 gs）是两套标识。生产环境建议在创建后通过 update_tenant 回写 name 字段附加短代码前缀，或在 default 租户的固定 ID 之外显式调用存储层 create_tenant 写入自定义 ID。

步骤 4：签发租户证书

每张证书在 SAN 中绑定 tenant_id，由 mTLS 握手时强制校验。证书 CN 采用 <service>.<tenant>.eneros.internal 三段式命名：

use eneros_security::{CertificateAuthority, CertificateSpec};

async fn issue_tenant_certs(ca: &CertificateAuthority) -> anyhow::Result<()> {
    // 国调 Agent Runtime 证书
    let gs_cert = ca.issue(
        CertificateSpec::new("agent-runtime.gs.eneros.internal")
            .tenant("gs")
            .sans(vec![
                "agent-runtime.gs.eneros.internal",
                "10.0.0.11",  // IP SAN
            ])
            .role("agent-runtime")
    ).await?;
    println!("issued gs cert: serial={}", gs_cert.serial);

    // 省调 Agent Runtime 证书
    let ps_cert = ca.issue(
        CertificateSpec::new("agent-runtime.ps.eneros.internal")
            .tenant("ps")
            .sans(vec!["agent-runtime.ps.eneros.internal"])
            .role("agent-runtime")
    ).await?;
    println!("issued ps cert: serial={}", ps_cert.serial);

    // 省调 SCADA Gateway 证书
    let gw_cert = ca.issue(
        CertificateSpec::new("scada-gw.ps.eneros.internal")
            .tenant("ps")
            .sans(vec!["scada-gw.ps.eneros.internal"])
            .role("scada-gateway")
    ).await?;
    println!("issued ps-gw cert: serial={}", gw_cert.serial);

    Ok(())
}

证书字段约定：

mTLS 握手时 Gateway 会提取证书中的 tenant_id OID 并与请求路径中的租户前缀比对，不一致直接拒绝。

步骤 5：成员管理与角色绑定

每个租户独立维护成员列表，成员角色决定其在租户内的权限。

use eneros_tenant::TenantManager;

fn setup_members(manager: &TenantManager, gs_id: &str, ps_id: &str) -> anyhow::Result<()> {
    // 国调成员
    manager.add_member(gs_id, "alice@gs.com", "admin")?;
    manager.add_member(gs_id, "bob@gs.com", "operator")?;
    manager.add_member(gs_id, "carol@grid-vendor.com", "liaison")?; // 厂家联络人

    // 省调成员
    manager.add_member(ps_id, "dave@ps.com", "admin")?;
    manager.add_member(ps_id, "eve@ps.com", "operator")?;
    manager.add_member(ps_id, "frank@ps.com", "viewer")?;

    // 校验成员存在性
    assert!(manager.is_member(gs_id, "alice@gs.com")?);
    assert!(!manager.is_member(ps_id, "alice@gs.com")?); // 跨租户不可见

    // 列出成员
    let gs_members = manager.list_members(gs_id)?;
    println!("gs members ({}):", gs_members.len());
    for m in &gs_members {
        println!("  - {} (role={}, added={})",
            m.user_id, m.role, m.added_at);
    }

    Ok(())
}

角色权限矩阵：

步骤 6：配额管理与用量跟踪

TenantUsageTracker 提供原子计数器，所有 read-modify-write 操作在 RwLock 写锁保护下完成，保证并发安全。

use std::sync::Arc;
use eneros_tenant::{
    QuotaResource, SqliteTenantStorage, TenantManager,
    TenantQuota, TenantUsageTracker,
};
use eneros_tenant::usage_tracker::spawn_daily_reset_task;

fn setup_quota_and_tracker(
    manager: &TenantManager,
    storage: Arc<SqliteTenantStorage>,
    gs_id: &str,
) -> anyhow::Result<Arc<TenantUsageTracker>> {
    // 1. 调整配额（如需扩容）
    let new_quota = TenantQuota {
        max_devices: 15_000,  // 从 10000 扩到 15000
        max_agents: 300,
        max_api_calls_per_day: 2_000_000,
        max_storage_mb: 204_800,
    };
    manager.update_quota(gs_id, &new_quota)?;
    println!("updated gs quota: {:?}", manager.get_quota(gs_id)?);

    // 2. 创建用量跟踪器
    let tracker = Arc::new(TenantUsageTracker::new(storage));

    // 3. 启动每日 0 点 UTC 重置任务
    spawn_daily_reset_task(tracker.clone());

    Ok(tracker)
}

fn register_device(
    tracker: &TenantUsageTracker,
    manager: &TenantManager,
    gs_id: &str,
) -> anyhow::Result<()> {
    // 1. 先校验配额（避免超限后回滚）
    let quota = manager.get_quota(gs_id)?;
    tracker.check_quota(gs_id, QuotaResource::Devices, &quota)?;

    // 2. 实际注册设备（业务逻辑省略）
    // ...device_registry.add(...)

    // 3. 递增计数器
    let new_count = tracker.increment_device_count(gs_id)?;
    println!("gs current_devices = {}", new_count);

    Ok(())
}

fn handle_api_request(
    tracker: &TenantUsageTracker,
    manager: &TenantManager,
    gs_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    // 每次请求扣减 API 调用配额
    let quota = manager.get_quota(gs_id)?;
    tracker.check_quota(gs_id, QuotaResource::ApiCalls, &quota)?;
    let calls = tracker.increment_api_calls(gs_id)?;
    println!("gs api_calls_today = {}/{}", calls, quota.max_api_calls_per_day);
    Ok(())
}

TenantUsageTracker 主要方法：

QuotaExceeded 错误携带 resource / limit / current 三个字段，Gateway 据此返回 HTTP 429：

HTTP/1.1 429 Too Many Requests
Content-Type: application/json
Retry-After: 3600

{
  "error": "quota_exceeded",
  "resource": "api_calls",
  "limit": 1000000,
  "current": 1000000,
  "reset_at": "2026-07-07T00:00:00Z"
}

步骤 7：API Key 绑定与请求路由

对于不支持 mTLS 的轻量客户端（如移动 App、Web 前端），使用 API Key 完成租户路由。每个 API Key 绑定到唯一的 (tenant_id, username, role) 三元组。

use eneros_tenant::TenantManager;

fn bind_api_keys(manager: &TenantManager, gs_id: &str, ps_id: &str) -> anyhow::Result<()> {
    // 国调 admin 用户的 API Key（生产环境应通过 KMS 生成）
    manager.bind_api_key(
        "ek_live_gs_alice_5f8a7b2c9d1e3f4a",
        "alice@gs.com",
        "admin",
        gs_id,
    )?;

    // 省调 operator 用户的 API Key
    manager.bind_api_key(
        "ek_live_ps_dave_2a3b4c5d6e7f8a9b",
        "dave@ps.com",
        "operator",
        ps_id,
    )?;

    // 验证绑定
    let binding = manager.get_api_key_tenant("ek_live_gs_alice_5f8a7b2c9d1e3f4a")?
        .expect("API key should be bound");
    assert_eq!(binding.tenant_id, gs_id);
    assert_eq!(binding.username, "alice@gs.com");
    assert_eq!(binding.role, "admin");

    // 未知 key 返回 None
    assert!(manager.get_api_key_tenant("ek_live_unknown_xxx")?.is_none());

    Ok(())
}

Gateway 中间件示例：

use axum::{extract::Request, http::StatusCode, middleware::Next, response::Response};
use eneros_tenant::TenantManager;

pub async fn tenant_routing_middleware(
    manager: TenantManager,
    req: Request,
    next: Next,
) -> Result<Response, StatusCode> {
    // 1. 优先从 mTLS 证书中提取 tenant_id
    let tenant_id = if let Some(cert) = req.extensions().get::<ClientCert>() {
        cert.tenant_id.clone()
    } else {
        // 2. 回退到 API Key
        let api_key = req
            .headers()
            .get("X-API-Key")
            .and_then(|h| h.to_str().ok())
            .ok_or(StatusCode::UNAUTHORIZED)?;

        let binding = manager
            .get_api_key_tenant(api_key)
            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
            .ok_or(StatusCode::UNAUTHORIZED)?;

        binding.tenant_id
    };

    // 3. 校验租户状态（Active 才放行）
    manager.validate_tenant_access(&tenant_id)
        .map_err(|e| match e {
            eneros_tenant::TenantError::NotFound(_) => StatusCode::NOT_FOUND,
            eneros_tenant::TenantError::InvalidStatus(_) => StatusCode::FORBIDDEN,
            _ => StatusCode::INTERNAL_SERVER_ERROR,
        })?;

    // 4. 注入租户上下文
    req.extensions_mut().insert(TenantContext {
        tenant_id: tenant_id.clone(),
        user_id: binding.username.clone(),
        role: binding.role.clone(),
    });

    Ok(next.run(req).await)
}

重要：get_api_key_tenant 仅返回 status='active' 的租户绑定。Suspended / Deleted 租户的 API Key 自动失效，无需手动清理。

步骤 8：跨租户访问控制

默认拒绝跨租户访问，显式授权才放行。liaison 角色用于厂家联络人跨租户只读访问。

use eneros_auth::{Policy, Subject, Action, Resource, AuthContext};

async fn setup_cross_tenant_policy(policy: &Policy) -> anyhow::Result<()> {
    // 1. 默认拒绝所有跨租户访问
    policy.deny(
        Subject::any_tenant(),
        Action::any(),
        Resource::tenant_pattern("*"),
    ).await?;

    // 2. 允许同租户内访问
    policy.allow(
        Subject::tenant("gs"),
        Action::any(),
        Resource::tenant("gs"),
    ).await?;

    // 3. 厂家联络人只读跨租户访问（带条件）
    policy.allow(
        Subject::tenant("ps"),
        Action::Read,
        Resource::tenant("gs").topic("/topics/state_estimate"),
    )
    .when(|ctx: &AuthContext| ctx.caller.has_role("liaison"))
    .await?;

    // 4. 审计场景：合规系统跨租户只读
    policy.allow(
        Subject::service("audit-collector"),
        Action::Read,
        Resource::tenant_pattern("*").topic("/topics/audit_log"),
    )
    .with_expiry(chrono::Duration::days(90))
    .await?;

    Ok(())
}

访问决策表：

步骤 9：每日 API 计数器重置

api_calls_today 计数器每天 UTC 0 点自动归零，last_reset_date 刷新为当天。spawn_daily_reset_task 在后台 tokio 任务中循环 sleep 到下一个 UTC 午夜。

use std::sync::Arc;
use eneros_tenant::{SqliteTenantStorage, TenantUsageTracker};
use eneros_tenant::usage_tracker::spawn_daily_reset_task;

fn start_daily_reset(storage: Arc<SqliteTenantStorage>) -> Arc<TenantUsageTracker> {
    let tracker = Arc::new(TenantUsageTracker::new(storage));

    // 后台任务：每天 UTC 0 点重置
    spawn_daily_reset_task(tracker.clone());

    // 启动时立即检查一次（防止服务重启后跨天未重置）
    if let Err(e) = tracker.maybe_reset_daily() {
        tracing::error!("initial daily reset check failed: {}", e);
    }

    tracker
}

maybe_reset_daily 与 reset_daily_counters 的差异：

设备数 / Agent 数 / 存储用量不受每日重置影响，仅 api_calls_today 归零。

步骤 10：完整部署流程

以下为一个完整的多租户部署示例，涵盖从存储初始化到中间件挂载的全过程：

use std::sync::Arc;
use axum::{routing::get, Router};
use eneros_tenant::{
    QuotaResource, SqliteTenantStorage, TenantManager, TenantQuota,
    TenantUsageTracker,
};
use eneros_tenant::usage_tracker::spawn_daily_reset_task;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // 1. 初始化存储
    let storage = Arc::new(SqliteTenantStorage::new("/var/lib/eneros/tenant.db")?);
    let manager = TenantManager::new(storage.clone());
    let tracker = Arc::new(TenantUsageTracker::new(storage.clone()));

    // 2. 启动每日重置任务
    tracker.maybe_reset_daily()?;
    spawn_daily_reset_task(tracker.clone());

    // 3. 幂等创建租户
    let gs_quota = TenantQuota {
        max_devices: 10_000,
        max_agents: 200,
        max_api_calls_per_day: 1_000_000,
        max_storage_mb: 102_400,
    };
    let gs = manager.ensure_tenant_by_name("国调中心", gs_quota.clone())?;
    let ps = manager.ensure_tenant_by_name("省调中心", TenantQuota {
        max_devices: 2_000,
        max_agents: 50,
        max_api_calls_per_day: 200_000,
        max_storage_mb: 20_480,
    })?;

    // 4. 注册成员
    manager.add_member(&gs.id, "alice@gs.com", "admin")?;
    manager.add_member(&ps.id, "dave@ps.com", "operator")?;

    // 5. 绑定 API Key
    manager.bind_api_key(
        "ek_live_gs_alice_xxx",
        "alice@gs.com",
        "admin",
        &gs.id,
    )?;

    // 6. 挂载 Axum 中间件
    let app = Router::new()
        .route("/healthz", get(|| async { "ok" }))
        .route("/api/v1/devices", get(list_devices))
        .layer(axum::middleware::from_fn_with_state(
            manager.clone(),
            tenant_routing_middleware,
        ))
        .with_state(manager);

    // 7. 启动 HTTP 服务
    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
    tracing::info!("gateway listening on :8080");
    axum::serve(listener, app).await?;

    Ok(())
}

// 业务 handler 通过 TenantContext 读取当前租户
async fn list_devices(
    ctx: axum::extract::Extension<TenantContext>,
) -> Result<String, StatusCode> {
    let tenant_id = &ctx.0.tenant_id;
    Ok(format!("listing devices for tenant {}", tenant_id))
}

上例中的 ensure_tenant_by_name 是项目封装的便捷方法，内部按 name 查找，未命中则调用 create_tenant。原版 TenantManager 仅提供 ensure_default_tenant。

验证

按以下清单逐项验证多租户隔离是否生效：

并发安全验证脚本：

#[cfg(test)]
mod tests {
    use super::*;
    use std::thread;

    #[test]
    fn test_concurrent_api_calls_are_atomic() {
        let storage = Arc::new(SqliteTenantStorage::new_in_memory().unwrap());
        let tracker = Arc::new(TenantUsageTracker::new(storage.clone()));
        let tenant = setup_test_tenant(&storage);

        let mut handles = vec![];
        for _ in 0..100 {
            let t = tracker.clone();
            let id = tenant.id.clone();
            handles.push(thread::spawn(move || {
                t.increment_api_calls(&id).unwrap();
            }));
        }
        for h in handles {
            h.join().unwrap();
        }

        let usage = tracker.get_usage(&tenant.id).unwrap();
        assert_eq!(usage.api_calls_today, 100, "100 并发应精确写入 100");
    }
}

调试与排错

查看租户存储内部状态：

# 查看所有租户
sqlite3 /var/lib/eneros/tenant.db \
  "SELECT id, name, status, max_devices, max_agents FROM tenants;"

# 查看某租户用量
sqlite3 /var/lib/eneros/tenant.db \
  "SELECT * FROM tenant_usage WHERE tenant_id='8a7b3c2d-1e2f-4a3b-9c8d-7e6f5a4b3c2d';"

# 查看所有 API Key 绑定
sqlite3 /var/lib/eneros/tenant.db \
  "SELECT api_key, username, role, tenant_id FROM tenant_api_keys;"

# 手动触发每日重置（应急用）
sqlite3 /var/lib/eneros/tenant.db \
  "UPDATE tenant_usage SET api_calls_today=0, last_reset_date=date('now');"

数据表结构参考

eneros-tenant 使用四张表，所有时间戳以 RFC 3339 字符串存储，日期以 YYYY-MM-DD 存储：

-- 租户主表
CREATE TABLE tenants (
    id TEXT PRIMARY KEY,                -- UUID v4 或 "default"
    name TEXT NOT NULL,
    status TEXT NOT NULL,               -- 'active' / 'suspended' / 'deleted'
    created_at TEXT NOT NULL,           -- RFC 3339
    updated_at TEXT NOT NULL,           -- RFC 3339
    max_devices INTEGER NOT NULL,
    max_agents INTEGER NOT NULL,
    max_api_calls_per_day INTEGER NOT NULL,
    max_storage_mb INTEGER NOT NULL
);

-- 成员表（复合主键）
CREATE TABLE tenant_members (
    tenant_id TEXT NOT NULL,
    user_id TEXT NOT NULL,
    role TEXT NOT NULL,
    added_at TEXT NOT NULL,
    PRIMARY KEY (tenant_id, user_id),
    FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);

-- 用量表（每租户一行）
CREATE TABLE tenant_usage (
    tenant_id TEXT PRIMARY KEY,
    current_devices INTEGER NOT NULL,
    current_agents INTEGER NOT NULL,
    api_calls_today INTEGER NOT NULL,
    storage_used_mb INTEGER NOT NULL,
    last_reset_date TEXT NOT NULL,      -- 'YYYY-MM-DD'
    FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);

-- API Key 绑定表
CREATE TABLE tenant_api_keys (
    api_key TEXT PRIMARY KEY,
    username TEXT NOT NULL,
    role TEXT NOT NULL,
    tenant_id TEXT NOT NULL,
    FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);

SQLite 初始化参数（SqliteTenantStorage::new 自动执行）：

PRAGMA journal_mode = WAL;       -- 写前日志，提升并发读
PRAGMA synchronous = NORMAL;     -- 平衡性能与持久化
PRAGMA foreign_keys = ON;        -- 强制外键约束

下一步

• 多租户与隔离 — 隔离模型详解
• 高可用集群部署 — 集群级隔离
• 插件开发 — 多租户插件签名与分发