如何使用pymongo终止MongoDB从节点上运行的查询

核心问题分析 #

MongoDB从节点（Secondary）的运行机制与主节点存在差异，你的脚本在从节点失效主要源于以下几点：

• 连接从节点时未明确指定读偏好，驱动可能默认路由到主节点，导致查询到的是主节点的操作而非从节点本地的。
• 从节点上$currentOp的默认行为无法精准捕获本地运行操作，需调整参数。
• 未过滤从节点同步主数据的系统操作，容易误判或无法终止目标用户查询。

修正后的脚本 #

import pymongo
import logging

# 连接从节点时明确指定读偏好，确保操作针对当前从节点
client = pymongo.MongoClient(
    "connection_string",
    readPreference="secondary",
    readPreferenceTags=[{}]
)
admin_db = client.get_database("admin")

pipeline = [
    {
        "$currentOp": {
            "allUsers": True,
            "idleSessions": False,  # 过滤空闲会话，聚焦活跃操作
            "localOps": True        # 强制只查询当前从节点的本地操作，关键参数
        }
    },
    {
        "$match": {
            "active": True,
            "secs_running": {"$gte": 60},  # 匹配运行时长超过1分钟的操作
            "ns": {"$in": ["test.fruits", "mydb.collection2"]},
            "op": {"$in": ["query", "command"]},
            "type": {"$ne": "replApply"}   # 排除从节点同步主数据的系统操作
        }
    },
]

ops = admin_db.aggregate(pipeline)

count = 0

for op in ops:
    try:
        admin_db.command({"killOp": 1, "op": op["opid"]})
        print(f"已终止操作: {op['opid']}")
        count += 1
    except pymongo.errors.OperationFailure as e:
        print(f"终止操作失败 {op['opid']}: {str(e)}")
        logging.error(f"终止操作失败 {op['opid']}: {str(e)}")

print(f"共终止 {count} 个长时操作")
logging.info(f"ops found: {count}")

关键修改说明 #

• 指定读偏好：连接时添加readPreference="secondary"，确保驱动直接与从节点交互，不会路由到主节点。
• 调整$currentOp参数：新增localOps: True，强制聚合查询仅返回当前从节点的本地操作；关闭idleSessions过滤无效会话。
• 过滤系统操作：添加"type": {"$ne": "replApply"}，排除从节点同步主数据的系统进程，只处理用户发起的查询/命令。
• 异常处理：增加try-except捕获终止操作时的权限问题、操作已完成等异常，提升脚本稳定性。
• 修正时长阈值：将原脚本的2秒改为60秒，匹配你“超过1分钟”的需求。

额外注意事项 #

• 确保执行脚本的MongoDB用户具备killOp和currentOp权限，在从节点上能正常执行相关操作。
• 若从节点为隐藏节点或延迟节点，需确保连接字符串可正常访问该节点，且读偏好设置匹配节点角色。

内容的提问来源于stack exchange，提问作者HIBUDDY