diff --git a/Main.py b/Main.py index a1570853..15df77b8 100644 --- a/Main.py +++ b/Main.py @@ -539,7 +539,7 @@ def get_django_wsgi_application(): app = get_wsgi_application() app = FilteredLoggingMiddleware(app) app = SecurityMiddleware(app) - from gvsds.apps.msync.sio import create_sio_app + from gvsdsdk.msync.sio import create_sio_app app = create_sio_app(app) return app diff --git a/gvsds/apps/gerp/views/msync_admin.py b/gvsds/apps/gerp/views/msync_admin.py index 7da3a3b1..266fd831 100644 --- a/gvsds/apps/gerp/views/msync_admin.py +++ b/gvsds/apps/gerp/views/msync_admin.py @@ -2,8 +2,8 @@ from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from . import ELTAuthenticatedAPIView from gvsdsdk.auth.permissions import PermissionVerifier -from gvsds.apps.msync.registry import ServiceRegistry -from gvsds.apps.msync.models import _utcnow +from gvsdsdk.msync.registry import ServiceRegistry +from gvsdsdk.msync.models import _utcnow import datetime import logging @@ -134,13 +134,13 @@ class MSyncServerDeregisterView(ELTAuthenticatedAPIView): # 通过 SIO 通知被注销的服务 try: - from gvsds.apps.msync.sio import sio, _service_sid_map, _map_lock, NS + from gvsdsdk.msync.sio import sio, _service_sid_map, _map_lock, NS with _map_lock: sid = _service_sid_map.pop(name, None) if sid: _sid_service_map_pop = None # 从 sid 映射中也移除 - from gvsds.apps.msync.sio import _sid_service_map + from gvsdsdk.msync.sio import _sid_service_map _sid_service_map.pop(sid, None) if sid: sio.emit('msync:force:deregister', { @@ -152,7 +152,7 @@ class MSyncServerDeregisterView(ELTAuthenticatedAPIView): # 广播服务下线 try: - from gvsds.apps.msync.sio import sio, NS + from gvsdsdk.msync.sio import sio, NS sio.emit('msync:service:offline', {'Service': name}, namespace=NS) except Exception: pass @@ -171,7 +171,7 @@ class MSyncServerFormProxyView(ELTAuthenticatedAPIView): def _execute_action(self, name, action, data): """通过 MSYNC Socket.IO 中转 ORM 请求到子服务器""" - from gvsds.apps.msync.sio import sio, _service_sid_map, _map_lock, NS + from gvsdsdk.msync.sio import sio, _service_sid_map, _map_lock, NS # 查找子服务器 SID with _map_lock: diff --git a/gvsds/apps/gerp/views/user_detail.py b/gvsds/apps/gerp/views/user_detail.py index 42c191ab..ac3d1f94 100644 --- a/gvsds/apps/gerp/views/user_detail.py +++ b/gvsds/apps/gerp/views/user_detail.py @@ -73,7 +73,7 @@ class UserDetailView(ELTAuthenticatedAPIView): def _query_sub(self, server_name, user_uuid_str): """子服务器：通过 form proxy 查询 User 表""" - from gvsds.apps.msync.registry import ServiceRegistry + from gvsdsdk.msync.registry import ServiceRegistry from gvsds.apps.gerp.views.msync_admin import _get_sub_db_connection node = ServiceRegistry.GetNode(server_name) @@ -237,7 +237,7 @@ class UserListAPIView(ELTAuthenticatedAPIView): def _query_sub(self, server_name, page_index, page_size, keyword): """子服务器：通过 form proxy 查询""" - from gvsds.apps.msync.registry import ServiceRegistry + from gvsdsdk.msync.registry import ServiceRegistry from gvsds.apps.gerp.views.msync_admin import _get_sub_db_connection node = ServiceRegistry.GetNode(server_name) diff --git a/gvsds/apps/msync/__init__.py b/gvsds/apps/msync/__init__.py deleted file mode 100644 index 84352b03..00000000 --- a/gvsds/apps/msync/__init__.py +++ /dev/null @@ -1 +0,0 @@ -default_app_config = 'gvsds.apps.msync.apps.MsyncConfig' diff --git a/gvsds/apps/msync/auth.py b/gvsds/apps/msync/auth.py deleted file mode 100644 index 747affee..00000000 --- a/gvsds/apps/msync/auth.py +++ /dev/null @@ -1,172 +0,0 @@ -"""模块：服务间认证 - -Django ORM 依赖的认证函数，纯函数已移至 gvsdsdk.msync.auth。 -""" - -import traceback -import datetime -import logging - -from gvsdsdk.msync.auth import ( - SignPayload, - VerifySignature, - GenerateChallenge, - BuildAuthPayload, - SignAuthPayload, - VerifyAuthPayload, - ServiceTokenEncoder, -) - -logger = logging.getLogger(__name__) - - -def _get_master_secret(): - from gvsds.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 gvsds.apps.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 gvsds.apps.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 - - # 调试：对比签名 - from gvsdsdk.msync.auth import SignAuthPayload - expected_sig = SignAuthPayload(signed_payload, raw_secret) - logger.info(f'[AUTH] 签名对比: expected={expected_sig}, actual={signature}, match={expected_sig == signature}') - - 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 gvsds.apps.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 gvsds.apps.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/gvsds/apps/msync/client.py b/gvsds/apps/msync/client.py deleted file mode 100644 index ca6adc39..00000000 --- a/gvsds/apps/msync/client.py +++ /dev/null @@ -1,3 +0,0 @@ -"""MSYNC 服务客户端——从 gvsdsdk 导入""" - -from gvsdsdk.msync.client import ServiceClient, ServiceClientError # noqa diff --git a/gvsds/apps/msync/local_tx.py b/gvsds/apps/msync/local_tx.py deleted file mode 100644 index d99a3bbe..00000000 --- a/gvsds/apps/msync/local_tx.py +++ /dev/null @@ -1,3 +0,0 @@ -"""本地事务管理器——从 gvsdsdk 导入""" - -from gvsdsdk.msync.local_tx import LocalTransactionManager, StepHandler, local_tx # noqa diff --git a/gvsds/apps/msync/migrations/__init__.py b/gvsds/apps/msync/migrations/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/gvsds/apps/msync/ogm.py b/gvsds/apps/msync/ogm.py deleted file mode 100644 index ad892581..00000000 --- a/gvsds/apps/msync/ogm.py +++ /dev/null @@ -1,267 +0,0 @@ -"""模块：对象全局映射（OGM）管理 -提供跨服务对象引用的注册、解析、更新、注销操作。 -OGM 是微服务间共享对象身份的核心机制： - - 每个本地对象注册后获得全局唯一 ID - - 其他服务通过全局 ID 解析到对象归属与本地 ID - - 支持按类型、按服务查询对象映射""" - -import uuid -import datetime -import logging -from gvsds.apps.msync.models import ObjectGlobalMapping, ServiceNode, _utcnow - -logger = logging.getLogger(__name__) - - -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, - } \ No newline at end of file diff --git a/gvsds/apps/msync/registry.py b/gvsds/apps/msync/registry.py deleted file mode 100644 index 86fb6813..00000000 --- a/gvsds/apps/msync/registry.py +++ /dev/null @@ -1,263 +0,0 @@ -"""模块：服务注册与发现 -提供服务节点的注册、注销、发现、心跳等操作。""" - -import uuid -import datetime -import logging -from gvsds.apps.msync.models import ServiceNode, _utcnow - -logger = logging.getLogger(__name__) - - -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 gvsds.apps.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, - ) \ No newline at end of file diff --git a/gvsds/apps/msync/transaction.py b/gvsds/apps/msync/transaction.py deleted file mode 100644 index 9ded9ceb..00000000 --- a/gvsds/apps/msync/transaction.py +++ /dev/null @@ -1,668 +0,0 @@ -"""模块：远程事务管理器 -基于 Saga 模式的跨服务事务编排引擎，运行于主服务器（账本服务器）。 -通过调用各工作服务的本地事务管理器端点来执行与补偿步骤。 - -架构： - 主服务器 (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() - -事务生命周期： - Pending → Running → Committed - ↘ Failed → Compensating → Compensated - ↘ TimedOut → Compensating → Compensated""" - -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 gvsds.apps.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}') \ No newline at end of file diff --git a/gvsds/common/Serializer.py b/gvsds/common/Serializer.py index e63a4801..576a5984 100644 --- a/gvsds/common/Serializer.py +++ b/gvsds/common/Serializer.py @@ -1,21 +1,2 @@ -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 \ No newline at end of file +# Re-export from gvsdsdk — DRF 序列化器基类已迁移至 SDK +from gvsdsdk.serializer_base import PascalCaseSerializer, PascalCaseModelSerializer # noqa: F401 diff --git a/gvsds/settings.py b/gvsds/settings.py index 129348cd..b9ebb33c 100644 --- a/gvsds/settings.py +++ b/gvsds/settings.py @@ -62,7 +62,7 @@ INSTALLED_APPS = _app_getlist('apps', 'INSTALLED_APPS', fallback=[ 'gvsds.apps.mc', 'gvsds.apps.doc', 'gvsds.apps.api', - 'gvsds.apps.msync', + 'gvsdsdk.msync', 'gvsds.apps.payment', 'gvsds.apps.commission', 'gvsds.apps.stats', @@ -80,7 +80,7 @@ MIDDLEWARE = _app_getlist('middleware', 'MIDDLEWARE', fallback=[ 'gvsds.ip_filter.IPFilterMiddleware', 'gvsds.middleware.DomainMiddleware', 'gvsds.middleware.CurrentRoleMiddleware', - 'gvsds.apps.msync.middleware.ServiceAuthMiddleware', + 'gvsdsdk.msync.middleware.ServiceAuthMiddleware', 'gvsds.middleware.AttackDetectMiddleware', 'gvsds.middleware.ErrorLoggingMiddleware', ]) diff --git a/gvsds/urls.py b/gvsds/urls.py index fe006276..669f36e6 100644 --- a/gvsds/urls.py +++ b/gvsds/urls.py @@ -29,7 +29,7 @@ urlpatterns = [ path('api/t4/', include('gvsds.apps.t4.urls')), path('api/doc/', include('gvsds.apps.doc.urls')), path('api/', include('gvsds.apps.api.urls')), - path('api/msync/', include('gvsds.apps.msync.urls')), + path('api/msync/', include('gvsdsdk.msync.urls')), path('api/payment/', include('gvsds.apps.payment.urls')), path('api/commission/', include('gvsds.apps.commission.urls')), path('api/stats/', include('gvsds.apps.stats.urls')), diff --git a/gvsds/apps/msync/MSYNC.md b/gvsdsdk/msync/MSYNC.md similarity index 91% rename from gvsds/apps/msync/MSYNC.md rename to gvsdsdk/msync/MSYNC.md index def485c8..05f27111 100644 --- a/gvsds/apps/msync/MSYNC.md +++ b/gvsdsdk/msync/MSYNC.md @@ -2,6 +2,7 @@ > GVsds 平台微服务架构的多端通信、服务间验权、对象全局映射（OGM）与分布式事务管理。 > 本服务器是**主服务器**（账本服务器），存储所有分布式服务地址，服务间需验权来互相获取 OGM 等信息。 +> 实现位于 `gvsdsdk/msync`（SDK 原生实现，非封装层）。 --- @@ -93,6 +94,12 @@ MSync 是为 GVsds 多域微服务平台设计的通信中间件，运行于** | `tx:write` | 创建/执行/补偿事务 | | `*` | 全部权限 | +### 密钥认证强制 + +- 新服务注册必须携带 `ServiceSecret`（与 `ServiceSecretHash` 自洽），否则拒绝 +- 除 register / auth / auth:refresh 外，所有 REST 与 SIO 业务操作必须持密钥签发的令牌 +- 心跳/注销仅允许操作本服务 + --- ## 分布式事务管理 @@ -135,7 +142,7 @@ MSync 采用**远程事务管理器 + 本地事务管理器**双层架构： - **正向执行**：远程管理器按步骤顺序调用各服务的本地 execute 端点 - **补偿回滚**：任一步骤失败时，远程管理器逆序调用已完成步骤的本地 compensate 端点 - **超时检测**：自动标记超时事务并触发补偿 -- **重试机制**：步骤失败后可手动重试（可配置最大次数） +- **重试机制**：正向步骤失败可 `RetryStep`，补偿失败可 `RetryCompensate`（可配置最大次数） ### 事务生命周期 @@ -269,10 +276,10 @@ Pending ──▶ Running ──▶ Committed (全部步骤成功) | 方法 | 端点 | 说明 | |------|------|------| | GET | `/api/msync/registry/` | 列出所有已注册服务 | -| POST | `/api/msync/registry/` | 注册新服务 | +| POST | `/api/msync/registry/` | 注册新服务（必须携带密钥） | | GET | `/api/msync/registry//` | 发现指定服务 | -| POST | `/api/msync/registry//heartbeat/` | 服务心跳 | -| DELETE | `/api/msync/registry//deregister/` | 注销服务 | +| POST | `/api/msync/registry//heartbeat/` | 服务心跳（需令牌，仅本服务） | +| DELETE | `/api/msync/registry//deregister/` | 注销服务（需令牌，仅本服务） | ### OGM 对象全局映射 @@ -296,7 +303,8 @@ Pending ──▶ Running ──▶ Committed (全部步骤成功) | 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/` | 重试正向失败步骤 | +| POST | `/api/msync/tx//retry-compensate/` | 重试补偿失败步骤 | ### 本地事务管理器（每个工作服务） @@ -360,6 +368,15 @@ POST /api/msync/tx/DTX-ABC123DEF456/retry/ } ``` +**重试补偿失败步骤**: +```json +POST /api/msync/tx/DTX-ABC123DEF456/retry-compensate/ +{ + "step_index": 1, + "access_token": "eyJhbGci..." +} +``` + --- ## Python API @@ -367,7 +384,7 @@ POST /api/msync/tx/DTX-ABC123DEF456/retry/ ### 服务注册（账本操作） ```python -from gvsds.apps.msync.registry import ServiceRegistry, RegistryError +from gvsdsdk.msync.registry import ServiceRegistry, RegistryError node = ServiceRegistry.Register( name='gerp', @@ -384,7 +401,7 @@ ServiceRegistry.Deregister('gerp') ### 服务客户端 ```python -from gvsds.apps.msync.client import ServiceClient +from gvsdsdk.msync.client import ServiceClient client = ServiceClient('gerp', 'secret', 'https://api.gvsds.com') client.Authenticate() @@ -395,7 +412,7 @@ info = client.DiscoverService('crb') ### OGM 管理 ```python -from gvsds.apps.msync.ogm import OGMManager +from gvsdsdk.msync.ogm import OGMManager ogm = OGMManager.Register('user', 'uuid-1', 'core', global_id='user:uuid1') info = OGMManager.Resolve('user:uuid1') @@ -406,7 +423,7 @@ OGMManager.Deregister('user:uuid1') ### 分布式事务 ```python -from gvsds.apps.msync.transaction import TransactionManager, TransactionError +from gvsdsdk.msync.transaction import TransactionManager, TransactionError # 创建事务 dtx = TransactionManager.Create( @@ -437,9 +454,12 @@ 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) @@ -453,7 +473,7 @@ running = TransactionManager.QueryByStatus(1) ### 本地事务管理器（工作服务端） ```python -from gvsds.apps.msync.local_tx import local_tx +from gvsdsdk.msync.local_tx import local_tx # 注册步骤处理器（在服务启动时） def create_user_action(payload): @@ -480,7 +500,7 @@ local_tx.unregister('create_user') ### 权限装饰器 ```python -from gvsds.apps.msync.middleware import require_service_scope, ServicePermission +from gvsdsdk.msync.middleware import require_service_scope, ServicePermission @require_service_scope('ogm:read') def my_view(request): @@ -498,11 +518,11 @@ class MyViewSet(ViewSet): ```python INSTALLED_APPS = [ - 'gvsds.apps.msync', + 'gvsdsdk.msync', ] MIDDLEWARE = [ - 'gvsds.apps.msync.middleware.ServiceAuthMiddleware', + 'gvsdsdk.msync.middleware.ServiceAuthMiddleware', ] ``` @@ -524,13 +544,14 @@ MASTER_SECRET = your-master-secret-for-token-signing | 文件 | 说明 | |------|------| | `models.py` | 5 个数据模型（ServiceNode, ServiceToken, ObjectGlobalMapping, DistributedTransaction, TransactionStep） | -| `auth.py` | HMAC 签名 + 自包含令牌编解码 | +| `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 路由 | diff --git a/gvsdsdk/msync/__init__.py b/gvsdsdk/msync/__init__.py index a07283b7..6b60a0e9 100644 --- a/gvsdsdk/msync/__init__.py +++ b/gvsdsdk/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/gvsds/apps/msync/admin.py b/gvsdsdk/msync/admin.py similarity index 97% rename from gvsds/apps/msync/admin.py rename to gvsdsdk/msync/admin.py index ec33dbb1..57be602d 100644 --- a/gvsds/apps/msync/admin.py +++ b/gvsdsdk/msync/admin.py @@ -1,5 +1,5 @@ from django.contrib import admin -from gvsds.apps.msync.models import ( +from gvsdsdk.msync.models import ( ServiceNode, ServiceToken, ObjectGlobalMapping, diff --git a/gvsds/apps/msync/apps.py b/gvsdsdk/msync/apps.py similarity index 70% rename from gvsds/apps/msync/apps.py rename to gvsdsdk/msync/apps.py index 61431011..277479a0 100644 --- a/gvsds/apps/msync/apps.py +++ b/gvsdsdk/msync/apps.py @@ -3,7 +3,7 @@ from django.apps import AppConfig class MsyncConfig(AppConfig): default_auto_field = 'django.db.models.BigAutoField' - name = 'gvsds.apps.msync' + name = 'gvsdsdk.msync' verbose_name = '微服务通信' verbose_name_plural = '微服务通信' @@ -17,10 +17,10 @@ class MsyncConfig(AppConfig): pass # 数据库可能尚未迁移 def _ensure_master_node(self): - from gvsds.apps.msync.models import ServiceNode, _utcnow - from gvsds.app_config import get as _app_get + from gvsdsdk.msync.models import ServiceNode, _utcnow + import uuid - service_domain = _app_get('msync', 'SERVICE_DOMAIN', fallback='https://www.gvsds.com') + service_domain = self._resolve_service_domain() master_uuid = uuid.UUID('00000000-0000-0000-0000-000000000001') try: @@ -51,5 +51,18 @@ class MsyncConfig(AppConfig): node.SetSecret(secret) node.save() - -import uuid + @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/gvsdsdk/msync/auth.py b/gvsdsdk/msync/auth.py index 61cca98c..89888fbb 100644 --- a/gvsdsdk/msync/auth.py +++ b/gvsdsdk/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/gvsds/apps/msync/middleware.py b/gvsdsdk/msync/middleware.py similarity index 98% rename from gvsds/apps/msync/middleware.py rename to gvsdsdk/msync/middleware.py index 366c5bbc..f10368a1 100644 --- a/gvsds/apps/msync/middleware.py +++ b/gvsdsdk/msync/middleware.py @@ -6,7 +6,7 @@ import functools import logging from django.http import JsonResponse from rest_framework.response import Response -from gvsds.apps.msync.auth import ValidateAccessToken +from gvsdsdk.msync.auth import ValidateAccessToken logger = logging.getLogger(__name__) diff --git a/gvsds/apps/msync/migrations/0001_initial.py b/gvsdsdk/msync/migrations/0001_initial.py similarity index 90% rename from gvsds/apps/msync/migrations/0001_initial.py rename to gvsdsdk/msync/migrations/0001_initial.py index 338c61a5..10021727 100644 --- a/gvsds/apps/msync/migrations/0001_initial.py +++ b/gvsdsdk/msync/migrations/0001_initial.py @@ -1,23 +1,20 @@ -import uuid import datetime from django.db import migrations, models import django.db.models.deletion -import gvsds.apps.core.models +import gvsdsdk.models class Migration(migrations.Migration): initial = True - dependencies = [ - ('core', '0002_user_is_staff_user_is_superuser'), - ] + dependencies = [] operations = [ migrations.CreateModel( name='ServiceNode', fields=[ - ('ServiceUUID', gvsds.apps.core.models.UUIDField(db_column='ServiceUUID', editable=False, primary_key=True, serialize=False)), + ('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)), @@ -38,7 +35,7 @@ class Migration(migrations.Migration): migrations.CreateModel( name='ServiceToken', fields=[ - ('TokenUUID', gvsds.apps.core.models.UUIDField(db_column='TokenUUID', editable=False, primary_key=True, serialize=False)), + ('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)), @@ -58,7 +55,7 @@ class Migration(migrations.Migration): migrations.CreateModel( name='ObjectGlobalMapping', fields=[ - ('OGMUUID', gvsds.apps.core.models.UUIDField(db_column='OGMUUID', editable=False, primary_key=True, serialize=False)), + ('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)), @@ -87,13 +84,13 @@ class Migration(migrations.Migration): migrations.CreateModel( name='DistributedTransaction', fields=[ - ('TransactionUUID', gvsds.apps.core.models.UUIDField(db_column='TransactionUUID', editable=False, primary_key=True, serialize=False)), + ('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', gvsds.apps.core.models.UUIDField(blank=True, db_column='ReferenceOGMUUID', null=True)), + ('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)), @@ -120,7 +117,7 @@ class Migration(migrations.Migration): migrations.CreateModel( name='TransactionStep', fields=[ - ('StepUUID', gvsds.apps.core.models.UUIDField(db_column='StepUUID', editable=False, primary_key=True, serialize=False)), + ('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)), diff --git a/gvsds/apps/msync/migrations/0002_add_service_secret_encrypted.py b/gvsdsdk/msync/migrations/0002_add_service_secret_encrypted.py similarity index 98% rename from gvsds/apps/msync/migrations/0002_add_service_secret_encrypted.py rename to gvsdsdk/msync/migrations/0002_add_service_secret_encrypted.py index 16c23d5b..b16f8716 100644 --- a/gvsds/apps/msync/migrations/0002_add_service_secret_encrypted.py +++ b/gvsdsdk/msync/migrations/0002_add_service_secret_encrypted.py @@ -13,4 +13,4 @@ class Migration(migrations.Migration): name='ServiceSecretEncrypted', field=models.CharField(db_column='ServiceSecretEncrypted', default='', max_length=512), ), - ] \ No newline at end of file + ] diff --git a/gvsdsdk/msync/migrations/__init__.py b/gvsdsdk/msync/migrations/__init__.py new file mode 100644 index 00000000..40a96afc --- /dev/null +++ b/gvsdsdk/msync/migrations/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 -*- diff --git a/gvsds/apps/msync/models.py b/gvsdsdk/msync/models.py similarity index 97% rename from gvsds/apps/msync/models.py rename to gvsdsdk/msync/models.py index 3b6effc4..0932529e 100644 --- a/gvsds/apps/msync/models.py +++ b/gvsdsdk/msync/models.py @@ -1,12 +1,14 @@ """模块：MSync 数据模型 微服务通信核心表：服务节点、服务令牌、对象全局映射（OGM）、分布式事务。 -所有 ORM 字段与方法遵循帕斯卡命名法。""" +所有 ORM 字段与方法遵循帕斯卡命名法。 + +模型归属 app_label='msync'，与 gvsds 主服务器 7 库路由及既有迁移兼容。""" import uuid import datetime from django.db import models -from gvsds.model_base import QModel -from gvsds.apps.core.models import UUIDField +from gvsdsdk.model_base import QModel +from gvsdsdk.models import UUIDField def _utcnow(): @@ -59,7 +61,7 @@ class ServiceNode(QModel): def SetSecret(self, raw_secret): from gvsdsdk.auth.crypto import SymmetricEncrypt - from gvsds.secrets import get as _secret_get + 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) @@ -67,7 +69,7 @@ class ServiceNode(QModel): if not self.ServiceSecretEncrypted: return None from gvsdsdk.auth.crypto import SymmetricDecrypt - from gvsds.secrets import get as _secret_get + 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) diff --git a/gvsdsdk/msync/ogm.py b/gvsdsdk/msync/ogm.py index 3f0d5665..702800b5 100644 --- a/gvsdsdk/msync/ogm.py +++ b/gvsdsdk/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/gvsdsdk/msync/registry.py b/gvsdsdk/msync/registry.py index 7397d6a7..fac1d33d 100644 --- a/gvsdsdk/msync/registry.py +++ b/gvsdsdk/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/gvsds/apps/msync/serializers.py b/gvsdsdk/msync/serializers.py similarity index 94% rename from gvsds/apps/msync/serializers.py rename to gvsdsdk/msync/serializers.py index a1b69801..1807f102 100644 --- a/gvsds/apps/msync/serializers.py +++ b/gvsdsdk/msync/serializers.py @@ -1,12 +1,6 @@ from rest_framework import serializers -from gvsds.common.Serializer import PascalCaseSerializer, PascalCaseModelSerializer -from gvsds.apps.msync.models import ( - ServiceNode, - ServiceToken, - ObjectGlobalMapping, - DistributedTransaction, - TransactionStep, -) +from gvsdsdk.serializer_base import PascalCaseSerializer, PascalCaseModelSerializer +from gvsdsdk.msync.models import ServiceNode class ServiceNodeSerializer(PascalCaseModelSerializer): @@ -101,4 +95,4 @@ class LocalCompensateSerializer(PascalCaseSerializer): 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) \ No newline at end of file + StepIndex = serializers.IntegerField(required=False, allow_null=True) diff --git a/gvsds/apps/msync/sio.py b/gvsdsdk/msync/sio.py similarity index 91% rename from gvsds/apps/msync/sio.py rename to gvsdsdk/msync/sio.py index b6fa1b5b..a13892c4 100644 --- a/gvsds/apps/msync/sio.py +++ b/gvsdsdk/msync/sio.py @@ -96,7 +96,7 @@ def on_disconnect(sid): _sid_auth_map.pop(sid, None) if service_name: try: - from gvsds.apps.msync.registry import ServiceRegistry + from gvsdsdk.msync.registry import ServiceRegistry ServiceRegistry.Deregister(service_name) except Exception: pass @@ -109,7 +109,7 @@ def on_disconnect(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 gvsds.apps.msync.registry import ServiceRegistry, RegistryError + from gvsdsdk.msync.registry import ServiceRegistry, RegistryError try: node = ServiceRegistry.Register( name=_d(data, 'ServiceName', ''), @@ -139,7 +139,7 @@ def on_register(sid, data): 'ServiceDomain': _d(data, 'ServiceDomain'), }, namespace=NS, skip_sid=sid) - from gvsds.apps.msync.serializers import ServiceNodeSerializer + from gvsdsdk.msync.serializers import ServiceNodeSerializer result = _ok(ServiceNodeSerializer(node).data) logger.info(f'[SIO] ← msync:register sid={sid} code=200') return result @@ -149,7 +149,7 @@ def on_register(sid, data): def on_auth(sid, data): service_name = _d(data, 'ServiceName') logger.info(f'[SIO] → msync:auth sid={sid} service={service_name}') - from gvsds.apps.msync.auth import AuthenticateService + from gvsdsdk.msync.auth import AuthenticateService try: token = AuthenticateService( service_name=service_name, @@ -199,7 +199,7 @@ def on_auth(sid, data): @sio.on('msync:auth:refresh', namespace=NS) def on_auth_refresh(sid, data): - from gvsds.apps.msync.auth import RefreshServiceToken + from gvsdsdk.msync.auth import RefreshServiceToken try: token = RefreshServiceToken(_d(data, 'RefreshToken', '')) if token is None: @@ -251,7 +251,7 @@ def on_heartbeat(sid, data): return _err('未注册', 401) if not auth_name or auth_name != service_name: return _err('需要密钥认证', 401) - from gvsds.apps.msync.registry import ServiceRegistry, RegistryError + from gvsdsdk.msync.registry import ServiceRegistry, RegistryError try: ServiceRegistry.Heartbeat(service_name) except RegistryError as e: @@ -272,7 +272,7 @@ def on_deregister(sid, data): # 仅允许注销本服务（防止伪造 ServiceName 注销他人） if not auth_name or auth_name != name: return _err('需要密钥认证', 401) - from gvsds.apps.msync.registry import ServiceRegistry, RegistryError + from gvsdsdk.msync.registry import ServiceRegistry, RegistryError try: ServiceRegistry.Deregister(name) except RegistryError as e: @@ -285,7 +285,7 @@ def on_deregister(sid, data): def on_discover(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.registry import ServiceRegistry + from gvsdsdk.msync.registry import ServiceRegistry active_only = _d(data, 'ActiveOnly', True) if not isinstance(active_only, bool): active_only = True @@ -297,7 +297,7 @@ def on_discover(sid, data): def on_discover_one(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.registry import ServiceRegistry + from gvsdsdk.msync.registry import ServiceRegistry info = ServiceRegistry.Discover(_d(data, 'ServiceName', '')) if info is None: return _err('服务不存在', 404) @@ -308,7 +308,7 @@ def on_discover_one(sid, data): def on_ogm_register(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.ogm import OGMManager, OGMError + from gvsdsdk.msync.ogm import OGMManager, OGMError try: ogm = OGMManager.Register( object_type=_d(data, 'ObjectType'), @@ -329,7 +329,7 @@ def on_ogm_register(sid, data): def on_ogm_batch(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.ogm import OGMManager, OGMError + from gvsdsdk.msync.ogm import OGMManager, OGMError try: results = OGMManager.RegisterBatch( items=_d(data, 'Items', []), @@ -344,7 +344,7 @@ def on_ogm_batch(sid, data): def on_ogm_resolve(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.ogm import OGMManager + from gvsdsdk.msync.ogm import OGMManager info = OGMManager.Resolve(_d(data, 'ObjectGlobalID', '')) if info is None: return _err('映射不存在', 404) @@ -355,7 +355,7 @@ def on_ogm_resolve(sid, data): def on_ogm_query(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.ogm import OGMManager + from gvsdsdk.msync.ogm import OGMManager object_type = _d(data, 'ObjectType') service_name = _d(data, 'ServiceName') local_id = _d(data, 'ObjectLocalID') @@ -381,7 +381,7 @@ def on_ogm_query(sid, data): def on_ogm_update(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.ogm import OGMManager + from gvsdsdk.msync.ogm import OGMManager ogm = OGMManager.Update(_d(data, 'ObjectGlobalID', ''), metadata=_d(data, 'Metadata')) if ogm is None: return _err('映射不存在', 404) @@ -392,7 +392,7 @@ def on_ogm_update(sid, data): def on_ogm_deregister(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.ogm import OGMManager + from gvsdsdk.msync.ogm import OGMManager success = OGMManager.Deregister(_d(data, 'ObjectGlobalID', '')) if not success: return _err('映射不存在', 404) @@ -403,7 +403,7 @@ def on_ogm_deregister(sid, data): def on_tx_create(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager, TransactionError + from gvsdsdk.msync.transaction import TransactionManager, TransactionError try: dtx = TransactionManager.Create( name=_d(data, 'TransactionName'), @@ -424,7 +424,7 @@ def on_tx_create(sid, data): def on_tx_execute(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager, TransactionError + from gvsdsdk.msync.transaction import TransactionManager, TransactionError try: dtx = TransactionManager.Execute( _d(data, 'TransactionCode', ''), @@ -439,7 +439,7 @@ def on_tx_execute(sid, data): def on_tx_compensate(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager, TransactionError + from gvsdsdk.msync.transaction import TransactionManager, TransactionError try: dtx = TransactionManager.Compensate( _d(data, 'TransactionCode', ''), @@ -454,7 +454,7 @@ def on_tx_compensate(sid, data): def on_tx_cancel(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager, TransactionError + from gvsdsdk.msync.transaction import TransactionManager, TransactionError try: TransactionManager.Cancel(_d(data, 'TransactionCode', '')) except TransactionError as e: @@ -466,7 +466,7 @@ def on_tx_cancel(sid, data): def on_tx_detail(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager + from gvsdsdk.msync.transaction import TransactionManager detail = TransactionManager.GetDetail(_d(data, 'TransactionCode', '')) if detail is None: return _err('事务不存在', 404) @@ -477,7 +477,7 @@ def on_tx_detail(sid, data): def on_tx_retry(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager, TransactionError + from gvsdsdk.msync.transaction import TransactionManager, TransactionError try: step = TransactionManager.RetryStep( _d(data, 'TransactionCode', ''), @@ -498,7 +498,7 @@ def on_tx_retry(sid, data): def on_tx_retry_compensate(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager, TransactionError + from gvsdsdk.msync.transaction import TransactionManager, TransactionError try: step = TransactionManager.RetryCompensate( _d(data, 'TransactionCode', ''), @@ -520,7 +520,7 @@ def on_tx_retry_compensate(sid, data): def on_tx_timeout_check(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager + from gvsdsdk.msync.transaction import TransactionManager timed_out = TransactionManager.CheckTimeouts() return _ok({'TimedOut': timed_out, 'Count': len(timed_out)}) @@ -529,7 +529,7 @@ def on_tx_timeout_check(sid, data): def on_tx_query(sid, data): if not _RequireAuth(sid): return _err('需要密钥认证', 401) - from gvsds.apps.msync.transaction import TransactionManager + from gvsdsdk.msync.transaction import TransactionManager status = _d(data, 'Status') if status is None: return _err('请提供 Status 参数', 400) @@ -542,4 +542,4 @@ def on_tx_query(sid, data): def create_sio_app(django_app): - return socketio.WSGIApp(sio, django_app) \ No newline at end of file + return socketio.WSGIApp(sio, django_app) diff --git a/gvsdsdk/msync/transaction.py b/gvsdsdk/msync/transaction.py index 26a34e78..4c4da79f 100644 --- a/gvsdsdk/msync/transaction.py +++ b/gvsdsdk/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/gvsds/apps/msync/urls.py b/gvsdsdk/msync/urls.py similarity index 98% rename from gvsds/apps/msync/urls.py rename to gvsdsdk/msync/urls.py index 827a1877..5af8f365 100644 --- a/gvsds/apps/msync/urls.py +++ b/gvsdsdk/msync/urls.py @@ -1,5 +1,5 @@ from django.urls import path -from gvsds.apps.msync import views +from gvsdsdk.msync import views app_name = 'msync' diff --git a/gvsds/apps/msync/views.py b/gvsdsdk/msync/views.py similarity index 97% rename from gvsds/apps/msync/views.py rename to gvsdsdk/msync/views.py index 148bb666..214316c3 100644 --- a/gvsds/apps/msync/views.py +++ b/gvsdsdk/msync/views.py @@ -3,17 +3,17 @@ import traceback from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated -from gvsds.apps.msync.auth import ( +from gvsdsdk.msync.auth import ( AuthenticateService, RefreshServiceToken, ValidateAccessToken, ) -from gvsds.apps.msync.middleware import ServiceTokenAuthentication -from gvsds.apps.msync.registry import ServiceRegistry, RegistryError -from gvsds.apps.msync.ogm import OGMManager, OGMError -from gvsds.apps.msync.transaction import TransactionManager, TransactionError -from gvsds.apps.msync.local_tx import local_tx -from gvsds.apps.msync.serializers import ( +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, @@ -155,7 +155,7 @@ class RegistryListView(APIView): return Response(_ok(services)) def post(self, request): - # 注册端点：新服务尚未拥有令牌，无需认证 + # 注册端点：新服务尚未拥有令牌，无需认证，但必须携带密钥 # TODO: 考虑添加速率限制以防止滥用注册 ser = ServiceRegisterSerializer(data=request.data) if not ser.IsValid(): @@ -534,4 +534,4 @@ class LocalHandlersView(APIView): def get(self, request): handlers = local_tx.ListHandlers() - return Response(_ok(handlers)) \ No newline at end of file + return Response(_ok(handlers)) diff --git a/gvsdsdk/secrets.py b/gvsdsdk/secrets.py new file mode 100644 index 00000000..5dd4b48b --- /dev/null +++ b/gvsdsdk/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/gvsdsdk/serializer_base.py b/gvsdsdk/serializer_base.py new file mode 100644 index 00000000..84ef00ac --- /dev/null +++ b/gvsdsdk/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