diff --git a/msync/MSYNC.md b/msync/MSYNC.md new file mode 100644 index 0000000..05f2711 --- /dev/null +++ b/msync/MSYNC.md @@ -0,0 +1,558 @@ +# MSync — 微服务通信库 + +> GVsds 平台微服务架构的多端通信、服务间验权、对象全局映射（OGM）与分布式事务管理。 +> 本服务器是**主服务器**（账本服务器），存储所有分布式服务地址，服务间需验权来互相获取 OGM 等信息。 +> 实现位于 `gvsdsdk/msync`（SDK 原生实现，非封装层）。 + +--- + +## 概述 + +MSync 是为 GVsds 多域微服务平台设计的通信中间件，运行于**主服务器**（账本服务器），提供以下核心能力： + +- **服务注册与发现** — 微服务节点的注册、注销、心跳、密钥轮换 +- **服务间 HMAC 签名认证** — 基于 HMAC-SHA256 的双向验权，自包含令牌签发 +- **OGM 对象全局映射** — 跨服务对象身份注册与解析 +- **分布式事务管理** — 基于 Saga 模式的跨服务事务编排、补偿与重试 + +### 什么是"账本服务器"？ + +账本服务器即**主服务器**，它维护一张"账本"——记录所有分布式微服务的地址、状态、密钥等信息。任何微服务要加入集群，必须先向账本服务器注册；任何微服务要与其他服务通信，必须先通过账本服务器验权获取令牌。 + +--- + +## 架构 + +``` + ┌──────────────────────────────────┐ + │ 账本服务器 (Master) │ + │ │ + │ ┌──────────┐ ┌──────────────┐ │ + │ │ Registry │ │ Auth (HMAC) │ │ + │ │ (服务账本)│ │ (服务间验权) │ │ + │ └──────────┘ └──────────────┘ │ + │ ┌──────────┐ ┌──────────────┐ │ + │ │ OGM │ │ Transaction │ │ + │ │(对象映射)│ │ (Saga编排) │ │ + │ └──────────┘ └──────────────┘ │ + └──────────┬───────────────────────┘ + │ + ┌────────────────┼────────────────┐ + │ │ │ + ┌─────┴─────┐ ┌────┴─────┐ ┌─────┴─────┐ + │ GERP │ │ CRB │ │ T4 │ + │ (企业ERP) │ │(货币经济)│ │(远程桌面) │ + └───────────┘ └──────────┘ └───────────┘ +``` + +--- + +## 认证流程 + +``` + 工作服务 账本服务器 目标服务 + (Worker) (Master) (Target) + + 1. 构建 AuthPayload ──────▶ + {service, challenge, ts} + + 2. HMAC-SHA256 签名 ──────▶ + signature = HMAC(payload, secret) + + 3. 查找 ServiceNode + 4. 验证签名 + 5. 签发 AccessToken + RefreshToken + ◀──── 返回令牌对 + + 6. 携带令牌请求 ─────────────────────────────────────────▶ + Authorization: Service + + 7. 本地验证令牌 + 8. 检查权限范围 + 9. 执行业务逻辑 + ◀───────────────────────────────── 返回结果 +``` + +### 令牌机制 + +| 令牌类型 | 有效期 | 用途 | +|---------|--------|------| +| AccessToken | 2 小时 | 服务间请求认证 | +| RefreshToken | 30 天 | 刷新 AccessToken | + +令牌为**自包含格式**（类似 JWT），目标服务可使用主密钥本地验证，无需回调账本服务器。 + +### 权限范围（Scope） + +| Scope | 说明 | +|-------|------| +| `ogm:read` | 读取 OGM 映射 | +| `ogm:write` | 注册/更新 OGM 映射 | +| `registry:read` | 发现服务 | +| `registry:write` | 注册/注销服务 | +| `tx:read` | 查询事务状态 | +| `tx:write` | 创建/执行/补偿事务 | +| `*` | 全部权限 | + +### 密钥认证强制 + +- 新服务注册必须携带 `ServiceSecret`（与 `ServiceSecretHash` 自洽），否则拒绝 +- 除 register / auth / auth:refresh 外，所有 REST 与 SIO 业务操作必须持密钥签发的令牌 +- 心跳/注销仅允许操作本服务 + +--- + +## 分布式事务管理 + +### 远程 + 本地双层架构 + +MSync 采用**远程事务管理器 + 本地事务管理器**双层架构： + +``` + 主服务器 (远程事务管理器 TransactionManager) + │ + │ 创建事务，编排步骤，管理状态 + │ + ├──▶ POST {gerp_domain}/api/msync/local/execute/ + │ {step_name: "create_user", payload: {...}} + │ → GERP 的 LocalTransactionManager.execute_step() + │ + ├──▶ POST {crb_domain}/api/msync/local/execute/ + │ {step_name: "create_wallet", payload: {...}} + │ → CRB 的 LocalTransactionManager.execute_step() + │ + └──▶ POST {crb_domain}/api/msync/local/compensate/ + {step_name: "create_wallet", payload: {...}, action_result: {...}} + → CRB 的 LocalTransactionManager.compensate_step() +``` + +- **远程事务管理器**（主服务器）：创建事务、编排步骤、管理状态、调用各服务的本地端点 +- **本地事务管理器**（每个工作服务）：注册步骤处理器、执行本地逻辑、补偿本地操作 + +### Saga 模式 + +``` + Step 0 (GERP) Step 1 (CRB) Step 2 (T4) + 创建用户 ────────▶ 创建钱包 ────────▶ 注册终端 + │ │ │ + │ 补偿:删除用户 │ 补偿:删除钱包 │ 补偿:注销终端 + ◀──────────────── ◀────────────────── ◀ +``` + +- **正向执行**：远程管理器按步骤顺序调用各服务的本地 execute 端点 +- **补偿回滚**：任一步骤失败时，远程管理器逆序调用已完成步骤的本地 compensate 端点 +- **超时检测**：自动标记超时事务并触发补偿 +- **重试机制**：正向步骤失败可 `RetryStep`，补偿失败可 `RetryCompensate`（可配置最大次数） + +### 事务生命周期 + +``` +Pending ──▶ Running ──▶ Committed (全部步骤成功) + │ + ├─▶ Failed ──▶ Compensating ──▶ Compensated (补偿成功) + │ + └─▶ TimedOut ──▶ Compensating ──▶ Compensated +``` + +| 状态 | 代码 | 说明 | +|------|------|------| +| 待执行 | 0 | 事务已创建，尚未开始 | +| 执行中 | 1 | 正在执行步骤 | +| 已提交 | 2 | 全部步骤成功完成 | +| 补偿中 | 3 | 正在逆序执行补偿操作 | +| 已补偿 | 4 | 补偿操作全部完成 | +| 失败 | 5 | 步骤执行失败且补偿未完成 | +| 超时 | 6 | 事务超时 | + +### 步骤状态 + +| 状态 | 代码 | 说明 | +|------|------|------| +| 待执行 | 0 | 步骤尚未开始 | +| 执行中 | 1 | 正在调用 action 端点 | +| 已完成 | 2 | action 调用成功 | +| 补偿中 | 3 | 正在调用 compensate 端点 | +| 已补偿 | 4 | compensate 调用成功 | +| 失败 | 5 | 调用失败 | +| 已跳过 | 6 | 事务取消时跳过 | + +--- + +## 数据模型 + +### ServiceNode — 服务节点（账本记录） + +| 字段 | 类型 | 说明 | +|------|------|------| +| ServiceUUID | BINARY(16) PK | 服务唯一标识 | +| ServiceName | VARCHAR(128) UNIQUE | 服务名称 | +| ServiceDomain | VARCHAR(256) | 服务域名/URL | +| ServiceSecretHash | VARCHAR(128) | bcrypt 哈希密钥 | +| ServiceScope | JSON | 权限范围列表 | +| ServiceStatus | SMALLINT | 状态 (1=活跃) | +| ServiceType | SMALLINT | 类型 (0=工作节点, 1=主节点) | +| ServiceMetadata | JSON | 服务元数据 | +| CreateTime | DATETIME | 注册时间 | +| LastHeartbeat | DATETIME | 最后心跳时间 | + +### ServiceToken — 服务令牌 + +| 字段 | 类型 | 说明 | +|------|------|------| +| TokenUUID | BINARY(16) PK | 令牌唯一标识 | +| ServiceUUID | FK → ServiceNode | 所属服务 | +| AccessToken | VARCHAR(512) UNIQUE | 访问令牌 | +| RefreshToken | VARCHAR(512) UNIQUE | 刷新令牌 | +| TokenScope | JSON | 令牌权限范围 | +| ExpiresAt | DATETIME | 访问令牌过期时间 | +| RefreshExpiresAt | DATETIME | 刷新令牌过期时间 | +| TokenStatus | SMALLINT | 状态 (1=有效, 0=已撤销) | + +### ObjectGlobalMapping — OGM 对象全局映射 + +| 字段 | 类型 | 说明 | +|------|------|------| +| OGMUUID | BINARY(16) PK | 映射唯一标识 | +| ObjectType | VARCHAR(64) | 对象类型 (user, company, asset...) | +| ObjectLocalID | VARCHAR(128) | 本地 ID | +| ObjectGlobalID | VARCHAR(128) UNIQUE | 全局 ID | +| OwnerServiceUUID | FK → ServiceNode | 所属服务 | +| ObjectMetadata | JSON | 对象元数据 | +| ObjectStatus | SMALLINT | 状态 (1=活跃, 0=已注销) | + +**唯一约束**: `(ObjectType, ObjectLocalID, OwnerServiceUUID)` + +### DistributedTransaction — 分布式事务 + +| 字段 | 类型 | 说明 | +|------|------|------| +| TransactionUUID | BINARY(16) PK | 事务唯一标识 | +| TransactionCode | VARCHAR(64) UNIQUE | 事务代码 (DTX-XXXXXXXXXXXX) | +| TransactionType | VARCHAR(16) | 事务类型 (saga/2pc) | +| TransactionName | VARCHAR(128) | 事务名称 | +| TransactionDescription | VARCHAR(512) | 描述 | +| TransactionStatus | SMALLINT | 事务状态 | +| InitiatorServiceUUID | FK → ServiceNode | 发起服务 | +| ReferenceOGMUUID | BINARY(16) | 关联 OGM | +| TransactionPayload | JSON | 事务级载荷 | +| TransactionResult | JSON | 事务结果 | +| TimeoutSeconds | INT | 超时秒数 (默认 300) | +| CreateTime | DATETIME | 创建时间 | +| EndTime | DATETIME | 结束时间 | + +### TransactionStep — 事务步骤 + +| 字段 | 类型 | 说明 | +|------|------|------| +| StepUUID | BINARY(16) PK | 步骤唯一标识 | +| TransactionUUID | FK → DistributedTransaction | 所属事务 | +| StepIndex | INT | 步骤序号 | +| StepName | VARCHAR(128) | 步骤名称 | +| ParticipantServiceUUID | FK → ServiceNode | 参与服务 | +| StepStatus | SMALLINT | 步骤状态 | +| ActionEndpoint | VARCHAR(512) | 正向操作端点 | +| ActionMethod | VARCHAR(8) | 正向操作方法 | +| ActionPayload | JSON | 正向操作载荷 | +| CompensateEndpoint | VARCHAR(512) | 补偿操作端点 | +| CompensateMethod | VARCHAR(8) | 补偿操作方法 | +| CompensatePayload | JSON | 补偿操作载荷 | +| StepResult | JSON | 步骤执行结果 | +| RetryCount | INT | 已重试次数 | +| MaxRetries | INT | 最大重试次数 (默认 3) | + +--- + +## API 端点 + +### 认证 + +| 方法 | 端点 | 说明 | +|------|------|------| +| POST | `/api/msync/auth/token/` | 服务认证获取令牌 | +| POST | `/api/msync/auth/token/refresh/` | 刷新令牌 | + +### 服务注册与发现（账本） + +| 方法 | 端点 | 说明 | +|------|------|------| +| GET | `/api/msync/registry/` | 列出所有已注册服务 | +| POST | `/api/msync/registry/` | 注册新服务（必须携带密钥） | +| GET | `/api/msync/registry//` | 发现指定服务 | +| POST | `/api/msync/registry//heartbeat/` | 服务心跳（需令牌，仅本服务） | +| DELETE | `/api/msync/registry//deregister/` | 注销服务（需令牌，仅本服务） | + +### OGM 对象全局映射 + +| 方法 | 端点 | 说明 | +|------|------|------| +| POST | `/api/msync/ogm/` | 注册对象映射 | +| POST | `/api/msync/ogm/batch/` | 批量注册 | +| GET | `/api/msync/ogm//` | 解析全局 ID | +| GET | `/api/msync/ogm/query/` | 按类型/服务查询 | +| PUT | `/api/msync/ogm//update/` | 更新元数据 | +| DELETE | `/api/msync/ogm//deregister/` | 注销映射 | + +### 分布式事务 + +| 方法 | 端点 | 说明 | +|------|------|------| +| POST | `/api/msync/tx/` | 创建事务 | +| GET | `/api/msync/tx/query/?status=0` | 按状态查询事务 | +| POST | `/api/msync/tx/timeout-check/` | 检测超时事务 | +| GET | `/api/msync/tx//` | 事务详情 | +| POST | `/api/msync/tx//execute/` | 执行事务 | +| POST | `/api/msync/tx//compensate/` | 补偿事务 | +| POST | `/api/msync/tx//cancel/` | 取消事务 | +| POST | `/api/msync/tx//retry/` | 重试正向失败步骤 | +| POST | `/api/msync/tx//retry-compensate/` | 重试补偿失败步骤 | + +### 本地事务管理器（每个工作服务） + +| 方法 | 端点 | 说明 | +|------|------|------| +| POST | `/api/msync/local/execute/` | 执行本地步骤（由主服务器调用） | +| POST | `/api/msync/local/compensate/` | 补偿本地步骤（由主服务器调用） | +| GET | `/api/msync/local/handlers/` | 列出已注册的步骤处理器 | + +**创建事务示例**: +```json +POST /api/msync/tx/ +{ + "name": "用户注册流程", + "description": "跨服务用户注册：创建用户→创建钱包→注册终端", + "transaction_type": "saga", + "timeout_seconds": 300, + "steps": [ + { + "name": "创建用户", + "participant_service": "gerp", + "action_endpoint": "https://gerp.gvsds.com/api/users/", + "action_payload": {"username": "newuser"}, + "compensate_endpoint": "https://gerp.gvsds.com/api/users/compensate/", + "compensate_payload": {"username": "newuser"} + }, + { + "name": "创建钱包", + "participant_service": "crb", + "action_endpoint": "https://crb.gvsds.com/api/wallets/", + "action_payload": {"owner": "newuser"}, + "compensate_endpoint": "https://crb.gvsds.com/api/wallets/compensate/", + "compensate_payload": {"owner": "newuser"} + }, + { + "name": "注册终端", + "participant_service": "t4", + "action_endpoint": "https://t4.gvsds.com/api/terminals/", + "action_payload": {"user": "newuser"}, + "compensate_endpoint": "https://t4.gvsds.com/api/terminals/compensate/", + "compensate_payload": {"user": "newuser"} + } + ] +} +``` + +**执行事务**: +```json +POST /api/msync/tx/DTX-ABC123DEF456/execute/ +{ + "access_token": "eyJhbGci..." +} +``` + +**重试失败步骤**: +```json +POST /api/msync/tx/DTX-ABC123DEF456/retry/ +{ + "step_index": 1, + "access_token": "eyJhbGci..." +} +``` + +**重试补偿失败步骤**: +```json +POST /api/msync/tx/DTX-ABC123DEF456/retry-compensate/ +{ + "step_index": 1, + "access_token": "eyJhbGci..." +} +``` + +--- + +## Python API + +### 服务注册（账本操作） + +```python +from gvsdsdk.msync.registry import ServiceRegistry, RegistryError + +node = ServiceRegistry.Register( + name='gerp', + domain='https://gerp.gvsds.com', + secret='my-super-secret-key', + scope=['ogm:read', 'ogm:write', 'tx:read', 'tx:write'], +) +info = ServiceRegistry.Discover('gerp') +ServiceRegistry.Heartbeat('gerp') +ServiceRegistry.RotateSecret('gerp', 'new-secret-key') +ServiceRegistry.Deregister('gerp') +``` + +### 服务客户端 + +```python +from gvsdsdk.msync.client import ServiceClient + +client = ServiceClient('gerp', 'secret', 'https://api.gvsds.com') +client.Authenticate() +ogm = client.ResolveOGM('user:550e8400') +info = client.DiscoverService('crb') +``` + +### OGM 管理 + +```python +from gvsdsdk.msync.ogm import OGMManager + +ogm = OGMManager.Register('user', 'uuid-1', 'core', global_id='user:uuid1') +info = OGMManager.Resolve('user:uuid1') +users = OGMManager.QueryByType('user') +OGMManager.Deregister('user:uuid1') +``` + +### 分布式事务 + +```python +from gvsdsdk.msync.transaction import TransactionManager, TransactionError + +# 创建事务 +dtx = TransactionManager.Create( + name='用户注册流程', + steps_data=[ + { + 'name': 'create_user', + 'participant_service': 'gerp', + 'action_payload': {'username': 'newuser'}, + 'compensate_payload': {'username': 'newuser'}, + }, + { + 'name': 'create_wallet', + 'participant_service': 'crb', + 'action_payload': {'owner': 'newuser'}, + 'compensate_payload': {'owner': 'newuser'}, + }, + ], + timeout_seconds=300, +) + +# 执行事务（远程管理器自动调用各服务的本地管理器） +dtx = TransactionManager.Execute(dtx.TransactionCode, access_token='...') + +# 查看详情 +detail = TransactionManager.GetDetail(dtx.TransactionCode) + +# 手动补偿 +dtx = TransactionManager.Compensate(dtx.TransactionCode, access_token='...') + +# 重试正向失败步骤 +step = TransactionManager.RetryStep(dtx.TransactionCode, step_index=1, access_token='...') + +# 重试补偿失败步骤 +step = TransactionManager.RetryCompensate(dtx.TransactionCode, step_index=1, access_token='...') + +# 取消待执行事务 +TransactionManager.Cancel(dtx.TransactionCode) + +# 检测超时 +timed_out = TransactionManager.CheckTimeouts() + +# 按状态查询 +running = TransactionManager.QueryByStatus(1) +``` + +### 本地事务管理器（工作服务端） + +```python +from gvsdsdk.msync.local_tx import local_tx + +# 注册步骤处理器（在服务启动时） +def create_user_action(payload): + user = User.objects.create(username=payload['username']) + return {'user_uuid': str(user.UserUUID)} + +def create_user_compensate(payload, action_result): + User.objects.filter(username=payload['username']).delete() + return {'deleted': True} + +local_tx.register( + 'create_user', + action=create_user_action, + compensate=create_user_compensate, +) + +# 列出已注册处理器 +handlers = local_tx.list_handlers() + +# 注销处理器 +local_tx.unregister('create_user') +``` + +### 权限装饰器 + +```python +from gvsdsdk.msync.middleware import require_service_scope, ServicePermission + +@require_service_scope('ogm:read') +def my_view(request): + pass + +class MyViewSet(ViewSet): + permission_classes = [ServicePermission('tx:read', 'tx:write')] +``` + +--- + +## 配置 + +### settings.py + +```python +INSTALLED_APPS = [ + 'gvsdsdk.msync', +] + +MIDDLEWARE = [ + 'gvsdsdk.msync.middleware.ServiceAuthMiddleware', +] +``` + +### 数据库路由 + +msync 使用 `default` 数据库（Users 库），与 core、gerp 共享。 + +### secrets.ini + +```ini +[msync] +MASTER_SECRET = your-master-secret-for-token-signing +``` + +--- + +## 文件清单 + +| 文件 | 说明 | +|------|------| +| `models.py` | 5 个数据模型（ServiceNode, ServiceToken, ObjectGlobalMapping, DistributedTransaction, TransactionStep） | +| `auth.py` | HMAC 签名 + 自包含令牌编解码 + 服务端签发/验证 | +| `client.py` | ServiceClient 跨服务请求客户端 | +| `middleware.py` | 服务认证中间件 + 权限装饰器 | +| `registry.py` | 服务注册/发现/心跳/密钥轮换 | +| `ogm.py` | OGM 对象全局映射管理 | +| `transaction.py` | 远程事务管理器（Saga 编排、补偿、重试、超时，调用本地管理器） | +| `local_tx.py` | 本地事务管理器（步骤处理器注册、本地执行与补偿） | +| `sio.py` | Socket.IO 服务端（注册/认证/心跳/OGM/事务事件 + 密钥认证守卫） | +| `views.py` | REST API 视图 | +| `serializers.py` | DRF 序列化器 | +| `urls.py` | API 路由 | +| `admin.py` | Django Admin 注册 | diff --git a/msync/__init__.py b/msync/__init__.py index a07283b..6b60a0e 100644 --- a/msync/__init__.py +++ b/msync/__init__.py @@ -1,19 +1,17 @@ """GVSDS SDK MSYNC 子包 -将 MSYNC 相关模块统一组织在 gvsdsdk.msync 命名空间下。 - -自有实现（无 Django 依赖，子服务器可直接使用）： - - auth: HMAC-SHA256 签名验证、服务令牌编解码 +MSYNC 全部实现位于 gvsdsdk.msync（原生实现，非封装层）： + - models: ServiceNode / ServiceToken / OGM / 分布式事务模型 + - auth: HMAC-SHA256 签名验证、服务令牌编解码、服务端签发/验证 + - registry / ogm / transaction: 服务注册发现 / OGM 管理 / Saga 编排 - client: 微服务通信客户端（REST） - local_tx: 本地事务步骤处理器 - -服务端转发（依赖 Django ORM，仅在主服务器可用）： - - registry: 服务注册/发现 - - ogm: 对象全局映射 - - transaction: Saga 分布式事务 + - middleware / views / serializers / urls / sio / admin: Django 集成层 """ -# 自有实现（无 Django 依赖，始终可用） +default_app_config = 'gvsdsdk.msync.apps.MsyncConfig' + +# 自有实现（无 Django ORM 依赖，子服务器可直接使用） from gvsdsdk.msync.auth import ( # noqa SignPayload, VerifySignature, @@ -26,7 +24,7 @@ from gvsdsdk.msync.auth import ( # noqa from gvsdsdk.msync.client import ServiceClient, ServiceClientError # noqa from gvsdsdk.msync.local_tx import LocalTransactionManager, StepHandler, local_tx # noqa -# 服务端转发（Django ORM 依赖，延迟导入避免子服务器初始化报错） +# ORM 依赖的模块延迟导入（仅在主服务器使用） def __getattr__(name): _server_exports = { 'ServiceRegistry': 'gvsdsdk.msync.registry', @@ -36,10 +34,10 @@ def __getattr__(name): 'TransactionManager': 'gvsdsdk.msync.transaction', 'TransactionError': 'gvsdsdk.msync.transaction', 'StepExecutionError': 'gvsdsdk.msync.transaction', - 'AuthenticateService': 'gvsds.apps.msync.auth', - 'RefreshServiceToken': 'gvsds.apps.msync.auth', - 'ValidateAccessToken': 'gvsds.apps.msync.auth', - 'IssueServiceToken': 'gvsds.apps.msync.auth', + 'AuthenticateService': 'gvsdsdk.msync.auth', + 'RefreshServiceToken': 'gvsdsdk.msync.auth', + 'ValidateAccessToken': 'gvsdsdk.msync.auth', + 'IssueServiceToken': 'gvsdsdk.msync.auth', } if name in _server_exports: import importlib @@ -57,7 +55,7 @@ __all__ = [ 'SignAuthPayload', 'VerifyAuthPayload', 'ServiceTokenEncoder', - # auth (服务端转发) + # auth (服务端) 'AuthenticateService', 'RefreshServiceToken', 'ValidateAccessToken', @@ -69,13 +67,13 @@ __all__ = [ 'LocalTransactionManager', 'StepHandler', 'local_tx', - # registry (服务端转发) + # registry (服务端) 'ServiceRegistry', 'RegistryError', - # ogm (服务端转发) + # ogm (服务端) 'OGMManager', 'OGMError', - # transaction (服务端转发) + # transaction (服务端) 'TransactionManager', 'TransactionError', 'StepExecutionError', diff --git a/msync/admin.py b/msync/admin.py new file mode 100644 index 0000000..57be602 --- /dev/null +++ b/msync/admin.py @@ -0,0 +1,45 @@ +from django.contrib import admin +from gvsdsdk.msync.models import ( + ServiceNode, + ServiceToken, + ObjectGlobalMapping, + DistributedTransaction, + TransactionStep, +) + + +@admin.register(ServiceNode) +class ServiceNodeAdmin(admin.ModelAdmin): + list_display = ('ServiceName', 'ServiceDomain', 'ServiceType', 'ServiceStatus', 'CreateTime', 'LastHeartbeat') + list_filter = ('ServiceType', 'ServiceStatus') + search_fields = ('ServiceName', 'ServiceDomain') + readonly_fields = ('ServiceUUID', 'ServiceSecretHash', 'CreateTime') + + +@admin.register(ServiceToken) +class ServiceTokenAdmin(admin.ModelAdmin): + list_display = ('TokenUUID', 'ServiceUUID', 'ExpiresAt', 'TokenStatus', 'CreateTime') + list_filter = ('TokenStatus',) + readonly_fields = ('TokenUUID', 'AccessToken', 'RefreshToken', 'CreateTime') + + +@admin.register(ObjectGlobalMapping) +class ObjectGlobalMappingAdmin(admin.ModelAdmin): + list_display = ('ObjectGlobalID', 'ObjectType', 'ObjectLocalID', 'OwnerServiceUUID', 'ObjectStatus', 'CreateTime') + list_filter = ('ObjectType', 'ObjectStatus') + search_fields = ('ObjectGlobalID', 'ObjectLocalID') + + +@admin.register(DistributedTransaction) +class DistributedTransactionAdmin(admin.ModelAdmin): + list_display = ('TransactionCode', 'TransactionName', 'TransactionType', 'TransactionStatus', 'InitiatorServiceUUID', 'CreateTime', 'EndTime') + list_filter = ('TransactionType', 'TransactionStatus') + search_fields = ('TransactionCode', 'TransactionName') + readonly_fields = ('TransactionUUID', 'CreateTime') + + +@admin.register(TransactionStep) +class TransactionStepAdmin(admin.ModelAdmin): + list_display = ('StepUUID', 'TransactionUUID', 'StepIndex', 'StepName', 'StepStatus', 'ParticipantServiceUUID', 'RetryCount') + list_filter = ('StepStatus',) + readonly_fields = ('StepUUID', 'CreateTime') diff --git a/msync/apps.py b/msync/apps.py new file mode 100644 index 0000000..277479a --- /dev/null +++ b/msync/apps.py @@ -0,0 +1,68 @@ +from django.apps import AppConfig + + +class MsyncConfig(AppConfig): + default_auto_field = 'django.db.models.BigAutoField' + name = 'gvsdsdk.msync' + verbose_name = '微服务通信' + verbose_name_plural = '微服务通信' + + def ready(self): + # 仅在主进程启动时确保 gvsds-master 节点存在 + import os + if os.environ.get('RUN_MAIN') == 'true' or not os.environ.get('RUN_MAIN'): + try: + self._ensure_master_node() + except Exception: + pass # 数据库可能尚未迁移 + + def _ensure_master_node(self): + from gvsdsdk.msync.models import ServiceNode, _utcnow + import uuid + + service_domain = self._resolve_service_domain() + master_uuid = uuid.UUID('00000000-0000-0000-0000-000000000001') + + try: + node = ServiceNode.objects.get(ServiceName='gvsds-master') + # 更新心跳、状态和元数据 + node.ServiceStatus = 1 + node.ServiceScope = ['*'] + node.ServiceDomain = service_domain + node.ServiceType = 1 + node.ServiceMetadata = {'DBName': 'Users'} + node.LastHeartbeat = _utcnow() + node.save(update_fields=['ServiceStatus', 'ServiceScope', 'ServiceDomain', 'ServiceType', 'ServiceMetadata', 'LastHeartbeat']) + except ServiceNode.DoesNotExist: + import bcrypt + secret = 'gvsds-master-internal' + secret_hash = bcrypt.hashpw(secret.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') + node = ServiceNode( + ServiceUUID=master_uuid.bytes, + ServiceName='gvsds-master', + ServiceDomain=service_domain, + ServiceSecretHash=secret_hash, + ServiceScope=['*'], + ServiceStatus=1, + ServiceType=1, + ServiceMetadata={'DBName': 'Users'}, + LastHeartbeat=_utcnow(), + ) + node.SetSecret(secret) + node.save() + + @staticmethod + def _resolve_service_domain(): + """解析主服务器公开域名：环境变量 → settings → 默认值（不依赖 gvsds）""" + import os + domain = os.environ.get('MSYNC_SERVICE_DOMAIN') or os.environ.get('APP_MSYNC_SERVICE_DOMAIN') + if domain: + return domain + try: + from django.conf import settings as _settings + domain = getattr(_settings, 'GVSDSDK_MSYNC_MASTER_URL', None) + if domain: + return str(domain).rstrip('/') + except Exception: + pass + return 'https://www.gvsds.com' diff --git a/msync/auth.py b/msync/auth.py index 61cca98..89888fb 100644 --- a/msync/auth.py +++ b/msync/auth.py @@ -1,8 +1,8 @@ """MSYNC 认证模块 -提供 HMAC-SHA256 签名验证、服务令牌编解码等纯函数。 -Django ORM 依赖的函数（AuthenticateService, IssueServiceToken 等） -仍在 gvsds.apps.msync.auth 中，由服务端调用。 +提供 HMAC-SHA256 签名验证、服务令牌编解码等纯函数， +以及 ORM 依赖的服务端认证函数（AuthenticateService, IssueServiceToken 等， +模型延迟导入，主服务器侧使用）。 """ import hmac @@ -13,6 +13,7 @@ import base64 import binascii import uuid import logging +import traceback import datetime logger = logging.getLogger(__name__) @@ -140,3 +141,153 @@ class ServiceTokenEncoder: return Payload except (ValueError, TypeError, json.JSONDecodeError, binascii.Error): return None + + +# ─── ORM 依赖的认证函数（主服务器侧使用，模型延迟导入）────────────── + +def _get_master_secret(): + """获取 MASTER_SECRET：环境变量 MSYNC_MASTER_SECRET 优先，其次 secrets.ini""" + from gvsdsdk.secrets import get as _secret_get + secret = _secret_get('msync', 'MASTER_SECRET', fallback=None) + if not secret: + from django.core.exceptions import ImproperlyConfigured + raise ImproperlyConfigured('msync MASTER_SECRET 未配置，请在 secrets.ini 或环境变量 MSYNC_MASTER_SECRET 中设置') + return secret + + +def IssueServiceToken(service_node, access_lifetime_hours=2, refresh_lifetime_days=30): + """为服务节点签发令牌对（AccessToken + RefreshToken）""" + from gvsdsdk.msync.models import ServiceToken + + master_secret = _get_master_secret() + + now_aware = datetime.datetime.now(datetime.timezone.utc) + access_expires_aware = now_aware + datetime.timedelta(hours=access_lifetime_hours) + refresh_expires_aware = now_aware + datetime.timedelta(days=refresh_lifetime_days) + + now_naive = now_aware.replace(tzinfo=None) + access_expires = access_expires_aware.replace(tzinfo=None) + refresh_expires = refresh_expires_aware.replace(tzinfo=None) + + service_scope = service_node.ServiceScope + if service_scope is None: + service_scope = ['ogm:read', 'ogm:write', 'registry:read'] + elif isinstance(service_scope, str): + service_scope = [service_scope] + + encoder = ServiceTokenEncoder(master_secret) + access_token = encoder.Encode( + service_name=service_node.ServiceName, + scope=service_scope, + expires_at=access_expires_aware, + ) + refresh_token = encoder.Encode( + service_name=service_node.ServiceName, + scope=service_scope, + expires_at=refresh_expires_aware, + ) + + import uuid + token = ServiceToken( + TokenUUID=uuid.uuid4().bytes, + ServiceUUID=service_node, + AccessToken=access_token, + RefreshToken=refresh_token, + TokenScope=service_scope, + ExpiresAt=access_expires, + RefreshExpiresAt=refresh_expires, + ) + token.save() + return token + + +def AuthenticateService(service_name, signed_payload, signature): + """验证服务身份并签发令牌""" + from gvsdsdk.msync.models import ServiceNode, ServiceToken + + try: + node = ServiceNode.objects.get(ServiceName=service_name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + logger.error(f'[AUTH] 服务节点不存在或未激活: {service_name}') + return None + except Exception as e: + logger.error(f'查询服务节点异常 [{service_name}]: {e}\n{traceback.format_exc()}') + return None + + import time + ts = signed_payload.get('ts', 0) + now = time.time() + ts_diff = abs(now - ts) + if ts_diff > 300: + logger.error(f'[AUTH] 时间戳校验失败: ts={ts}, now={now}, diff={ts_diff:.1f}s') + return None + + raw_secret = node.GetSecret() + if raw_secret is None: + logger.error(f'服务 {service_name} 密钥加密副本缺失，请重新注册') + return None + + if not VerifyAuthPayload(signed_payload, signature, raw_secret): + logger.error(f'[AUTH] 签名验证失败: service={service_name}, payload={signed_payload}') + return None + + try: + ServiceToken.objects.filter(ServiceUUID=node, TokenStatus=1).update(TokenStatus=0) + except Exception as e: + logger.error(f'撤销旧令牌异常 [{service_name}]: {e}') + + try: + return IssueServiceToken(node) + except Exception as e: + logger.error(f'签发服务令牌异常 [{service_name}]: {e}\n{traceback.format_exc()}') + return None + + +def ValidateAccessToken(access_token): + """验证 AccessToken 并返回载荷信息""" + from gvsdsdk.msync.models import ServiceToken + + master_secret = _get_master_secret() + encoder = ServiceTokenEncoder(master_secret) + payload = encoder.Decode(access_token) + if payload is None: + return None + + try: + token_record = ServiceToken.objects.get(AccessToken=access_token, TokenStatus=1) + if token_record.IsExpired(): + token_record.Revoke() + return None + except ServiceToken.DoesNotExist: + return None + + return payload + + +def RefreshServiceToken(refresh_token): + """使用 RefreshToken 刷新令牌对""" + from gvsdsdk.msync.models import ServiceToken, ServiceNode + + master_secret = _get_master_secret() + encoder = ServiceTokenEncoder(master_secret) + payload = encoder.Decode(refresh_token) + if payload is None: + return None + + try: + old_token = ServiceToken.objects.get(RefreshToken=refresh_token, TokenStatus=1) + except ServiceToken.DoesNotExist: + return None + + if old_token.IsRefreshExpired(): + old_token.Revoke() + return None + + service_name = payload.get('sub') + try: + node = ServiceNode.objects.get(ServiceName=service_name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + return None + + old_token.Revoke() + return IssueServiceToken(node) diff --git a/msync/middleware.py b/msync/middleware.py new file mode 100644 index 0000000..f10368a --- /dev/null +++ b/msync/middleware.py @@ -0,0 +1,154 @@ +"""模块：服务请求验证中间件与权限装饰器 +用于验证来自其他微服务的请求，提取服务身份信息。 +在 Django 视图中使用 @require_service_scope 装饰器保护端点。""" + +import functools +import logging +from django.http import JsonResponse +from rest_framework.response import Response +from gvsdsdk.msync.auth import ValidateAccessToken + +logger = logging.getLogger(__name__) + + +class ServiceAuthMiddleware: + """服务认证中间件——从请求中提取并验证服务令牌 + + 将验证结果附加到 request 对象： + - request.service_name: 发起请求的服务名称 + - request.service_scope: 服务的权限范围列表 + - request.service_token: 原始令牌字符串 + """ + + SERVICE_AUTH_HEADER = 'Authorization' + SERVICE_AUTH_PREFIX = 'Service' + + def __init__(self, get_response): + self.get_response = get_response + + def __call__(self, request): + request.service_name = None + request.service_scope = [] + request.service_token = None + + auth_header = request.META.get('HTTP_AUTHORIZATION', '') + if auth_header.startswith(f'{self.SERVICE_AUTH_PREFIX} '): + token = auth_header[len(self.SERVICE_AUTH_PREFIX) + 1:] + payload = ValidateAccessToken(token) + if payload: + request.service_name = payload.get('sub') + request.service_scope = payload.get('scope', []) + request.service_token = token + + response = self.get_response(request) + return response + + +def require_service_scope(*required_scopes): + """视图装饰器——要求请求来自具有指定权限范围的已认证服务 + + 用法: + @require_service_scope('ogm:read') + def my_view(request): + # request.service_name 已设置 + ... + """ + + def decorator(view_func): + @functools.wraps(view_func) + def wrapper(request, *args, **kwargs): + if not request.service_name: + return JsonResponse( + {'code': 401, 'msg': '需要服务认证', 'data': None}, + status=401, + ) + + if '*' not in request.service_scope: + for scope in required_scopes: + if scope not in request.service_scope: + return JsonResponse( + {'code': 403, 'msg': f'缺少权限: {scope}', 'data': None}, + status=403, + ) + + return view_func(request, *args, **kwargs) + + return wrapper + + return decorator + + +def require_service_scope_drq(*required_scopes): + """DRF 视图装饰器——返回 DRF Response 而非 Django JsonResponse""" + + def decorator(view_func): + @functools.wraps(view_func) + def wrapper(self, request, *args, **kwargs): + if not request.service_name: + return Response( + {'code': 401, 'msg': '需要服务认证', 'data': None}, + status=401, + ) + + if '*' not in request.service_scope: + for scope in required_scopes: + if scope not in request.service_scope: + return Response( + {'code': 403, 'msg': f'缺少权限: {scope}', 'data': None}, + status=403, + ) + + return view_func(self, request, *args, **kwargs) + + return wrapper + + return decorator + + +class ServicePermission: + """DRF 权限类——用于 ViewSet 的 permission_classes""" + + def __init__(self, *required_scopes): + self.required_scopes = required_scopes + + def has_permission(self, request, view): + if not request.service_name: + return False + if '*' in request.service_scope: + return True + return all(s in request.service_scope for s in self.required_scopes) + + +class ServiceTokenAuthentication: + """DRF 认证类——从请求头中提取并验证服务令牌 + + 配合 ServiceAuthMiddleware 使用，中间件已将验证结果附加到 request 对象。 + 此类使 DRF 的 request.user 和 request.auth 可用于权限检查。 + """ + + def authenticate(self, request): + service_name = getattr(request, 'service_name', None) + if not service_name: + return None + # 返回 (user, auth) 元组，使 DRF 认为请求已认证 + # service_name 作为 user 标识，service_token 作为 auth + return (ServicePrincipal(request), request.service_token) + + def authenticate_header(self, request): + return 'Service' + + +class ServicePrincipal: + """服务主体——作为 DRF 认证后的 request.user""" + + def __init__(self, request): + self.service_name = request.service_name + self.service_scope = request.service_scope + self.service_token = request.service_token + + @property + def is_authenticated(self): + return True + + def __str__(self): + return f'Service:{self.service_name}' diff --git a/msync/migrations/0001_initial.py b/msync/migrations/0001_initial.py new file mode 100644 index 0000000..1002172 --- /dev/null +++ b/msync/migrations/0001_initial.py @@ -0,0 +1,151 @@ +import datetime +from django.db import migrations, models +import django.db.models.deletion +import gvsdsdk.models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [] + + operations = [ + migrations.CreateModel( + name='ServiceNode', + fields=[ + ('ServiceUUID', gvsdsdk.models.UUIDField(verbose_name='服务UUID', editable=False, primary_key=True, serialize=False, binary_length=16)), + ('ServiceName', models.CharField(db_column='ServiceName', max_length=128, unique=True)), + ('ServiceDomain', models.CharField(db_column='ServiceDomain', max_length=256)), + ('ServiceSecretHash', models.CharField(db_column='ServiceSecretHash', max_length=128)), + ('ServiceScope', models.JSONField(db_column='ServiceScope', default=list)), + ('ServiceStatus', models.SmallIntegerField(db_column='ServiceStatus', default=1)), + ('ServiceType', models.SmallIntegerField(db_column='ServiceType', default=0)), + ('ServiceMetadata', models.JSONField(db_column='ServiceMetadata', default=dict)), + ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)), + ('LastHeartbeat', models.DateTimeField(blank=True, db_column='LastHeartbeat', null=True)), + ], + options={ + 'verbose_name': '服务节点', + 'verbose_name_plural': '服务节点', + 'db_table': 'ServiceNode', + 'ordering': ['-CreateTime'], + }, + ), + migrations.CreateModel( + name='ServiceToken', + fields=[ + ('TokenUUID', gvsdsdk.models.UUIDField(verbose_name='令牌UUID', editable=False, primary_key=True, serialize=False, binary_length=16)), + ('AccessToken', models.CharField(db_column='AccessToken', max_length=512, unique=True)), + ('RefreshToken', models.CharField(db_column='RefreshToken', max_length=512, unique=True)), + ('TokenScope', models.JSONField(db_column='TokenScope', default=list)), + ('ExpiresAt', models.DateTimeField(db_column='ExpiresAt')), + ('RefreshExpiresAt', models.DateTimeField(db_column='RefreshExpiresAt')), + ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)), + ('TokenStatus', models.SmallIntegerField(db_column='TokenStatus', default=1)), + ('ServiceUUID', models.ForeignKey(db_column='ServiceUUID', on_delete=django.db.models.deletion.CASCADE, related_name='tokens', to='msync.servicenode')), + ], + options={ + 'verbose_name': '服务令牌', + 'verbose_name_plural': '服务令牌', + 'db_table': 'ServiceToken', + 'ordering': ['-CreateTime'], + }, + ), + migrations.CreateModel( + name='ObjectGlobalMapping', + fields=[ + ('OGMUUID', gvsdsdk.models.UUIDField(verbose_name='OGM UUID', editable=False, primary_key=True, serialize=False, binary_length=16)), + ('ObjectType', models.CharField(db_column='ObjectType', max_length=64)), + ('ObjectLocalID', models.CharField(db_column='ObjectLocalID', max_length=128)), + ('ObjectGlobalID', models.CharField(db_column='ObjectGlobalID', max_length=128, unique=True)), + ('ObjectMetadata', models.JSONField(db_column='ObjectMetadata', default=dict)), + ('ObjectStatus', models.SmallIntegerField(db_column='ObjectStatus', default=1)), + ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)), + ('UpdateTime', models.DateTimeField(blank=True, db_column='UpdateTime', null=True)), + ('OwnerServiceUUID', models.ForeignKey(db_column='OwnerServiceUUID', on_delete=django.db.models.deletion.CASCADE, related_name='owned_objects', to='msync.servicenode')), + ], + options={ + 'verbose_name': '对象全局映射', + 'verbose_name_plural': '对象全局映射', + 'db_table': 'OGM', + 'ordering': ['-CreateTime'], + 'unique_together': {('ObjectType', 'ObjectLocalID', 'OwnerServiceUUID')}, + }, + ), + migrations.AddIndex( + model_name='objectglobalmapping', + index=models.Index(fields=['ObjectType'], name='idx_ogm_type'), + ), + migrations.AddIndex( + model_name='objectglobalmapping', + index=models.Index(fields=['ObjectGlobalID'], name='idx_ogm_gid'), + ), + migrations.CreateModel( + name='DistributedTransaction', + fields=[ + ('TransactionUUID', gvsdsdk.models.UUIDField(verbose_name='事务UUID', editable=False, primary_key=True, serialize=False, binary_length=16)), + ('TransactionCode', models.CharField(db_column='TransactionCode', max_length=64, unique=True)), + ('TransactionType', models.CharField(choices=[('saga', 'Saga'), ('2pc', '两阶段提交')], db_column='TransactionType', default='saga', max_length=16)), + ('TransactionName', models.CharField(db_column='TransactionName', max_length=128)), + ('TransactionDescription', models.CharField(db_column='TransactionDescription', default='', max_length=512)), + ('TransactionStatus', models.SmallIntegerField(choices=[(0, '待执行'), (1, '执行中'), (2, '已提交'), (3, '补偿中'), (4, '已补偿'), (5, '失败'), (6, '超时')], db_column='TransactionStatus', default=0)), + ('ReferenceOGMUUID', gvsdsdk.models.UUIDField(verbose_name='关联OGM UUID', blank=True, null=True, binary_length=16)), + ('TransactionPayload', models.JSONField(db_column='TransactionPayload', default=dict)), + ('TransactionResult', models.JSONField(db_column='TransactionResult', default=dict)), + ('TimeoutSeconds', models.IntegerField(db_column='TimeoutSeconds', default=300)), + ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)), + ('UpdateTime', models.DateTimeField(blank=True, db_column='UpdateTime', null=True)), + ('EndTime', models.DateTimeField(blank=True, db_column='EndTime', null=True)), + ('InitiatorServiceUUID', models.ForeignKey(blank=True, db_column='InitiatorServiceUUID', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='initiated_dtx', to='msync.servicenode')), + ], + options={ + 'verbose_name': '分布式事务', + 'verbose_name_plural': '分布式事务', + 'db_table': 'DistributedTransaction', + 'ordering': ['-CreateTime'], + }, + ), + migrations.AddIndex( + model_name='distributedtransaction', + index=models.Index(fields=['TransactionStatus'], name='idx_dtx_status'), + ), + migrations.AddIndex( + model_name='distributedtransaction', + index=models.Index(fields=['InitiatorServiceUUID'], name='idx_dtx_initiator'), + ), + migrations.CreateModel( + name='TransactionStep', + fields=[ + ('StepUUID', gvsdsdk.models.UUIDField(verbose_name='步骤UUID', editable=False, primary_key=True, serialize=False, binary_length=16)), + ('StepIndex', models.IntegerField(db_column='StepIndex')), + ('StepName', models.CharField(db_column='StepName', max_length=128)), + ('StepDescription', models.CharField(db_column='StepDescription', default='', max_length=512)), + ('StepStatus', models.SmallIntegerField(choices=[(0, '待执行'), (1, '执行中'), (2, '已完成'), (3, '补偿中'), (4, '已补偿'), (5, '失败'), (6, '已跳过')], db_column='StepStatus', default=0)), + ('ActionEndpoint', models.CharField(db_column='ActionEndpoint', max_length=512)), + ('ActionMethod', models.CharField(db_column='ActionMethod', default='POST', max_length=8)), + ('ActionPayload', models.JSONField(db_column='ActionPayload', default=dict)), + ('CompensateEndpoint', models.CharField(db_column='CompensateEndpoint', default='', max_length=512)), + ('CompensateMethod', models.CharField(db_column='CompensateMethod', default='POST', max_length=8)), + ('CompensatePayload', models.JSONField(db_column='CompensatePayload', default=dict)), + ('StepResult', models.JSONField(db_column='StepResult', default=dict)), + ('RetryCount', models.IntegerField(db_column='RetryCount', default=0)), + ('MaxRetries', models.IntegerField(db_column='MaxRetries', default=3)), + ('CreateTime', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)), + ('StartTime', models.DateTimeField(blank=True, db_column='StartTime', null=True)), + ('EndTime', models.DateTimeField(blank=True, db_column='EndTime', null=True)), + ('ParticipantServiceUUID', models.ForeignKey(blank=True, db_column='ParticipantServiceUUID', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='participated_steps', to='msync.servicenode')), + ('TransactionUUID', models.ForeignKey(db_column='TransactionUUID', on_delete=django.db.models.deletion.CASCADE, related_name='steps', to='msync.distributedtransaction')), + ], + options={ + 'verbose_name': '事务步骤', + 'verbose_name_plural': '事务步骤', + 'db_table': 'TransactionStep', + 'ordering': ['TransactionUUID', 'StepIndex'], + }, + ), + migrations.AddIndex( + model_name='transactionstep', + index=models.Index(fields=['TransactionUUID', 'StepIndex'], name='idx_step_tx_order'), + ), + ] diff --git a/msync/migrations/0002_add_service_secret_encrypted.py b/msync/migrations/0002_add_service_secret_encrypted.py new file mode 100644 index 0000000..b16f871 --- /dev/null +++ b/msync/migrations/0002_add_service_secret_encrypted.py @@ -0,0 +1,16 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('msync', '0001_initial'), + ] + + operations = [ + migrations.AddField( + model_name='servicenode', + name='ServiceSecretEncrypted', + field=models.CharField(db_column='ServiceSecretEncrypted', default='', max_length=512), + ), + ] diff --git a/msync/migrations/__init__.py b/msync/migrations/__init__.py new file mode 100644 index 0000000..40a96af --- /dev/null +++ b/msync/migrations/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 -*- diff --git a/msync/migrations/__pycache__/0001_initial.cpython-312.pyc b/msync/migrations/__pycache__/0001_initial.cpython-312.pyc new file mode 100644 index 0000000..8853c97 Binary files /dev/null and b/msync/migrations/__pycache__/0001_initial.cpython-312.pyc differ diff --git a/msync/migrations/__pycache__/0002_add_service_secret_encrypted.cpython-312.pyc b/msync/migrations/__pycache__/0002_add_service_secret_encrypted.cpython-312.pyc new file mode 100644 index 0000000..2358b88 Binary files /dev/null and b/msync/migrations/__pycache__/0002_add_service_secret_encrypted.cpython-312.pyc differ diff --git a/msync/migrations/__pycache__/__init__.cpython-312.pyc b/msync/migrations/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..81b167e Binary files /dev/null and b/msync/migrations/__pycache__/__init__.cpython-312.pyc differ diff --git a/msync/models.py b/msync/models.py new file mode 100644 index 0000000..0932529 --- /dev/null +++ b/msync/models.py @@ -0,0 +1,317 @@ +"""模块：MSync 数据模型 +微服务通信核心表：服务节点、服务令牌、对象全局映射（OGM）、分布式事务。 +所有 ORM 字段与方法遵循帕斯卡命名法。 + +模型归属 app_label='msync'，与 gvsds 主服务器 7 库路由及既有迁移兼容。""" + +import uuid +import datetime +from django.db import models +from gvsdsdk.model_base import QModel +from gvsdsdk.models import UUIDField + + +def _utcnow(): + """返回 naive UTC datetime（兼容 Django USE_TZ=False）""" + return datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + + +class ServiceNode(QModel): + """服务节点表——注册的微服务实例（账本服务器存储的所有分布式服务地址）""" + + ServiceUUID = UUIDField(primary_key=True, verbose_name='服务UUID') + ServiceName = models.CharField(max_length=128, unique=True, verbose_name='服务名称') + ServiceDomain = models.CharField(max_length=256, verbose_name='服务域名') + ServiceSecretHash = models.CharField(max_length=128, verbose_name='密钥哈希') + ServiceSecretEncrypted = models.CharField(max_length=512, default='', verbose_name='加密密钥') + ServiceScope = models.JSONField(default=list, verbose_name='服务权限') + ServiceStatus = models.SmallIntegerField(default=1, verbose_name='服务状态') + ServiceType = models.SmallIntegerField(default=0, verbose_name='服务类型') + ServiceMetadata = models.JSONField(default=dict, verbose_name='服务元数据') + CreateTime = models.DateTimeField(default=_utcnow, verbose_name='创建时间') + LastHeartbeat = models.DateTimeField(null=True, blank=True, verbose_name='最后心跳') + + class Meta: + db_table = 'ServiceNode' + app_label = 'msync' + verbose_name = '服务节点' + verbose_name_plural = '服务节点' + ordering = ['-CreateTime'] + + def __str__(self): + return self.ServiceName + + @property + def IsMaster(self): + return self.ServiceType == 1 + + @property + def IsActive(self): + return self.ServiceStatus == 1 + + def VerifySecret(self, raw_secret): + import bcrypt + try: + return bcrypt.checkpw( + raw_secret.encode('utf-8'), + self.ServiceSecretHash.encode('utf-8') + ) + except Exception: + return False + + def SetSecret(self, raw_secret): + from gvsdsdk.auth.crypto import SymmetricEncrypt + from gvsdsdk.secrets import get as _secret_get + master = _secret_get('msync', 'MASTER_SECRET', fallback='gvsds-msync-default-secret') + self.ServiceSecretEncrypted = SymmetricEncrypt(raw_secret, master) + + def GetSecret(self): + if not self.ServiceSecretEncrypted: + return None + from gvsdsdk.auth.crypto import SymmetricDecrypt + from gvsdsdk.secrets import get as _secret_get + master = _secret_get('msync', 'MASTER_SECRET', fallback='gvsds-msync-default-secret') + try: + return SymmetricDecrypt(self.ServiceSecretEncrypted, master) + except Exception: + return None + + def HasScope(self, scope_name): + if not self.ServiceScope: + return False + return scope_name in self.ServiceScope or '*' in self.ServiceScope + + +class ServiceToken(QModel): + """服务令牌表——服务间认证凭据""" + + TokenUUID = UUIDField(primary_key=True, verbose_name='令牌UUID') + ServiceUUID = models.ForeignKey( + ServiceNode, on_delete=models.CASCADE, + related_name='tokens', verbose_name='服务UUID', + ) + AccessToken = models.CharField(max_length=512, unique=True, verbose_name='访问令牌') + RefreshToken = models.CharField(max_length=512, unique=True, verbose_name='刷新令牌') + TokenScope = models.JSONField(default=list, verbose_name='令牌权限') + ExpiresAt = models.DateTimeField(verbose_name='过期时间') + RefreshExpiresAt = models.DateTimeField(verbose_name='刷新过期时间') + CreateTime = models.DateTimeField(default=_utcnow, verbose_name='创建时间') + TokenStatus = models.SmallIntegerField(default=1, verbose_name='令牌状态') + + class Meta: + db_table = 'ServiceToken' + app_label = 'msync' + verbose_name = '服务令牌' + verbose_name_plural = '服务令牌' + ordering = ['-CreateTime'] + + def IsExpired(self): + return _utcnow() > self.ExpiresAt.replace(tzinfo=None) + + def IsRefreshExpired(self): + return _utcnow() > self.RefreshExpiresAt.replace(tzinfo=None) + + def IsValid(self): + return self.TokenStatus == 1 and not self.IsExpired() + + def Revoke(self): + self.TokenStatus = 0 + self.save(update_fields=['TokenStatus']) + + +class ObjectGlobalMapping(QModel): + """对象全局映射表——跨服务对象引用注册""" + + OGMUUID = UUIDField(primary_key=True, verbose_name='OGM UUID') + ObjectType = models.CharField(max_length=64, verbose_name='对象类型') + ObjectLocalID = models.CharField(max_length=128, verbose_name='本地ID') + ObjectGlobalID = models.CharField(max_length=128, unique=True, db_index=True, verbose_name='全局ID') + OwnerServiceUUID = models.ForeignKey( + ServiceNode, on_delete=models.CASCADE, + related_name='owned_objects', verbose_name='所属服务UUID', + ) + ObjectMetadata = models.JSONField(default=dict, verbose_name='对象元数据') + ObjectStatus = models.SmallIntegerField(default=1, verbose_name='对象状态') + CreateTime = models.DateTimeField(default=_utcnow, verbose_name='创建时间') + UpdateTime = models.DateTimeField(null=True, blank=True, verbose_name='更新时间') + + class Meta: + db_table = 'OGM' + app_label = 'msync' + verbose_name = '对象全局映射' + verbose_name_plural = '对象全局映射' + unique_together = [('ObjectType', 'ObjectLocalID', 'OwnerServiceUUID')] + ordering = ['-CreateTime'] + indexes = [ + models.Index(fields=['ObjectType'], name='idx_ogm_type'), + models.Index(fields=['ObjectGlobalID'], name='idx_ogm_gid'), + ] + + def __str__(self): + return f'{self.ObjectType}:{self.ObjectGlobalID}' + + @property + def IsAlive(self): + return self.ObjectStatus == 1 + + def Touch(self): + self.UpdateTime = _utcnow() + self.save(update_fields=['UpdateTime']) + + +class DistributedTransaction(QModel): + """分布式事务表——跨服务事务编排（Saga 模式）""" + + TX_STATUS_PENDING = 0 + TX_STATUS_RUNNING = 1 + TX_STATUS_COMMITTED = 2 + TX_STATUS_COMPENSATING = 3 + TX_STATUS_COMPENSATED = 4 + TX_STATUS_FAILED = 5 + TX_STATUS_TIMED_OUT = 6 + + TX_STATUS_CHOICES = [ + (TX_STATUS_PENDING, '待执行'), + (TX_STATUS_RUNNING, '执行中'), + (TX_STATUS_COMMITTED, '已提交'), + (TX_STATUS_COMPENSATING, '补偿中'), + (TX_STATUS_COMPENSATED, '已补偿'), + (TX_STATUS_FAILED, '失败'), + (TX_STATUS_TIMED_OUT, '超时'), + ] + + TX_TYPE_SAGA = 'saga' + TX_TYPE_TWO_PHASE = '2pc' + + TX_TYPE_CHOICES = [ + (TX_TYPE_SAGA, 'Saga'), + (TX_TYPE_TWO_PHASE, '两阶段提交'), + ] + + TransactionUUID = UUIDField(primary_key=True, verbose_name='事务UUID') + TransactionCode = models.CharField(max_length=64, unique=True, verbose_name='事务编号') + TransactionType = models.CharField(max_length=16, choices=TX_TYPE_CHOICES, default=TX_TYPE_SAGA, verbose_name='事务类型') + TransactionName = models.CharField(max_length=128, verbose_name='事务名称') + TransactionDescription = models.CharField(max_length=512, default='', verbose_name='事务描述') + TransactionStatus = models.SmallIntegerField( + default=TX_STATUS_PENDING, choices=TX_STATUS_CHOICES, verbose_name='事务状态', + ) + InitiatorServiceUUID = models.ForeignKey( + ServiceNode, on_delete=models.SET_NULL, + null=True, blank=True, related_name='initiated_dtx', verbose_name='发起服务UUID', + ) + ReferenceOGMUUID = UUIDField(null=True, blank=True, verbose_name='关联OGM UUID') + TransactionPayload = models.JSONField(default=dict, verbose_name='事务载荷') + TransactionResult = models.JSONField(default=dict, verbose_name='事务结果') + TimeoutSeconds = models.IntegerField(default=300, verbose_name='超时秒数') + CreateTime = models.DateTimeField(default=_utcnow, verbose_name='创建时间') + UpdateTime = models.DateTimeField(null=True, blank=True, verbose_name='更新时间') + EndTime = models.DateTimeField(null=True, blank=True, verbose_name='结束时间') + + class Meta: + db_table = 'DistributedTransaction' + app_label = 'msync' + verbose_name = '分布式事务' + verbose_name_plural = '分布式事务' + ordering = ['-CreateTime'] + indexes = [ + models.Index(fields=['TransactionStatus'], name='idx_dtx_status'), + models.Index(fields=['InitiatorServiceUUID'], name='idx_dtx_initiator'), + ] + + def __str__(self): + return self.TransactionCode + + @property + def IsTerminal(self): + return self.TransactionStatus in ( + self.TX_STATUS_COMMITTED, + self.TX_STATUS_COMPENSATED, + self.TX_STATUS_FAILED, + self.TX_STATUS_TIMED_OUT, + ) + + @property + def IsTimedOut(self): + if self.TransactionStatus in (self.TX_STATUS_COMMITTED, self.TX_STATUS_COMPENSATED): + return False + if not self.TimeoutSeconds or not self.CreateTime: + return False + elapsed = (_utcnow() - self.CreateTime.replace(tzinfo=None)).total_seconds() + return elapsed > self.TimeoutSeconds + + +class TransactionStep(QModel): + """事务步骤表——Saga 编排中的每一步""" + + STEP_STATUS_PENDING = 0 + STEP_STATUS_RUNNING = 1 + STEP_STATUS_COMPLETED = 2 + STEP_STATUS_COMPENSATING = 3 + STEP_STATUS_COMPENSATED = 4 + STEP_STATUS_FAILED = 5 + STEP_STATUS_SKIPPED = 6 + + STEP_STATUS_CHOICES = [ + (STEP_STATUS_PENDING, '待执行'), + (STEP_STATUS_RUNNING, '执行中'), + (STEP_STATUS_COMPLETED, '已完成'), + (STEP_STATUS_COMPENSATING, '补偿中'), + (STEP_STATUS_COMPENSATED, '已补偿'), + (STEP_STATUS_FAILED, '失败'), + (STEP_STATUS_SKIPPED, '已跳过'), + ] + + StepUUID = UUIDField(primary_key=True, verbose_name='步骤UUID') + TransactionUUID = models.ForeignKey( + DistributedTransaction, on_delete=models.CASCADE, + related_name='steps', verbose_name='事务UUID', + ) + StepIndex = models.IntegerField(verbose_name='步骤序号') + StepName = models.CharField(max_length=128, verbose_name='步骤名称') + StepDescription = models.CharField(max_length=512, default='', verbose_name='步骤描述') + ParticipantServiceUUID = models.ForeignKey( + ServiceNode, on_delete=models.SET_NULL, + null=True, blank=True, related_name='participated_steps', verbose_name='参与服务UUID', + ) + StepStatus = models.SmallIntegerField( + default=STEP_STATUS_PENDING, choices=STEP_STATUS_CHOICES, verbose_name='步骤状态', + ) + ActionEndpoint = models.CharField(max_length=512, verbose_name='执行端点') + ActionMethod = models.CharField(max_length=8, default='POST', verbose_name='执行方法') + ActionPayload = models.JSONField(default=dict, verbose_name='执行载荷') + CompensateEndpoint = models.CharField(max_length=512, default='', verbose_name='补偿端点') + CompensateMethod = models.CharField(max_length=8, default='POST', verbose_name='补偿方法') + CompensatePayload = models.JSONField(default=dict, verbose_name='补偿载荷') + StepResult = models.JSONField(default=dict, verbose_name='步骤结果') + RetryCount = models.IntegerField(default=0, verbose_name='重试次数') + MaxRetries = models.IntegerField(default=3, verbose_name='最大重试') + CreateTime = models.DateTimeField(default=_utcnow, verbose_name='创建时间') + StartTime = models.DateTimeField(null=True, blank=True, verbose_name='开始时间') + EndTime = models.DateTimeField(null=True, blank=True, verbose_name='结束时间') + + class Meta: + db_table = 'TransactionStep' + app_label = 'msync' + verbose_name = '事务步骤' + verbose_name_plural = '事务步骤' + ordering = ['TransactionUUID', 'StepIndex'] + indexes = [ + models.Index(fields=['TransactionUUID', 'StepIndex'], name='idx_step_tx_order'), + ] + + def __str__(self): + return f'{self.StepIndex}: {self.StepName}' + + @property + def CanRetry(self): + return self.RetryCount < self.MaxRetries + + @property + def IsTerminal(self): + return self.StepStatus in ( + self.STEP_STATUS_COMPLETED, + self.STEP_STATUS_COMPENSATED, + self.STEP_STATUS_FAILED, + self.STEP_STATUS_SKIPPED, + ) diff --git a/msync/ogm.py b/msync/ogm.py index 3f0d566..702800b 100644 --- a/msync/ogm.py +++ b/msync/ogm.py @@ -1,17 +1,267 @@ -"""MSYNC OGM（对象全局映射）模块 +"""模块：对象全局映射（OGM）管理 +提供跨服务对象引用的注册、解析、更新、注销操作。 +OGM 是微服务间共享对象身份的核心机制： + - 每个本地对象注册后获得全局唯一 ID + - 其他服务通过全局 ID 解析到对象归属与本地 ID + - 支持按类型、按服务查询对象映射""" -薄封装：直接转发到 gvsds.apps.msync.ogm 原生层。 -""" +import uuid +import datetime +import logging +from gvsdsdk.msync.models import ObjectGlobalMapping, ServiceNode, _utcnow -try: - from gvsds.apps.msync.ogm import OGMManager, OGMError -except ImportError: - raise ImportError( - 'OGMManager 仅在主服务器环境可用。' - '子服务器请通过 ServiceClient.ResolveOGM() 使用 HTTP 转发。' - ) +logger = logging.getLogger(__name__) -__all__ = [ - 'OGMManager', - 'OGMError', -] + +class OGMError(Exception): + pass + + +class OGMManager: + """OGM 管理器——对象全局映射的增删改查""" + + @staticmethod + def Register(object_type, local_id, service_name, global_id=None, metadata=None): + """注册对象全局映射 + + Args: + object_type: 对象类型（如 'user', 'company', 'asset'） + local_id: 对象在所属服务中的本地 ID + service_name: 所属服务名称 + global_id: 自定义全局 ID（不指定则自动生成） + metadata: 对象元数据 + + Returns: + ObjectGlobalMapping 实例 + + Raises: + OGMError: 注册失败 + """ + try: + service = ServiceNode.objects.get(ServiceName=service_name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + raise OGMError(f'服务不存在或已禁用: {service_name}') + + if ObjectGlobalMapping.objects.filter( + ObjectType=object_type, + ObjectLocalID=local_id, + OwnerServiceUUID=service, + ).exists(): + raise OGMError(f'对象映射已存在: {object_type}:{local_id}@{service_name}') + + if global_id is None: + global_id = f'{object_type}:{uuid.uuid4().hex[:16]}' + + if ObjectGlobalMapping.objects.filter(ObjectGlobalID=global_id).exists(): + raise OGMError(f'全局 ID 已被占用: {global_id}') + + ogm = ObjectGlobalMapping( + OGMUUID=uuid.uuid4().bytes, + ObjectType=object_type, + ObjectLocalID=str(local_id), + ObjectGlobalID=global_id, + OwnerServiceUUID=service, + ObjectMetadata=metadata or {}, + ObjectStatus=1, + ) + ogm.save() + logger.info(f'OGM 注册: {global_id} -> {object_type}:{local_id}@{service_name}') + return ogm + + @staticmethod + def RegisterBatch(items, service_name): + """批量注册对象全局映射 + + Args: + items: 列表，每项为 dict，含 ObjectType, LocalID, GlobalID(可选), metadata(可选) + service_name: 所属服务名称 + + Returns: + 注册成功的 ObjectGlobalMapping 列表 + """ + try: + service = ServiceNode.objects.get(ServiceName=service_name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + raise OGMError(f'服务不存在或已禁用: {service_name}') + + results = [] + for item in items: + object_type = item['ObjectType'] + local_id = str(item['ObjectLocalID']) + global_id = item.get('ObjectGlobalID') or f'{object_type}:{uuid.uuid4().hex[:16]}' + metadata = item.get('Metadata', {}) + + if ObjectGlobalMapping.objects.filter( + ObjectType=object_type, + ObjectLocalID=local_id, + OwnerServiceUUID=service, + ).exists(): + logger.warning(f'OGM 跳过（已存在）: {object_type}:{local_id}@{service_name}') + continue + + ogm = ObjectGlobalMapping( + OGMUUID=uuid.uuid4().bytes, + ObjectType=object_type, + ObjectLocalID=local_id, + ObjectGlobalID=global_id, + OwnerServiceUUID=service, + ObjectMetadata=metadata, + ObjectStatus=1, + ) + ogm.save() + results.append(ogm) + + logger.info(f'OGM 批量注册: {len(results)}/{len(items)} 成功') + return results + + @staticmethod + def Resolve(global_id): + """通过全局 ID 解析对象映射 + + Args: + global_id: 全局 ID + + Returns: + 对象信息字典或 None + """ + try: + ogm = ObjectGlobalMapping.objects.select_related('OwnerServiceUUID').get( + ObjectGlobalID=global_id, + ObjectStatus=1, + ) + except ObjectGlobalMapping.DoesNotExist: + return None + + return OGMManager._to_dict(ogm) + + @staticmethod + def ResolveLocal(object_type, local_id, service_name): + """通过本地信息解析对象映射 + + Args: + object_type: 对象类型 + local_id: 本地 ID + service_name: 服务名称 + + Returns: + 对象信息字典或 None + """ + try: + service = ServiceNode.objects.get(ServiceName=service_name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + return None + + try: + ogm = ObjectGlobalMapping.objects.select_related('OwnerServiceUUID').get( + ObjectType=object_type, + ObjectLocalID=str(local_id), + OwnerServiceUUID=service, + ObjectStatus=1, + ) + except ObjectGlobalMapping.DoesNotExist: + return None + + return OGMManager._to_dict(ogm) + + @staticmethod + def QueryByType(object_type, active_only=True): + """按对象类型查询映射 + + Args: + object_type: 对象类型 + active_only: 是否只返回活跃对象 + + Returns: + 对象信息字典列表 + """ + qs = ObjectGlobalMapping.objects.select_related('OwnerServiceUUID').filter(ObjectType=object_type) + if active_only: + qs = qs.filter(ObjectStatus=1) + return [OGMManager._to_dict(ogm) for ogm in qs] + + @staticmethod + def QueryByService(service_name, object_type=None, active_only=True): + """按服务查询映射 + + Args: + service_name: 服务名称 + object_type: 可选的对象类型过滤 + active_only: 是否只返回活跃对象 + + Returns: + 对象信息字典列表 + """ + try: + service = ServiceNode.objects.get(ServiceName=service_name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + return [] + + qs = ObjectGlobalMapping.objects.select_related('OwnerServiceUUID').filter(OwnerServiceUUID=service) + if object_type: + qs = qs.filter(ObjectType=object_type) + if active_only: + qs = qs.filter(ObjectStatus=1) + return [OGMManager._to_dict(ogm) for ogm in qs] + + @staticmethod + def Update(global_id, metadata=None): + """更新对象映射的元数据 + + Args: + global_id: 全局 ID + metadata: 新的元数据（合并更新） + + Returns: + 更新后的 ObjectGlobalMapping 实例或 None + """ + try: + ogm = ObjectGlobalMapping.objects.get(ObjectGlobalID=global_id, ObjectStatus=1) + except ObjectGlobalMapping.DoesNotExist: + return None + + if metadata: + existing = ogm.ObjectMetadata or {} + existing.update(metadata) + ogm.ObjectMetadata = existing + + ogm.UpdateTime = _utcnow() + ogm.save() + return ogm + + @staticmethod + def Deregister(global_id): + """注销对象全局映射（软删除） + + Args: + global_id: 全局 ID + + Returns: + True 如果成功 + """ + try: + ogm = ObjectGlobalMapping.objects.get(ObjectGlobalID=global_id, ObjectStatus=1) + except ObjectGlobalMapping.DoesNotExist: + return False + + ogm.ObjectStatus = 0 + ogm.UpdateTime = _utcnow() + ogm.save(update_fields=['ObjectStatus', 'UpdateTime']) + logger.info(f'OGM 注销: {global_id}') + return True + + @staticmethod + def _to_dict(ogm): + """将 OGM 实例转换为信息字典""" + service = ogm.OwnerServiceUUID + return { + 'OGMUUID': str(ogm.OGMUUID) if isinstance(ogm.OGMUUID, bytes) else ogm.OGMUUID, + 'ObjectType': ogm.ObjectType, + 'ObjectLocalID': ogm.ObjectLocalID, + 'ObjectGlobalID': ogm.ObjectGlobalID, + 'OwnerService': service.ServiceName if service else None, + 'OwnerDomain': service.ServiceDomain if service else None, + 'Metadata': ogm.ObjectMetadata, + 'Status': ogm.ObjectStatus, + 'CreateTime': ogm.CreateTime.isoformat() if ogm.CreateTime else None, + 'UpdateTime': ogm.UpdateTime.isoformat() if ogm.UpdateTime else None, + } diff --git a/msync/registry.py b/msync/registry.py index 7397d6a..fac1d33 100644 --- a/msync/registry.py +++ b/msync/registry.py @@ -1,17 +1,263 @@ -"""MSYNC 注册中心模块 +"""模块：服务注册与发现 +提供服务节点的注册、注销、发现、心跳等操作。""" -薄封装：直接转发到 gvsds.apps.msync.registry 原生层。 -""" +import uuid +import datetime +import logging +from gvsdsdk.msync.models import ServiceNode, _utcnow -try: - from gvsds.apps.msync.registry import ServiceRegistry, RegistryError -except ImportError: - raise ImportError( - 'ServiceRegistry 仅在主服务器环境可用。' - '子服务器请通过 ServiceClient.DiscoverService() 使用 HTTP 转发。' - ) +logger = logging.getLogger(__name__) -__all__ = [ - 'ServiceRegistry', - 'RegistryError', -] + +class RegistryError(Exception): + pass + + +class ServiceRegistry: + """服务注册中心——管理微服务的注册、发现与生命周期""" + + SERVICE_TYPE_WORKER = 0 + SERVICE_TYPE_MASTER = 1 + + DEFAULT_SCOPE = ['ogm:read', 'ogm:write', 'registry:read'] + + @staticmethod + def Register(name, domain, secret_hash, raw_secret=None, scope=None, service_type=0, metadata=None): + """注册新的服务节点（已存在则验证密钥哈希并更新域信息） + + Args: + name: 服务名称（唯一标识） + domain: 服务域名或 URL + secret_hash: 服务密钥的 SHA256 哈希（客户端计算，不再传输明文） + raw_secret: 服务原始密钥（用于加密存储，认证时需要） + scope: 权限范围列表 + service_type: 服务类型（0=工作节点, 1=主节点） + metadata: 服务元数据 + + Returns: + ServiceNode 实例 + + Raises: + RegistryError: 注册失败 + """ + import hashlib + # 密钥强制：raw_secret 与 secret_hash 必须自洽； + # 新服务节点必须携带明文密钥，否则后续 HMAC 认证无法完成 + if raw_secret: + computed_hash = hashlib.sha256(raw_secret.encode('utf-8')).hexdigest() + if secret_hash and computed_hash != secret_hash: + raise RegistryError('服务密钥哈希不一致') + secret_hash = secret_hash or computed_hash + elif not secret_hash: + raise RegistryError('服务密钥缺失') + + existing = ServiceNode.objects.filter(ServiceName=name).first() + if existing: + if not existing.ServiceSecretHash or existing.ServiceSecretHash != secret_hash: + raise RegistryError('服务密钥验证失败') + existing.ServiceDomain = domain + existing.ServiceScope = scope or ServiceRegistry.DEFAULT_SCOPE + existing.ServiceStatus = 1 + existing.ServiceType = service_type + existing.ServiceMetadata = metadata or {} + if raw_secret: + existing.SetSecret(raw_secret) + existing.save() + logger.info(f'服务重新注册: {name} ({domain})') + return existing + + if not raw_secret: + raise RegistryError('新服务注册必须提供 ServiceSecret 密钥') + + node = ServiceNode( + ServiceUUID=uuid.uuid4().bytes, + ServiceName=name, + ServiceDomain=domain, + ServiceScope=scope or ServiceRegistry.DEFAULT_SCOPE, + ServiceStatus=1, + ServiceType=service_type, + ServiceMetadata=metadata or {}, + ServiceSecretHash=secret_hash, + ) + if raw_secret: + node.SetSecret(raw_secret) + node.save() + logger.info(f'服务注册成功: {name} ({domain})') + return node + + @staticmethod + def Deregister(name): + """注销服务节点（软删除，设置状态为禁用） + + Args: + name: 服务名称 + + Returns: + True 如果成功 + + Raises: + RegistryError: 服务不存在 + """ + try: + node = ServiceNode.objects.get(ServiceName=name) + except ServiceNode.DoesNotExist: + raise RegistryError(f'服务不存在: {name}') + + node.ServiceStatus = 0 + node.save(update_fields=['ServiceStatus']) + logger.info(f'服务已注销: {name}') + return True + + @staticmethod + def Discover(name): + """发现指定服务 + + Args: + name: 服务名称 + + Returns: + 服务信息字典或 None + """ + try: + node = ServiceNode.objects.get(ServiceName=name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + return None + + return { + 'ServiceName': node.ServiceName, + 'ServiceDomain': node.ServiceDomain, + 'ServiceScope': node.ServiceScope, + 'ServiceType': node.ServiceType, + 'ServiceMetadata': node.ServiceMetadata, + 'ServiceStatus': node.ServiceStatus, + 'LastHeartbeat': node.LastHeartbeat.isoformat() if node.LastHeartbeat else None, + } + + @staticmethod + def GetNode(name): + """获取服务节点实例（不按状态过滤，已注销也可查询）；不存在返回 None""" + return ServiceNode.objects.filter(ServiceName=name).first() + + @staticmethod + def DiscoverAll(active_only=True, page=1, page_size=50): + """发现所有已注册服务（分页） + + Args: + active_only: 是否只返回活跃服务 + page: 页码（从1开始） + page_size: 每页数量 + + Returns: + 包含分页元数据和服务列表的字典 + """ + qs = ServiceNode.objects.all() + if active_only: + qs = qs.filter(ServiceStatus=1) + + total = qs.count() + offset = (page - 1) * page_size + page_qs = qs[offset:offset + page_size] + + services = [ + { + 'ServiceName': node.ServiceName, + 'ServiceDomain': node.ServiceDomain, + 'ServiceScope': node.ServiceScope, + 'ServiceType': node.ServiceType, + 'ServiceMetadata': node.ServiceMetadata, + 'ServiceStatus': node.ServiceStatus, + 'LastHeartbeat': node.LastHeartbeat.isoformat() if node.LastHeartbeat else None, + } + for node in page_qs + ] + + return { + 'total': total, + 'page': page, + 'page_size': page_size, + 'services': services, + } + + @staticmethod + def Heartbeat(name): + """服务心跳——更新最后心跳时间 + + Args: + name: 服务名称 + + Returns: + True 如果成功 + + Raises: + RegistryError: 服务不存在 + """ + try: + node = ServiceNode.objects.get(ServiceName=name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + raise RegistryError(f'服务不存在或已禁用: {name}') + + node.LastHeartbeat = _utcnow() + node.save(update_fields=['LastHeartbeat']) + return True + + @staticmethod + def UpdateScope(name, scope): + """更新服务权限范围 + + Args: + name: 服务名称 + scope: 新的权限范围列表 + + Raises: + RegistryError: 服务不存在 + """ + try: + node = ServiceNode.objects.get(ServiceName=name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + raise RegistryError(f'服务不存在或已禁用: {name}') + + node.ServiceScope = scope + node.save(update_fields=['ServiceScope']) + logger.info(f'服务权限更新: {name} -> {scope}') + return True + + @staticmethod + def RotateSecret(name, new_secret): + """轮换服务密钥 + + Args: + name: 服务名称 + new_secret: 新密钥（明文） + + Raises: + RegistryError: 服务不存在 + """ + try: + node = ServiceNode.objects.get(ServiceName=name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + raise RegistryError(f'服务不存在或已禁用: {name}') + + node.SetSecret(new_secret) + node.save(update_fields=['ServiceSecretHash', 'ServiceSecretEncrypted']) + + from gvsdsdk.msync.models import ServiceToken + ServiceToken.objects.filter(ServiceUUID=node, TokenStatus=1).update(TokenStatus=0) + + logger.info(f'服务密钥已轮换: {name}') + return True + + @staticmethod + def GetInactiveServices(timeout_minutes=5): + """获取超时未心跳的服务 + + Args: + timeout_minutes: 超时分钟数 + + Returns: + ServiceNode QuerySet + """ + threshold = _utcnow() - datetime.timedelta(minutes=timeout_minutes) + return ServiceNode.objects.filter( + ServiceStatus=1, + LastHeartbeat__lt=threshold, + ) diff --git a/msync/serializers.py b/msync/serializers.py new file mode 100644 index 0000000..1807f10 --- /dev/null +++ b/msync/serializers.py @@ -0,0 +1,98 @@ +from rest_framework import serializers +from gvsdsdk.serializer_base import PascalCaseSerializer, PascalCaseModelSerializer +from gvsdsdk.msync.models import ServiceNode + + +class ServiceNodeSerializer(PascalCaseModelSerializer): + class Meta: + model = ServiceNode + fields = ['ServiceName', 'ServiceDomain', 'ServiceScope', 'ServiceType', 'ServiceMetadata', 'ServiceStatus', 'CreateTime', 'LastHeartbeat'] + read_only_fields = ['CreateTime', 'LastHeartbeat'] + + +class ServiceTokenResponseSerializer(PascalCaseSerializer): + AccessToken = serializers.CharField() + RefreshToken = serializers.CharField() + Scope = serializers.ListField() + ExpiresAt = serializers.FloatField() + + +class AuthTokenRequestSerializer(PascalCaseSerializer): + ServiceName = serializers.CharField() + Payload = serializers.DictField() + Signature = serializers.CharField() + + +class RefreshTokenRequestSerializer(PascalCaseSerializer): + RefreshToken = serializers.CharField() + + +class OGMRegisterSerializer(PascalCaseSerializer): + ObjectType = serializers.CharField() + ObjectLocalID = serializers.CharField() + ServiceName = serializers.CharField() + ObjectGlobalID = serializers.CharField(required=False, allow_blank=True) + Metadata = serializers.DictField(required=False, default=dict) + + +class OGMBatchRegisterSerializer(PascalCaseSerializer): + Items = OGMRegisterSerializer(many=True) + ServiceName = serializers.CharField() + + +class ServiceRegisterSerializer(PascalCaseSerializer): + ServiceName = serializers.CharField() + ServiceDomain = serializers.CharField() + ServiceSecret = serializers.CharField(write_only=True) + ServiceSecretHash = serializers.CharField(required=False, allow_blank=True) + ServiceScope = serializers.ListField(required=False, default=['ogm:read', 'ogm:write', 'registry:read']) + ServiceType = serializers.IntegerField(default=0) + ServiceMetadata = serializers.DictField(required=False, default=dict) + + +class TransactionStepSerializer(PascalCaseSerializer): + StepName = serializers.CharField() + StepDescription = serializers.CharField(required=False, default='') + ParticipantServiceUUID = serializers.CharField(required=False, allow_blank=True) + ActionEndpoint = serializers.CharField() + ActionMethod = serializers.CharField(default='POST') + ActionPayload = serializers.DictField(required=False, default=dict) + CompensateEndpoint = serializers.CharField(required=False, default='') + CompensateMethod = serializers.CharField(default='POST') + CompensatePayload = serializers.DictField(required=False, default=dict) + MaxRetries = serializers.IntegerField(default=3) + + +class TransactionCreateSerializer(PascalCaseSerializer): + TransactionName = serializers.CharField() + TransactionDescription = serializers.CharField(required=False, default='') + TransactionType = serializers.ChoiceField(choices=['saga', '2pc'], default='saga') + Steps = TransactionStepSerializer(many=True) + InitiatorServiceName = serializers.CharField(required=False, allow_blank=True) + ReferenceOGMUUID = serializers.CharField(required=False, allow_blank=True) + Payload = serializers.DictField(required=False, default=dict) + TimeoutSeconds = serializers.IntegerField(default=300) + + +class TransactionExecuteSerializer(PascalCaseSerializer): + AccessToken = serializers.CharField(required=False, allow_blank=True) + + +class TransactionRetryStepSerializer(PascalCaseSerializer): + StepIndex = serializers.IntegerField() + AccessToken = serializers.CharField(required=False, allow_blank=True) + + +class LocalExecuteSerializer(PascalCaseSerializer): + StepName = serializers.CharField() + Payload = serializers.DictField(required=False, default=dict) + TransactionCode = serializers.CharField(required=False, allow_blank=True) + StepIndex = serializers.IntegerField(required=False, allow_null=True) + + +class LocalCompensateSerializer(PascalCaseSerializer): + StepName = serializers.CharField() + Payload = serializers.DictField(required=False, default=dict) + ActionResult = serializers.DictField(required=False, default=dict) + TransactionCode = serializers.CharField(required=False, allow_blank=True) + StepIndex = serializers.IntegerField(required=False, allow_null=True) diff --git a/msync/sio.py b/msync/sio.py new file mode 100644 index 0000000..a13892c --- /dev/null +++ b/msync/sio.py @@ -0,0 +1,545 @@ +"""模块：MSYNC 服务端（Socket.IO） + +替代 REST API，提供实时双向通信。事件协议字段全部使用 PascalCase。 + +事件协议: + 客户端 → 服务端: + msync:register 注册服务（bootstrap，必须携带密钥） + msync:auth HMAC 签名认证（密钥验证，成功后才可调用业务事件） + msync:auth:refresh 刷新令牌 + msync:heartbeat 应用层心跳（需已认证） + msync:deregister 注销服务（需已认证，仅限本服务） + msync:discover 发现所有服务（需已认证） + msync:discover:one 发现单个服务（需已认证） + msync:ogm:register 注册 OGM（需已认证） + msync:ogm:batch 批量注册 OGM（需已认证） + msync:ogm:resolve 解析 OGM（需已认证） + msync:ogm:query 查询 OGM（需已认证） + msync:ogm:update 更新 OGM（需已认证） + msync:ogm:deregister 注销 OGM（需已认证） + msync:tx:create 创建事务（需已认证） + msync:tx:execute 执行事务（需已认证） + msync:tx:compensate 补偿事务（需已认证） + msync:tx:cancel 取消事务（需已认证） + msync:tx:detail 事务详情（需已认证） + msync:tx:retry 重试步骤（需已认证） + msync:tx:retry_compensate 重试补偿（需已认证） + msync:tx:timeout_check 超时检测（需已认证） + msync:tx:query 按状态查询（需已认证） + + 鉴权规则: 除 register / auth / auth:refresh 外，所有业务事件必须 + 先通过 msync:auth 密钥认证（HMAC 签名验证），否则返回 401。 + + 服务端 → 客户端（推送）: + msync:service:online 服务上线 + msync:service:offline 服务下线 +""" + +import logging +import threading +import traceback +import socketio + +logger = logging.getLogger(__name__) + +NS = '/msync' + +sio = socketio.Server(async_mode='gevent', cors_allowed_origins=[ + 'https://*.gvsds.com', + 'https://*.gvsds.dev', + 'https://*.yc8z.org', + 'https://*.ycbz.org', + 'https://*.gvsds.org', + 'https://*.gvsds.studio', + 'http://localhost:*', + 'http://127.0.0.1:*', +], logger=True, engineio_logger=True) + +_sid_service_map = {} +_service_sid_map = {} +_sid_auth_map = {} +_map_lock = threading.Lock() + + +def _ok(data=None, msg='success'): + return {'code': 200, 'msg': msg, 'data': data} + + +def _err(msg, code=400): + return {'code': code, 'msg': msg, 'data': None} + + +def _d(data, key, default=None): + """安全获取字典字段""" + if not isinstance(data, dict): + return default + return data.get(key, default) + + +def _RequireAuth(sid): + """校验 sid 是否已完成密钥认证（msync:auth）；通过返回服务名，否则返回 None""" + with _map_lock: + return _sid_auth_map.get(sid) + + +@sio.on('connect', namespace=NS) +def on_connect(sid, environ): + logger.info(f'[SIO] → connect sid={sid}') + + +@sio.on('disconnect', namespace=NS) +def on_disconnect(sid): + with _map_lock: + service_name = _sid_service_map.pop(sid, None) + if service_name: + _service_sid_map.pop(service_name, None) + _sid_auth_map.pop(sid, None) + if service_name: + try: + from gvsdsdk.msync.registry import ServiceRegistry + ServiceRegistry.Deregister(service_name) + except Exception: + pass + sio.emit('msync:service:offline', {'Service': service_name}, namespace=NS, skip_sid=sid) + logger.info(f'[SIO] 服务断开: {service_name} ({sid})') + else: + logger.info(f'[SIO] 客户端断开: {sid}') + + +@sio.on('msync:register', namespace=NS) +def on_register(sid, data): + logger.info(f'[SIO] → msync:register sid={sid} name={_d(data, "ServiceName")}') + from gvsdsdk.msync.registry import ServiceRegistry, RegistryError + try: + node = ServiceRegistry.Register( + name=_d(data, 'ServiceName', ''), + domain=_d(data, 'ServiceDomain', ''), + secret_hash=_d(data, 'ServiceSecretHash', ''), + raw_secret=_d(data, 'ServiceSecret'), + scope=_d(data, 'ServiceScope'), + service_type=_d(data, 'ServiceType', 0), + metadata=_d(data, 'ServiceMetadata', {}), + ) + except RegistryError as e: + return _err(str(e), 409) + except Exception as e: + logger.error(f'注册异常: {e}\n{traceback.format_exc()}') + return _err(str(e), 500) + + service_name = _d(data, 'ServiceName') + with _map_lock: + _sid_service_map[sid] = service_name + _service_sid_map[service_name] = sid + # 若以不同服务名重新注册，旧认证状态失效 + if _sid_auth_map.get(sid) != service_name: + _sid_auth_map.pop(sid, None) + + sio.emit('msync:service:online', { + 'ServiceName': service_name, + 'ServiceDomain': _d(data, 'ServiceDomain'), + }, namespace=NS, skip_sid=sid) + + from gvsdsdk.msync.serializers import ServiceNodeSerializer + result = _ok(ServiceNodeSerializer(node).data) + logger.info(f'[SIO] ← msync:register sid={sid} code=200') + return result + + +@sio.on('msync:auth', namespace=NS) +def on_auth(sid, data): + service_name = _d(data, 'ServiceName') + logger.info(f'[SIO] → msync:auth sid={sid} service={service_name}') + from gvsdsdk.msync.auth import AuthenticateService + try: + token = AuthenticateService( + service_name=service_name, + signed_payload=_d(data, 'Payload'), + signature=_d(data, 'Signature'), + ) + if token is None: + return _err('认证失败', 401) + + try: + scope = token.TokenScope + if not isinstance(scope, list): + scope = list(scope) if scope else [] + except Exception: + scope = [] + + try: + from django.utils.dateparse import parse_datetime + if isinstance(token.ExpiresAt, str): + dt = parse_datetime(token.ExpiresAt) + if dt is None: + raise ValueError + else: + dt = token.ExpiresAt + # 若是 naive datetime，按 UTC 处理（DB 存的就是 naive UTC） + if dt.tzinfo is None: + import datetime as _dt + dt = dt.replace(tzinfo=_dt.timezone.utc) + expires_ts = dt.timestamp() + except Exception: + import time as _time + expires_ts = _time.time() + 7200 + + with _map_lock: + _sid_auth_map[sid] = service_name + + return _ok({ + 'AccessToken': token.AccessToken, + 'RefreshToken': token.RefreshToken, + 'Scope': scope, + 'ExpiresAt': expires_ts, + }) + except Exception as e: + logger.error(f'[SIO] msync:auth 异常: {e}\n{traceback.format_exc()}') + return _err(str(e), 500) + + +@sio.on('msync:auth:refresh', namespace=NS) +def on_auth_refresh(sid, data): + from gvsdsdk.msync.auth import RefreshServiceToken + try: + token = RefreshServiceToken(_d(data, 'RefreshToken', '')) + if token is None: + return _err('刷新失败', 401) + + try: + scope = token.TokenScope + if not isinstance(scope, list): + scope = list(scope) if scope else [] + except Exception: + scope = [] + + try: + from django.utils.dateparse import parse_datetime + if isinstance(token.ExpiresAt, str): + dt = parse_datetime(token.ExpiresAt) + if dt is None: + raise ValueError + else: + dt = token.ExpiresAt + if dt.tzinfo is None: + import datetime as _dt + dt = dt.replace(tzinfo=_dt.timezone.utc) + expires_ts = dt.timestamp() + except Exception: + import time as _time + expires_ts = _time.time() + 7200 + + with _map_lock: + _sid_auth_map[sid] = token.ServiceUUID.ServiceName + + return _ok({ + 'AccessToken': token.AccessToken, + 'RefreshToken': token.RefreshToken, + 'Scope': scope, + 'ExpiresAt': expires_ts, + }) + except Exception as e: + logger.error(f'Socket令牌刷新异常: {e}\n{traceback.format_exc()}') + return _err(str(e), 500) + + +@sio.on('msync:heartbeat', namespace=NS) +def on_heartbeat(sid, data): + with _map_lock: + service_name = _sid_service_map.get(sid) + auth_name = _sid_auth_map.get(sid) + if not service_name: + return _err('未注册', 401) + if not auth_name or auth_name != service_name: + return _err('需要密钥认证', 401) + from gvsdsdk.msync.registry import ServiceRegistry, RegistryError + try: + ServiceRegistry.Heartbeat(service_name) + except RegistryError as e: + return _err(str(e), 404) + return _ok(msg='心跳成功') + + +@sio.on('msync:deregister', namespace=NS) +def on_deregister(sid, data): + with _map_lock: + service_name = _sid_service_map.pop(sid, None) + if service_name: + _service_sid_map.pop(service_name, None) + auth_name = _sid_auth_map.pop(sid, None) + name = _d(data, 'ServiceName') or service_name + if not name: + return _err('未指定服务名', 400) + # 仅允许注销本服务（防止伪造 ServiceName 注销他人） + if not auth_name or auth_name != name: + return _err('需要密钥认证', 401) + from gvsdsdk.msync.registry import ServiceRegistry, RegistryError + try: + ServiceRegistry.Deregister(name) + except RegistryError as e: + return _err(str(e), 404) + sio.emit('msync:service:offline', {'Service': name}, namespace=NS, skip_sid=sid) + return _ok(msg='注销成功') + + +@sio.on('msync:discover', namespace=NS) +def on_discover(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.registry import ServiceRegistry + active_only = _d(data, 'ActiveOnly', True) + if not isinstance(active_only, bool): + active_only = True + services = ServiceRegistry.DiscoverAll(active_only=active_only) + return _ok(services) + + +@sio.on('msync:discover:one', namespace=NS) +def on_discover_one(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.registry import ServiceRegistry + info = ServiceRegistry.Discover(_d(data, 'ServiceName', '')) + if info is None: + return _err('服务不存在', 404) + return _ok(info) + + +@sio.on('msync:ogm:register', namespace=NS) +def on_ogm_register(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.ogm import OGMManager, OGMError + try: + ogm = OGMManager.Register( + object_type=_d(data, 'ObjectType'), + local_id=_d(data, 'ObjectLocalID'), + service_name=_d(data, 'ServiceName'), + global_id=_d(data, 'ObjectGlobalID'), + metadata=_d(data, 'Metadata', {}), + ) + except OGMError as e: + return _err(str(e), 409) + except Exception as e: + logger.error(f'OGM注册异常: {e}\n{traceback.format_exc()}') + return _err(str(e), 500) + return _ok(OGMManager._to_dict(ogm)) + + +@sio.on('msync:ogm:batch', namespace=NS) +def on_ogm_batch(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.ogm import OGMManager, OGMError + try: + results = OGMManager.RegisterBatch( + items=_d(data, 'Items', []), + service_name=_d(data, 'ServiceName'), + ) + except OGMError as e: + return _err(str(e), 409) + return _ok([OGMManager._to_dict(o) for o in results]) + + +@sio.on('msync:ogm:resolve', namespace=NS) +def on_ogm_resolve(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.ogm import OGMManager + info = OGMManager.Resolve(_d(data, 'ObjectGlobalID', '')) + if info is None: + return _err('映射不存在', 404) + return _ok(info) + + +@sio.on('msync:ogm:query', namespace=NS) +def on_ogm_query(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.ogm import OGMManager + object_type = _d(data, 'ObjectType') + service_name = _d(data, 'ServiceName') + local_id = _d(data, 'ObjectLocalID') + + if local_id and object_type and service_name: + info = OGMManager.ResolveLocal(object_type, local_id, service_name) + if info is None: + return _err('映射不存在', 404) + return _ok(info) + + if object_type: + results = OGMManager.QueryByType(object_type) + return _ok(results) + + if service_name: + results = OGMManager.QueryByService(service_name, object_type=object_type) + return _ok(results) + + return _err('请提供 ObjectType 或 ServiceName 参数', 400) + + +@sio.on('msync:ogm:update', namespace=NS) +def on_ogm_update(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.ogm import OGMManager + ogm = OGMManager.Update(_d(data, 'ObjectGlobalID', ''), metadata=_d(data, 'Metadata')) + if ogm is None: + return _err('映射不存在', 404) + return _ok(OGMManager._to_dict(ogm)) + + +@sio.on('msync:ogm:deregister', namespace=NS) +def on_ogm_deregister(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.ogm import OGMManager + success = OGMManager.Deregister(_d(data, 'ObjectGlobalID', '')) + if not success: + return _err('映射不存在', 404) + return _ok(msg='注销成功') + + +@sio.on('msync:tx:create', namespace=NS) +def on_tx_create(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager, TransactionError + try: + dtx = TransactionManager.Create( + name=_d(data, 'TransactionName'), + steps_data=_d(data, 'Steps'), + description=_d(data, 'TransactionDescription', ''), + transaction_type=_d(data, 'TransactionType', 'saga'), + initiator_service_name=_d(data, 'InitiatorServiceName'), + reference_ogm_uuid=_d(data, 'ReferenceOGMUUID'), + payload=_d(data, 'Payload'), + timeout_seconds=_d(data, 'TimeoutSeconds', 300), + ) + except TransactionError as e: + return _err(str(e), 400) + return _ok(TransactionManager.GetDetail(dtx.TransactionCode)) + + +@sio.on('msync:tx:execute', namespace=NS) +def on_tx_execute(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager, TransactionError + try: + dtx = TransactionManager.Execute( + _d(data, 'TransactionCode', ''), + access_token=_d(data, 'AccessToken'), + ) + except TransactionError as e: + return _err(str(e), 400) + return _ok(TransactionManager.GetDetail(dtx.TransactionCode)) + + +@sio.on('msync:tx:compensate', namespace=NS) +def on_tx_compensate(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager, TransactionError + try: + dtx = TransactionManager.Compensate( + _d(data, 'TransactionCode', ''), + access_token=_d(data, 'AccessToken'), + ) + except TransactionError as e: + return _err(str(e), 400) + return _ok(TransactionManager.GetDetail(dtx.TransactionCode)) + + +@sio.on('msync:tx:cancel', namespace=NS) +def on_tx_cancel(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager, TransactionError + try: + TransactionManager.Cancel(_d(data, 'TransactionCode', '')) + except TransactionError as e: + return _err(str(e), 400) + return _ok(msg='事务已取消') + + +@sio.on('msync:tx:detail', namespace=NS) +def on_tx_detail(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager + detail = TransactionManager.GetDetail(_d(data, 'TransactionCode', '')) + if detail is None: + return _err('事务不存在', 404) + return _ok(detail) + + +@sio.on('msync:tx:retry', namespace=NS) +def on_tx_retry(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager, TransactionError + try: + step = TransactionManager.RetryStep( + _d(data, 'TransactionCode', ''), + step_index=_d(data, 'StepIndex', 0), + access_token=_d(data, 'AccessToken'), + ) + except TransactionError as e: + return _err(str(e), 400) + return _ok({ + 'StepIndex': step.StepIndex, + 'StepName': step.StepName, + 'Status': step.get_StepStatus_display(), + 'RetryCount': step.RetryCount, + }) + + +@sio.on('msync:tx:retry_compensate', namespace=NS) +def on_tx_retry_compensate(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager, TransactionError + try: + step = TransactionManager.RetryCompensate( + _d(data, 'TransactionCode', ''), + step_index=_d(data, 'StepIndex', 0), + access_token=_d(data, 'AccessToken'), + ) + except TransactionError as e: + return _err(str(e), 400) + return _ok({ + 'StepIndex': step.StepIndex, + 'StepName': step.StepName, + 'Status': step.get_StepStatus_display(), + 'RetryCount': step.RetryCount, + 'TransactionStatus': step.TransactionUUID.get_TransactionStatus_display(), + }) + + +@sio.on('msync:tx:timeout_check', namespace=NS) +def on_tx_timeout_check(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager + timed_out = TransactionManager.CheckTimeouts() + return _ok({'TimedOut': timed_out, 'Count': len(timed_out)}) + + +@sio.on('msync:tx:query', namespace=NS) +def on_tx_query(sid, data): + if not _RequireAuth(sid): + return _err('需要密钥认证', 401) + from gvsdsdk.msync.transaction import TransactionManager + status = _d(data, 'Status') + if status is None: + return _err('请提供 Status 参数', 400) + try: + status_int = int(status) + except (ValueError, TypeError): + return _err('Status 必须为整数', 400) + results = TransactionManager.QueryByStatus(status_int) + return _ok(results) + + +def create_sio_app(django_app): + return socketio.WSGIApp(sio, django_app) diff --git a/msync/transaction.py b/msync/transaction.py index 26a34e7..4c4da79 100644 --- a/msync/transaction.py +++ b/msync/transaction.py @@ -1,18 +1,668 @@ -"""MSYNC 分布式事务模块 +"""模块：远程事务管理器 +基于 Saga 模式的跨服务事务编排引擎，运行于主服务器（账本服务器）。 +通过调用各工作服务的本地事务管理器端点来执行与补偿步骤。 -薄封装：直接转发到 gvsds.apps.msync.transaction 原生层。 -""" +架构： + 主服务器 (TransactionManager) + │ + ├──▶ POST {service_domain}/api/msync/local/execute/ + │ {StepName, payload} + │ → 工作服务的 LocalTransactionManager.execute_step() + │ + └──▶ POST {service_domain}/api/msync/local/compensate/ + {StepName, payload, ActionResult} + → 工作服务的 LocalTransactionManager.compensate_step() -try: - from gvsds.apps.msync.transaction import TransactionManager, TransactionError, StepExecutionError -except ImportError: - raise ImportError( - 'TransactionManager 仅在主服务器环境可用。' - '子服务器请通过 LocalTransactionManager 注册本地步骤处理器。' - ) +事务生命周期： + Pending → Running → Committed + ↘ Failed → Compensating → Compensated + ↘ TimedOut → Compensating → Compensated""" -__all__ = [ - 'TransactionManager', - 'TransactionError', - 'StepExecutionError', -] +import uuid +import json +import logging +import os +import datetime +import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry +from django.db import transaction as db_transaction +from gvsdsdk.msync.models import ( + DistributedTransaction, + TransactionStep, + ServiceNode, + _utcnow, +) + +logger = logging.getLogger(__name__) + +LOCAL_EXECUTE_PATH = '/api/msync/local/execute/' +LOCAL_COMPENSATE_PATH = '/api/msync/local/compensate/' + +# ─── 模块级 HTTP 会话（连接池 + 重试 + TLS 验证）───────────────── +_http_session = requests.Session() +_http_session.verify = True # 显式启用 TLS 证书验证 + +# 支持通过环境变量指定 CA 证书包（开发/部署灵活性） +_ca_bundle = os.environ.get('REQUESTS_CA_BUNDLE') or os.environ.get('CURL_CA_BUNDLE') +if _ca_bundle: + _http_session.verify = _ca_bundle + +# 配置连接池重试策略 +_retry_adapter = HTTPAdapter( + max_retries=Retry( + total=3, + backoff_factor=0.5, + status_forcelist=[502, 503, 504], + allowed_methods=['POST'], + ), + pool_connections=10, + pool_maxsize=10, +) +_http_session.mount('https://', _retry_adapter) +_http_session.mount('http://', _retry_adapter) + + +class TransactionError(Exception): + pass + + +class StepExecutionError(Exception): + def __init__(self, message, step_index=None, status_code=None, response_body=None): + super().__init__(message) + self.StepIndex = step_index + self.StatusCode = status_code + self.ResponseBody = response_body + + +class TransactionManager: + """远程事务管理器——Saga 编排引擎（运行于主服务器） + + 调用各工作服务的本地事务管理器来执行步骤： + - 执行步骤 → POST {service_domain}/api/msync/local/execute/ + - 补偿步骤 → POST {service_domain}/api/msync/local/compensate/ + """ + + @staticmethod + @db_transaction.atomic + def Create(name, steps_data, description='', transaction_type='saga', + initiator_service_name=None, reference_ogm_uuid=None, + payload=None, timeout_seconds=300): + """创建分布式事务 + + Args: + name: 事务名称 + steps_data: 步骤数据列表，每项为 dict: + { + 'name': '步骤名称（对应本地事务管理器注册的处理器名）', + 'description': '步骤描述', + 'ParticipantService': '服务名称（账本中注册的服务名）', + 'ActionPayload': {...}, + 'CompensatePayload': {...}, + 'MaxRetries': 3, + } + description: 事务描述 + transaction_type: 事务类型 (saga/2pc) + initiator_service_name: 发起服务名称 + reference_ogm_uuid: 关联 OGM UUID + payload: 事务级载荷 + timeout_seconds: 超时秒数 + + Returns: + DistributedTransaction 实例 + """ + if not steps_data: + raise TransactionError('事务至少需要一个步骤') + + initiator = None + if initiator_service_name: + try: + initiator = ServiceNode.objects.get(ServiceName=initiator_service_name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + raise TransactionError(f'发起服务不存在: {initiator_service_name}') + + tx_code = f'DTX-{uuid.uuid4().hex[:12].upper()}' + + dtx = DistributedTransaction( + TransactionUUID=uuid.uuid4().bytes, + TransactionCode=tx_code, + TransactionType=transaction_type, + TransactionName=name, + TransactionDescription=description, + TransactionStatus=DistributedTransaction.TX_STATUS_PENDING, + InitiatorServiceUUID=initiator, + ReferenceOGMUUID=reference_ogm_uuid, + TransactionPayload=payload or {}, + TimeoutSeconds=timeout_seconds, + ) + dtx.save() + + for idx, step_data in enumerate(steps_data): + participant = None + participant_name = step_data.get('ParticipantService') + if participant_name: + try: + participant = ServiceNode.objects.get(ServiceName=participant_name, ServiceStatus=1) + except ServiceNode.DoesNotExist: + raise TransactionError(f'参与服务不存在: {participant_name}') + + step = TransactionStep( + StepUUID=uuid.uuid4().bytes, + TransactionUUID=dtx, + StepIndex=idx, + StepName=step_data['name'], + StepDescription=step_data.get('description', ''), + ParticipantServiceUUID=participant, + StepStatus=TransactionStep.STEP_STATUS_PENDING, + ActionEndpoint=step_data.get('ActionEndpoint', ''), + ActionMethod='POST', + ActionPayload=step_data.get('ActionPayload', {}), + CompensateEndpoint=step_data.get('CompensateEndpoint', ''), + CompensateMethod='POST', + CompensatePayload=step_data.get('CompensatePayload', {}), + MaxRetries=step_data.get('MaxRetries', 3), + ) + step.save() + + logger.info(f'分布式事务创建: {tx_code} ({name}), {len(steps_data)} 步骤') + return dtx + + @staticmethod + @db_transaction.atomic + def Execute(transaction_code, access_token=None): + """执行分布式事务（Saga 正向流程） + + 按步骤顺序依次调用各工作服务的本地事务管理器， + 任一步骤失败则触发补偿流程。 + + Args: + transaction_code: 事务代码 + access_token: 服务间认证令牌 + + Returns: + DistributedTransaction 实例 + """ + try: + dtx = DistributedTransaction.objects.select_for_update().get(TransactionCode=transaction_code) + except DistributedTransaction.DoesNotExist: + raise TransactionError(f'事务不存在: {transaction_code}') + + if dtx.TransactionStatus not in ( + DistributedTransaction.TX_STATUS_PENDING, + DistributedTransaction.TX_STATUS_RUNNING, + ): + raise TransactionError(f'事务状态不允许执行: {dtx.get_TransactionStatus_display()}') + + if dtx.IsTimedOut: + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_TIMED_OUT + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'UpdateTime']) + db_transaction.on_commit( + lambda: TransactionManager.Compensate(transaction_code, access_token=access_token) + ) + raise TransactionError(f'事务已超时: {transaction_code}') + + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_RUNNING + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'UpdateTime']) + + steps = list(dtx.steps.all().order_by('StepIndex')) + + for step in steps: + if step.StepStatus == TransactionStep.STEP_STATUS_COMPLETED: + continue + + if dtx.IsTimedOut: + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_TIMED_OUT + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'UpdateTime']) + db_transaction.on_commit( + lambda tc=transaction_code, at=access_token: TransactionManager.Compensate(tc, access_token=at) + ) + raise TransactionError(f'事务执行超时: {transaction_code}') + + try: + result = TransactionManager._execute_step(step, access_token=access_token) + step.StepStatus = TransactionStep.STEP_STATUS_COMPLETED + step.StepResult = result + step.EndTime = _utcnow() + step.save(update_fields=['StepStatus', 'StepResult', 'EndTime']) + except StepExecutionError as e: + step.StepStatus = TransactionStep.STEP_STATUS_FAILED + step.StepResult = { + 'error': str(e), + 'status_code': e.StatusCode, + 'response': e.ResponseBody, + } + step.EndTime = _utcnow() + step.save(update_fields=['StepStatus', 'StepResult', 'EndTime']) + + logger.warning(f'事务步骤失败: {transaction_code} step={step.StepIndex}, 触发补偿') + db_transaction.on_commit( + lambda tc=transaction_code, at=access_token: TransactionManager.Compensate(tc, access_token=at) + ) + return DistributedTransaction.objects.get(TransactionCode=transaction_code) + + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_COMMITTED + dtx.EndTime = _utcnow() + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'EndTime', 'UpdateTime']) + + logger.info(f'分布式事务提交: {transaction_code}') + return dtx + + @staticmethod + @db_transaction.atomic + def Compensate(transaction_code, access_token=None): + """补偿分布式事务（Saga 反向流程） + + 逆序遍历已完成步骤，调用各工作服务的本地事务管理器补偿端点。 + + Args: + transaction_code: 事务代码 + access_token: 服务间认证令牌 + """ + try: + dtx = DistributedTransaction.objects.select_for_update().get(TransactionCode=transaction_code) + except DistributedTransaction.DoesNotExist: + raise TransactionError(f'事务不存在: {transaction_code}') + + if dtx.TransactionStatus == DistributedTransaction.TX_STATUS_COMPENSATED: + return dtx + + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_COMPENSATING + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'UpdateTime']) + + completed_steps = list( + dtx.steps.filter(StepStatus=TransactionStep.STEP_STATUS_COMPLETED) + .order_by('-StepIndex') + ) + + compensation_failed = False + for step in completed_steps: + step.StepStatus = TransactionStep.STEP_STATUS_COMPENSATING + step.save(update_fields=['StepStatus']) + + try: + result = TransactionManager._compensate_step(step, access_token=access_token) + step.StepStatus = TransactionStep.STEP_STATUS_COMPENSATED + step.StepResult.update({'compensate_result': result}) + step.EndTime = _utcnow() + step.save(update_fields=['StepStatus', 'StepResult', 'EndTime']) + except StepExecutionError as e: + # 补偿失败：步骤置 FAILED（而非停留在 COMPENSATING）， + # 使 RetryCompensate 可再次补偿，避免事务永远补不回来 + compensation_failed = True + step.StepStatus = TransactionStep.STEP_STATUS_FAILED + step.StepResult.update({ + 'compensate_error': str(e), + 'compensate_status_code': e.StatusCode, + }) + step.EndTime = _utcnow() + step.save(update_fields=['StepStatus', 'StepResult', 'EndTime']) + logger.error(f'补偿步骤失败: {transaction_code} step={step.StepIndex}: {e}') + + if compensation_failed: + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_FAILED + else: + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_COMPENSATED + + dtx.EndTime = _utcnow() + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'EndTime', 'UpdateTime']) + + logger.info(f'分布式事务补偿完成: {transaction_code} -> {dtx.get_TransactionStatus_display()}') + return dtx + + @staticmethod + def RetryStep(transaction_code, step_index, access_token=None): + """重试失败步骤""" + try: + dtx = DistributedTransaction.objects.get(TransactionCode=transaction_code) + except DistributedTransaction.DoesNotExist: + raise TransactionError(f'事务不存在: {transaction_code}') + + try: + step = dtx.steps.get(StepIndex=step_index) + except TransactionStep.DoesNotExist: + raise TransactionError(f'步骤不存在: {step_index}') + + if step.StepStatus not in ( + TransactionStep.STEP_STATUS_FAILED, + TransactionStep.STEP_STATUS_PENDING, + ): + raise TransactionError(f'步骤状态不允许重试: {step.get_StepStatus_display()}') + + if not step.CanRetry: + raise TransactionError(f'步骤已达到最大重试次数: {step.MaxRetries}') + + step.RetryCount += 1 + step.StepStatus = TransactionStep.STEP_STATUS_PENDING + step.save(update_fields=['RetryCount', 'StepStatus']) + + try: + result = TransactionManager._execute_step(step, access_token=access_token) + step.StepStatus = TransactionStep.STEP_STATUS_COMPLETED + step.StepResult = result + step.EndTime = _utcnow() + step.save(update_fields=['StepStatus', 'StepResult', 'EndTime']) + except StepExecutionError as e: + step.StepStatus = TransactionStep.STEP_STATUS_FAILED + step.StepResult = { + 'error': str(e), + 'status_code': e.StatusCode, + 'response': e.ResponseBody, + 'retry_count': step.RetryCount, + } + step.EndTime = _utcnow() + step.save(update_fields=['StepStatus', 'StepResult', 'EndTime']) + + return step + + @staticmethod + @db_transaction.atomic + def RetryCompensate(transaction_code, step_index, access_token=None): + """重试补偿失败的步骤 + + 仅允许对补偿失败（StepResult 含 compensate_error）的步骤重试补偿； + 补偿成功后若不存在仍补偿失败的步骤，事务置 Compensated。 + + Args: + transaction_code: 事务代码 + step_index: 步骤序号 + access_token: 服务间认证令牌 + + Returns: + TransactionStep 实例 + """ + try: + dtx = DistributedTransaction.objects.select_for_update().get(TransactionCode=transaction_code) + except DistributedTransaction.DoesNotExist: + raise TransactionError(f'事务不存在: {transaction_code}') + + if dtx.TransactionStatus == DistributedTransaction.TX_STATUS_COMPENSATED: + raise TransactionError('事务已补偿完成') + + try: + step = dtx.steps.get(StepIndex=step_index) + except TransactionStep.DoesNotExist: + raise TransactionError(f'步骤不存在: {step_index}') + + if step.StepStatus != TransactionStep.STEP_STATUS_FAILED: + raise TransactionError(f'仅补偿失败的步骤可重试补偿: {step.get_StepStatus_display()}') + if 'compensate_error' not in (step.StepResult or {}): + raise TransactionError('该步骤为正向执行失败，请使用 RetryStep 重试执行') + + if not step.CanRetry: + raise TransactionError(f'步骤已达到最大重试次数: {step.MaxRetries}') + + step.RetryCount += 1 + step.StepStatus = TransactionStep.STEP_STATUS_COMPENSATING + step.save(update_fields=['RetryCount', 'StepStatus']) + + try: + result = TransactionManager._compensate_step(step, access_token=access_token) + step.StepStatus = TransactionStep.STEP_STATUS_COMPENSATED + step.StepResult.update({'compensate_result': result}) + step.EndTime = _utcnow() + step.save(update_fields=['StepStatus', 'StepResult', 'EndTime']) + except StepExecutionError as e: + step.StepStatus = TransactionStep.STEP_STATUS_FAILED + step.StepResult.update({ + 'compensate_error': str(e), + 'compensate_status_code': e.StatusCode, + }) + step.EndTime = _utcnow() + step.save(update_fields=['StepStatus', 'StepResult', 'EndTime']) + logger.error(f'补偿重试失败: {transaction_code} step={step.StepIndex}: {e}') + + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_FAILED + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'UpdateTime']) + return step + + # 补偿成功：若仍存在补偿失败的步骤则事务保持 FAILED，否则 COMPENSATED + remaining_failed = any( + 'compensate_error' in (s.StepResult or {}) + for s in dtx.steps.all() + ) + if remaining_failed: + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_FAILED + else: + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_COMPENSATED + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'UpdateTime']) + + logger.info(f'补偿重试完成: {transaction_code} step={step.StepIndex} -> {dtx.get_TransactionStatus_display()}') + return step + + @staticmethod + def GetDetail(transaction_code): + """获取事务详情（含所有步骤）""" + try: + dtx = DistributedTransaction.objects.prefetch_related('steps').get( + TransactionCode=transaction_code, + ) + except DistributedTransaction.DoesNotExist: + return None + + steps = [] + for step in dtx.steps.all().order_by('StepIndex'): + steps.append({ + 'Index': step.StepIndex, + 'Name': step.StepName, + 'Description': step.StepDescription, + 'Participant': step.ParticipantServiceUUID.ServiceName if step.ParticipantServiceUUID else None, + 'Status': step.get_StepStatus_display(), + 'StatusCode': step.StepStatus, + 'Result': step.StepResult, + 'RetryCount': step.RetryCount, + 'MaxRetries': step.MaxRetries, + 'StartTime': step.StartTime.isoformat() if step.StartTime else None, + 'EndTime': step.EndTime.isoformat() if step.EndTime else None, + }) + + return { + 'TransactionCode': dtx.TransactionCode, + 'Name': dtx.TransactionName, + 'Description': dtx.TransactionDescription, + 'Type': dtx.TransactionType, + 'Status': dtx.get_TransactionStatus_display(), + 'StatusCode': dtx.TransactionStatus, + 'Initiator': dtx.InitiatorServiceUUID.ServiceName if dtx.InitiatorServiceUUID else None, + 'Payload': dtx.TransactionPayload, + 'Result': dtx.TransactionResult, + 'TimeoutSeconds': dtx.TimeoutSeconds, + 'Steps': steps, + 'CreateTime': dtx.CreateTime.isoformat(), + 'EndTime': dtx.EndTime.isoformat() if dtx.EndTime else None, + } + + @staticmethod + def Cancel(transaction_code): + """取消待执行的事务""" + try: + dtx = DistributedTransaction.objects.get(TransactionCode=transaction_code) + except DistributedTransaction.DoesNotExist: + raise TransactionError(f'事务不存在: {transaction_code}') + + if dtx.TransactionStatus != DistributedTransaction.TX_STATUS_PENDING: + raise TransactionError(f'只有待执行状态的事务可以取消') + + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_FAILED + dtx.EndTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'EndTime']) + + dtx.steps.filter(StepStatus=TransactionStep.STEP_STATUS_PENDING).update( + StepStatus=TransactionStep.STEP_STATUS_SKIPPED, + EndTime=_utcnow(), + ) + + logger.info(f'分布式事务取消: {transaction_code}') + return dtx + + @staticmethod + def CheckTimeouts(): + """检测所有超时事务并触发补偿""" + timed_out = [] + running_txs = DistributedTransaction.objects.filter( + TransactionStatus__in=( + DistributedTransaction.TX_STATUS_PENDING, + DistributedTransaction.TX_STATUS_RUNNING, + ), + ) + + for dtx in running_txs: + if dtx.IsTimedOut: + TransactionManager._mark_timed_out(dtx) + try: + TransactionManager.Compensate(dtx.TransactionCode) + except TransactionError: + pass + timed_out.append(dtx.TransactionCode) + + if timed_out: + logger.warning(f'检测到 {len(timed_out)} 个超时事务: {timed_out}') + return timed_out + + @staticmethod + def QueryByStatus(status, limit=50): + """按状态查询事务""" + txs = DistributedTransaction.objects.filter( + TransactionStatus=status, + ).order_by('-CreateTime')[:limit] + + return [ + { + 'TransactionCode': tx.TransactionCode, + 'Name': tx.TransactionName, + 'Type': tx.TransactionType, + 'Status': tx.get_TransactionStatus_display(), + 'Initiator': tx.InitiatorServiceUUID.ServiceName if tx.InitiatorServiceUUID else None, + 'CreateTime': tx.CreateTime.isoformat(), + 'EndTime': tx.EndTime.isoformat() if tx.EndTime else None, + } + for tx in txs + ] + + # ─── 内部方法 ─────────────────────────────────────────── + + @staticmethod + def _mark_timed_out(dtx): + dtx.TransactionStatus = DistributedTransaction.TX_STATUS_TIMED_OUT + dtx.UpdateTime = _utcnow() + dtx.save(update_fields=['TransactionStatus', 'UpdateTime']) + + @staticmethod + def _get_service_domain(step): + """获取步骤参与服务的域名""" + if step.ParticipantServiceUUID: + return step.ParticipantServiceUUID.ServiceDomain.rstrip('/') + return None + + @staticmethod + def _execute_step(step, access_token=None): + """执行单个步骤——调用参与服务的本地事务管理器 execute 端点""" + step.StepStatus = TransactionStep.STEP_STATUS_RUNNING + step.StartTime = _utcnow() + step.save(update_fields=['StepStatus', 'StartTime']) + + service_domain = TransactionManager._get_service_domain(step) + if not service_domain: + raise StepExecutionError( + f'步骤无参与服务或服务域名未配置: step={step.StepIndex}', + step_index=step.StepIndex, + ) + + endpoint = f'{service_domain}{LOCAL_EXECUTE_PATH}' + payload = { + 'StepName': step.StepName, + 'payload': step.ActionPayload, + 'TransactionCode': step.TransactionUUID.TransactionCode, + 'StepIndex': step.StepIndex, + } + + return TransactionManager._call_service( + endpoint=endpoint, + payload=payload, + access_token=access_token, + ) + + @staticmethod + def _compensate_step(step, access_token=None): + """补偿单个步骤——调用参与服务的本地事务管理器 compensate 端点""" + service_domain = TransactionManager._get_service_domain(step) + if not service_domain: + raise StepExecutionError( + f'步骤无参与服务或服务域名未配置: step={step.StepIndex}', + step_index=step.StepIndex, + ) + + endpoint = f'{service_domain}{LOCAL_COMPENSATE_PATH}' + payload = { + 'StepName': step.StepName, + 'payload': step.CompensatePayload, + 'ActionResult': step.StepResult, + 'TransactionCode': step.TransactionUUID.TransactionCode, + 'StepIndex': step.StepIndex, + } + + return TransactionManager._call_service( + endpoint=endpoint, + payload=payload, + access_token=access_token, + ) + + @staticmethod + def _call_service(endpoint, payload=None, access_token=None): + """调用远程服务的本地事务管理器端点（使用连接池 + TLS 验证）""" + headers = {'Content-Type': 'application/json'} + if access_token: + headers['Authorization'] = f'Service {access_token}' + + try: + resp = _http_session.post( + endpoint, + json=payload or {}, + headers=headers, + timeout=30, + ) + response_data = resp.text + + # 非 2xx：无论响应体格式（HTML 错误页/网关页等）一律视为失败， + # 避免步骤被误判为 Completed + if not resp.ok: + raise StepExecutionError( + f'本地事务管理器调用失败: HTTP {resp.status_code}', + status_code=resp.status_code, + response_body=response_data[:4096], + ) + + content_type = resp.headers.get('Content-Type', '') + if 'application/json' in content_type: + try: + result = resp.json() + except ValueError: + # JSONDecodeError 是 ValueError 子类：响应体声明 JSON 却无法解析时， + # 转 StepExecutionError，保证失败必进补偿链路，事务不悬挂 Running + raise StepExecutionError( + f'本地事务管理器返回非法 JSON: {response_data[:512]}', + status_code=resp.status_code, + response_body=response_data[:4096], + ) + if isinstance(result, dict) and result.get('code') == 200: + return result.get('data', result) + elif isinstance(result, dict) and 'success' in result: + if not result['success']: + raise StepExecutionError( + result.get('error', '本地执行失败'), + response_body=response_data, + ) + return result.get('result', result) + return result + return {'raw': response_data} + except requests.exceptions.RequestException as e: + # 统一兜底：连接失败/超时/重试耗尽等任何 requests 异常 + # 一律转 StepExecutionError，确保失败必进补偿链路 + raise StepExecutionError(f'本地事务管理器调用失败: {e}') diff --git a/msync/urls.py b/msync/urls.py new file mode 100644 index 0000000..5af8f36 --- /dev/null +++ b/msync/urls.py @@ -0,0 +1,36 @@ +from django.urls import path +from gvsdsdk.msync import views + +app_name = 'msync' + +urlpatterns = [ + path('auth/token/', views.AuthTokenView.as_view(), name='auth_token'), + path('auth/token/refresh/', views.AuthTokenRefreshView.as_view(), name='auth_token_refresh'), + path('auth/introspect/', views.AuthTokenIntrospectView.as_view(), name='auth_introspect'), + + path('registry/', views.RegistryListView.as_view(), name='registry_list'), + path('registry//', views.RegistryDetailView.as_view(), name='registry_detail'), + path('registry//heartbeat/', views.RegistryHeartbeatView.as_view(), name='registry_heartbeat'), + path('registry//deregister/', views.RegistryDeregisterView.as_view(), name='registry_deregister'), + + path('ogm/', views.OGMRegisterView.as_view(), name='ogm_register'), + path('ogm/batch/', views.OGMBatchRegisterView.as_view(), name='ogm_batch_register'), + path('ogm/query/', views.OGMQueryView.as_view(), name='ogm_query'), + path('ogm//', views.OGMResolveView.as_view(), name='ogm_resolve'), + path('ogm//update/', views.OGMUpdateView.as_view(), name='ogm_update'), + path('ogm//deregister/', views.OGMDeregisterView.as_view(), name='ogm_deregister'), + + path('tx/', views.TransactionCreateView.as_view(), name='tx_create'), + path('tx/query/', views.TransactionQueryView.as_view(), name='tx_query'), + path('tx/timeout-check/', views.TransactionTimeoutCheckView.as_view(), name='tx_timeout_check'), + path('tx//', views.TransactionDetailView.as_view(), name='tx_detail'), + path('tx//execute/', views.TransactionExecuteView.as_view(), name='tx_execute'), + path('tx//compensate/', views.TransactionCompensateView.as_view(), name='tx_compensate'), + path('tx//cancel/', views.TransactionCancelView.as_view(), name='tx_cancel'), + path('tx//retry/', views.TransactionRetryStepView.as_view(), name='tx_retry_step'), + path('tx//retry-compensate/', views.TransactionRetryCompensateView.as_view(), name='tx_retry_compensate'), + + path('local/execute/', views.LocalExecuteView.as_view(), name='local_execute'), + path('local/compensate/', views.LocalCompensateView.as_view(), name='local_compensate'), + path('local/handlers/', views.LocalHandlersView.as_view(), name='local_handlers'), +] diff --git a/msync/views.py b/msync/views.py new file mode 100644 index 0000000..214316c --- /dev/null +++ b/msync/views.py @@ -0,0 +1,537 @@ +import logging +import traceback +from rest_framework.views import APIView +from rest_framework.response import Response +from rest_framework.permissions import IsAuthenticated +from gvsdsdk.msync.auth import ( + AuthenticateService, + RefreshServiceToken, + ValidateAccessToken, +) +from gvsdsdk.msync.middleware import ServiceTokenAuthentication +from gvsdsdk.msync.registry import ServiceRegistry, RegistryError +from gvsdsdk.msync.ogm import OGMManager, OGMError +from gvsdsdk.msync.transaction import TransactionManager, TransactionError +from gvsdsdk.msync.local_tx import local_tx +from gvsdsdk.msync.serializers import ( + AuthTokenRequestSerializer, + RefreshTokenRequestSerializer, + ServiceRegisterSerializer, + ServiceNodeSerializer, + OGMRegisterSerializer, + OGMBatchRegisterSerializer, + ServiceTokenResponseSerializer, + TransactionCreateSerializer, + TransactionExecuteSerializer, + TransactionRetryStepSerializer, + LocalExecuteSerializer, + LocalCompensateSerializer, +) + +logger = logging.getLogger(__name__) + + +def _ok(data=None, msg='success'): + return {'code': 200, 'msg': msg, 'data': data} + + +def _err(msg, code=400): + return {'code': code, 'msg': msg, 'data': None} + + +class AuthTokenView(APIView): + authentication_classes = [] + permission_classes = [] + + def post(self, request): + try: + ser = AuthTokenRequestSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + token = AuthenticateService( + service_name=ser.ValidatedData['ServiceName'], + signed_payload=ser.ValidatedData['Payload'], + signature=ser.ValidatedData['Signature'], + ) + if token is None: + return Response(_err('认证失败', 401), status=401) + + try: + scope = token.TokenScope + if not isinstance(scope, list): + scope = list(scope) if scope else [] + except Exception: + scope = [] + + try: + expires_ts = token.ExpiresAt.timestamp() + except Exception: + import time as _time + expires_ts = _time.time() + 7200 + + data = ServiceTokenResponseSerializer({ + 'AccessToken': token.AccessToken, + 'RefreshToken': token.RefreshToken, + 'Scope': scope, + 'ExpiresAt': expires_ts, + }).data + return Response(_ok(data)) + except Exception as e: + logger.error(f'认证令牌签发异常: {e}\n{traceback.format_exc()}') + return Response(_err('服务器内部错误', 500), status=500) + + +class AuthTokenRefreshView(APIView): + authentication_classes = [] + permission_classes = [] + + def post(self, request): + try: + ser = RefreshTokenRequestSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + token = RefreshServiceToken(ser.ValidatedData['RefreshToken']) + if token is None: + return Response(_err('刷新失败', 401), status=401) + + try: + scope = token.TokenScope + if not isinstance(scope, list): + scope = list(scope) if scope else [] + except Exception: + scope = [] + + try: + expires_ts = token.ExpiresAt.timestamp() + except Exception: + import time as _time + expires_ts = _time.time() + 7200 + + data = ServiceTokenResponseSerializer({ + 'AccessToken': token.AccessToken, + 'RefreshToken': token.RefreshToken, + 'Scope': scope, + 'ExpiresAt': expires_ts, + }).data + return Response(_ok(data)) + except Exception as e: + logger.error(f'令牌刷新异常: {e}\n{traceback.format_exc()}') + return Response(_err('服务器内部错误', 500), status=500) + + +class AuthTokenIntrospectView(APIView): + """Token 自省端点 — 供子服务器验证 ServiceToken 有效性 + + 子服务器无法直接访问主服务器的 Django ORM 和 MASTER_SECRET， + 因此通过此端点远程验证 token。 + """ + authentication_classes = [] + permission_classes = [] + + def post(self, request): + access_token = request.data.get('AccessToken', '') + if not access_token: + return Response(_err('AccessToken 不能为空', 400), status=400) + + try: + payload = ValidateAccessToken(access_token) + if payload is None: + return Response({'Valid': False, 'Active': False}) + return Response({'Valid': True, 'Active': True, 'Payload': payload}) + except Exception as e: + logger.error(f'Token 自省异常: {e}\n{traceback.format_exc()}') + return Response({'Valid': False, 'Active': False}) + + +class RegistryListView(APIView): + authentication_classes = [ServiceTokenAuthentication] + + def get(self, request): + if not request.user.is_authenticated: + return Response(_err('需要服务认证', 401), status=401) + services = ServiceRegistry.DiscoverAll(active_only=True) + return Response(_ok(services)) + + def post(self, request): + # 注册端点：新服务尚未拥有令牌，无需认证，但必须携带密钥 + # TODO: 考虑添加速率限制以防止滥用注册 + ser = ServiceRegisterSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + try: + node = ServiceRegistry.Register( + name=ser.ValidatedData['ServiceName'], + domain=ser.ValidatedData['ServiceDomain'], + secret_hash=ser.ValidatedData.get('ServiceSecretHash', ''), + raw_secret=ser.ValidatedData.get('ServiceSecret'), + scope=ser.ValidatedData.get('ServiceScope'), + service_type=ser.ValidatedData.get('ServiceType', 0), + metadata=ser.ValidatedData.get('ServiceMetadata', {}), + ) + except RegistryError as e: + return Response(_err(str(e), 409), status=409) + + data = ServiceNodeSerializer(node).data + return Response(_ok(data), status=201) + + +class RegistryDetailView(APIView): + authentication_classes = [ServiceTokenAuthentication] + + def get(self, request, name): + if not request.user.is_authenticated: + return Response(_err('需要服务认证', 401), status=401) + info = ServiceRegistry.Discover(name) + if info is None: + return Response(_err('服务不存在', 404), status=404) + return Response(_ok(info)) + + +class RegistryHeartbeatView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request, name): + # 仅允许持有该服务密钥签发的令牌执行本服务心跳 + if getattr(request, 'service_name', None) != name: + return Response(_err('无权操作其他服务', 403), status=403) + try: + ServiceRegistry.Heartbeat(name) + except RegistryError as e: + return Response(_err(str(e), 404), status=404) + return Response(_ok(msg='心跳成功')) + + +class RegistryDeregisterView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def delete(self, request, name): + # 仅允许持有该服务密钥签发的令牌注销本服务 + if getattr(request, 'service_name', None) != name: + return Response(_err('无权操作其他服务', 403), status=403) + try: + ServiceRegistry.Deregister(name) + except RegistryError as e: + return Response(_err(str(e), 404), status=404) + return Response(_ok(msg='注销成功')) + + +class OGMRegisterView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + ser = OGMRegisterSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + try: + ogm = OGMManager.Register( + object_type=ser.ValidatedData['ObjectType'], + local_id=ser.ValidatedData['ObjectLocalID'], + service_name=ser.ValidatedData['ServiceName'], + global_id=ser.ValidatedData.get('ObjectGlobalID'), + metadata=ser.ValidatedData.get('Metadata', {}), + ) + except OGMError as e: + return Response(_err(str(e), 409), status=409) + + return Response(_ok(OGMManager._to_dict(ogm)), status=201) + + +class OGMBatchRegisterView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + ser = OGMBatchRegisterSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + try: + results = OGMManager.RegisterBatch( + items=ser.ValidatedData['Items'], + service_name=ser.ValidatedData['ServiceName'], + ) + except OGMError as e: + return Response(_err(str(e), 409), status=409) + + data = [OGMManager._to_dict(ogm) for ogm in results] + return Response(_ok(data), status=201) + + +class OGMResolveView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def get(self, request, global_id): + info = OGMManager.Resolve(global_id) + if info is None: + return Response(_err('映射不存在', 404), status=404) + return Response(_ok(info)) + + +class OGMQueryView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def get(self, request): + ObjectType = request.query_params.get('ObjectType') + ServiceName = request.query_params.get('ServiceName') + LocalID = request.query_params.get('LocalID') + + if LocalID and ObjectType and ServiceName: + info = OGMManager.ResolveLocal(ObjectType, LocalID, ServiceName) + if info is None: + return Response(_err('映射不存在', 404), status=404) + return Response(_ok(info)) + + if ObjectType: + results = OGMManager.QueryByType(ObjectType) + return Response(_ok(results)) + + if ServiceName: + results = OGMManager.QueryByService(ServiceName, object_type=ObjectType) + return Response(_ok(results)) + + return Response(_err('请提供 ObjectType 或 ServiceName 参数', 400), status=400) + + +class OGMUpdateView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def put(self, request, global_id): + metadata = request.data.get('Metadata') + ogm = OGMManager.Update(global_id, metadata=metadata) + if ogm is None: + return Response(_err('映射不存在', 404), status=404) + return Response(_ok(OGMManager._to_dict(ogm))) + + +class OGMDeregisterView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def delete(self, request, global_id): + success = OGMManager.Deregister(global_id) + if not success: + return Response(_err('映射不存在', 404), status=404) + return Response(_ok(msg='注销成功')) + + +class TransactionCreateView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + ser = TransactionCreateSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + try: + dtx = TransactionManager.Create( + name=ser.ValidatedData['TransactionName'], + steps_data=ser.ValidatedData['Steps'], + description=ser.ValidatedData.get('TransactionDescription', ''), + transaction_type=ser.ValidatedData.get('TransactionType', 'saga'), + initiator_service_name=ser.ValidatedData.get('InitiatorServiceName'), + reference_ogm_uuid=ser.ValidatedData.get('ReferenceOGMUUID'), + payload=ser.ValidatedData.get('Payload'), + timeout_seconds=ser.ValidatedData.get('TimeoutSeconds', 300), + ) + except TransactionError as e: + return Response(_err(str(e), 400), status=400) + + return Response(_ok(TransactionManager.GetDetail(dtx.TransactionCode)), status=201) + + +class TransactionExecuteView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request, transaction_code): + ser = TransactionExecuteSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + try: + dtx = TransactionManager.Execute( + transaction_code, + access_token=ser.ValidatedData.get('AccessToken'), + ) + except TransactionError as e: + return Response(_err(str(e), 400), status=400) + + return Response(_ok(TransactionManager.GetDetail(dtx.TransactionCode))) + + +class TransactionCompensateView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request, transaction_code): + access_token = request.data.get('AccessToken') + try: + dtx = TransactionManager.Compensate( + transaction_code, + access_token=access_token, + ) + except TransactionError as e: + return Response(_err(str(e), 400), status=400) + + return Response(_ok(TransactionManager.GetDetail(dtx.TransactionCode))) + + +class TransactionCancelView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request, transaction_code): + try: + TransactionManager.Cancel(transaction_code) + except TransactionError as e: + return Response(_err(str(e), 400), status=400) + + return Response(_ok(msg='事务已取消')) + + +class TransactionDetailView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def get(self, request, transaction_code): + detail = TransactionManager.GetDetail(transaction_code) + if detail is None: + return Response(_err('事务不存在', 404), status=404) + return Response(_ok(detail)) + + +class TransactionRetryStepView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request, transaction_code): + ser = TransactionRetryStepSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + try: + step = TransactionManager.RetryStep( + transaction_code, + step_index=ser.ValidatedData['StepIndex'], + access_token=ser.ValidatedData.get('AccessToken'), + ) + except TransactionError as e: + return Response(_err(str(e), 400), status=400) + + return Response(_ok({ + 'StepIndex': step.StepIndex, + 'StepName': step.StepName, + 'Status': step.get_StepStatus_display(), + 'RetryCount': step.RetryCount, + })) + + +class TransactionRetryCompensateView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request, transaction_code): + ser = TransactionRetryStepSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + try: + step = TransactionManager.RetryCompensate( + transaction_code, + step_index=ser.ValidatedData['StepIndex'], + access_token=ser.ValidatedData.get('AccessToken'), + ) + except TransactionError as e: + return Response(_err(str(e), 400), status=400) + + return Response(_ok({ + 'StepIndex': step.StepIndex, + 'StepName': step.StepName, + 'Status': step.get_StepStatus_display(), + 'RetryCount': step.RetryCount, + 'TransactionStatus': step.TransactionUUID.get_TransactionStatus_display(), + })) + + +class TransactionTimeoutCheckView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + timed_out = TransactionManager.CheckTimeouts() + return Response(_ok({'TimedOut': timed_out, 'Count': len(timed_out)})) + + +class TransactionQueryView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def get(self, request): + status = request.query_params.get('status') + if status is None: + return Response(_err('请提供 status 参数', 400), status=400) + + try: + status_int = int(status) + except ValueError: + return Response(_err('status 必须为整数', 400), status=400) + + results = TransactionManager.QueryByStatus(status_int) + return Response(_ok(results)) + + +class LocalExecuteView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + ser = LocalExecuteSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + result = local_tx.ExecuteStep( + name=ser.ValidatedData['StepName'], + payload=ser.ValidatedData.get('Payload'), + ) + + if result['success']: + return Response(_ok(result['result'])) + return Response(_err(result['error'], 500), status=500) + + +class LocalCompensateView(APIView): + authentication_classes = [ServiceTokenAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + ser = LocalCompensateSerializer(data=request.data) + if not ser.IsValid(): + return Response(_err(ser.errors, 400), status=400) + + result = local_tx.CompensateStep( + name=ser.ValidatedData['StepName'], + payload=ser.ValidatedData.get('Payload'), + action_result=ser.ValidatedData.get('ActionResult'), + ) + + if result['success']: + return Response(_ok(result['result'])) + return Response(_err(result['error'], 500), status=500) + + +class LocalHandlersView(APIView): + + def get(self, request): + handlers = local_tx.ListHandlers() + return Response(_ok(handlers)) diff --git a/secrets.py b/secrets.py new file mode 100644 index 0000000..5dd4b48 --- /dev/null +++ b/secrets.py @@ -0,0 +1,66 @@ +"""GVSDS SDK 配置与密钥访问 + +环境变量优先，其次读取项目根目录 secrets.ini。 +与 gvsds.secrets 同源（同一 secrets.ini），供 SDK 在无 gvsds 依赖时使用。 + +env 键规则：`{SECTION}_{KEY}` 全大写（如 MSYNC_MASTER_SECRET）。 +""" + +import os +import configparser +from pathlib import Path + +_SECRETS_FILE = Path(__file__).resolve().parent.parent / "secrets.ini" +_config = None + + +def _load_config(): + global _config + if _config is not None: + return _config + _config = configparser.ConfigParser() + if _SECRETS_FILE.exists(): + _config.read(str(_SECRETS_FILE), encoding="utf-8") + return _config + + +def get(section, key, fallback=None): + cfg = _load_config() + env_key = f"{section.upper()}_{key.upper()}" + env_val = os.environ.get(env_key) + if env_val is not None: + return env_val + return cfg.get(section, key, fallback=fallback) + + +def getint(section, key, fallback=0): + val = get(section, key, fallback=None) + if val is None: + return fallback + try: + return int(val) + except (ValueError, TypeError): + return fallback + + +def getfloat(section, key, fallback=0.0): + val = get(section, key, fallback=None) + if val is None: + return fallback + try: + return float(val) + except (ValueError, TypeError): + return fallback + + +def getboolean(section, key, fallback=False): + val = get(section, key, fallback=None) + if val is None: + return fallback + return val.lower() in ("true", "1", "yes", "on") + + +def reload(): + global _config + _config = None + _load_config() diff --git a/serializer_base.py b/serializer_base.py new file mode 100644 index 0000000..84ef00a --- /dev/null +++ b/serializer_base.py @@ -0,0 +1,26 @@ +"""GVSDS SDK DRF 序列化器基类 + +提供 PascalCase 命名风格的 DRF 序列化器基类（is_valid/validated_data 的 +PascalCase 别名）。gvsds.common.Serializer 从本模块 re-export。""" + +from rest_framework import serializers + + +class PascalCaseSerializer(serializers.Serializer): + + def IsValid(self, raise_exception=False): + return self.is_valid(raise_exception=raise_exception) + + @property + def ValidatedData(self): + return self.validated_data + + +class PascalCaseModelSerializer(serializers.ModelSerializer): + + def IsValid(self, raise_exception=False): + return self.is_valid(raise_exception=raise_exception) + + @property + def ValidatedData(self): + return self.validated_data