在 SQLite 实战项目里，业务数据和操作日志通常被写进同一个数据库文件，但排查问题时只靠日志文件很难快速看出某张表在哪个时间段出现大量慢插入或事务回滚。借助 New Relic 自定义事件，可以把 SQLite 的连接、写入、查询耗时和异常状态上报成结构化事件，用 NRQL 直接聚合分析。本文以一个 Python 服务为例，说明如何设计事件属性、选择上报方式，并把 SQLite 操作与 New Relic 监控打通。
一、为什么 SQLite 项目要接入自定义事件
SQLite 是典型的嵌入式数据库，没有独立服务进程，很多性能问题只能从应用侧观察。比如锁等待、执行计划变化、同步写入和事务提交时间，这些信息不会像 MySQL 慢日志一样独立存在。如果把这些指标设计成事件上报，可以按时间序列查看波动，不用再去翻分散的日志文件。
自定义事件比普通日志更适合做告警和聚合。日志文本需要解析，事件天然带属性，例如 table、operation、duration_ms、row_count、error_type。New Relic 的事件查询 NRQL 可以对属性做过滤和聚合，快速发现某个表在特定时间窗口的平均耗时是否上升，或者写入失败率是否异常。对于 SQLite 这种本地文件型数据库，应用内的观测数据往往是最直接的故障信号。
一个常见的实战场景是：用户行为表持续写入，原始数据量增长很快，写入高峰时可能出现锁等待。如果只记录一条简单日志，很难判断是表结构问题、事务过长还是磁盘 I/O 抖动。把每次写入的耗时和错误类型作为事件上报后，就能通过聚合视图快速定位。
下面是一个基础的 SQLite 表结构，后续的监控封装都会围绕它展开：
CREATE TABLE IF NOT EXISTS user_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
event_type TEXT NOT NULL,
payload TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE INDEX idx_user_events_user_id ON user_events(user_id);
CREATE INDEX idx_user_events_created_at ON user_events(created_at);
二、New Relic 自定义事件的上报路径与选型
New Relic 提供了多种自定义事件上报途径。最直接的是 Event API，通过 HTTPS 向 Insight Collector 发送 JSON 数据，适合没有官方 agent 的环境。它的优点是接入简单，不依赖进程内 SDK；缺点是需要自己处理批量、重试和网络异常。一般在跨语言脚本、Serverless 函数或一次性任务中使用。
第二种是 Telemetry SDK，官方推荐用于自定义遥测数据。Python、Node.js、Go 等语言都有对应 SDK，内部会做批量聚合、压缩和失败重试，适合新项目或独立监控服务。第三种是 New Relic APM agent 自带的 record_custom_event 方法。如果项目已经安装了 Python agent，并且通过 initialize 加载了配置文件，那么在一个业务事务中直接调用即可上报，不需要额外维护 HTTP 客户端。这也是本文采用的方式，因为代码量最少，和现有应用关联最自然。
无论选哪种方式，都需要注意事件类型的命名。事件类型只能包含字母、数字、冒号和下划线，且不能使用 New Relic 保留前缀。属性值建议只使用字符串、数值和布尔类型，列表或复杂对象需要序列化成 JSON 字符串。单个事件的总大小有上限，不要把大段 payload 原始内容塞进去，只保留摘要字段即可。
下面是在 Python agent 中封装一个事件上报函数的示例：
import newrelic.agent
newrelic.agent.initialize('newrelic.ini')
def record_sqlite_event(table, operation, duration_ms, row_count=0, error_type=None):
event_type = 'SqliteOperation'
params = {
'table': table,
'operation': operation,
'duration_ms': duration_ms,
'row_count': row_count,
}
if error_type:
params['error_type'] = error_type
newrelic.agent.record_custom_event(event_type, params)
三、SQLite 操作封装与自动上报实战
在实际项目中，不推荐每次执行 SQL 都创建新的 SQLite 连接，因为连接和解析的开销不小，而且多线程场景下还容易触发 check_same_thread 限制。更合适的做法是封装一个存储类，在初始化时创建连接并设置 WAL 模式，对外暴露业务方法。这样既能统一管理事务，也能在方法内部集中记录耗时和异常。
下面的封装演示了如何记录插入操作的成功和失败事件。成功事件包含表名、操作类型、耗时和影响行数；失败事件额外包含 error_type，方便后续统计各类 SQLite 异常出现的比例。使用 with self._conn 可以确保事务正确提交或回滚。
import sqlite3
import time
import newrelic.agent
class SqliteStore:
def __init__(self, db_path):
self.db_path = db_path
self._conn = sqlite3.connect(db_path)
self._conn.execute('PRAGMA journal_mode=WAL;')
self._conn.execute('PRAGMA synchronous=NORMAL;')
def insert_user_event(self, user_id, event_type, payload):
start = time.time()
try:
with self._conn:
self._conn.execute(
'INSERT INTO user_events(user_id, event_type, payload) VALUES(?,?,?)',
(user_id, event_type, payload)
)
duration_ms = (time.time() - start) * 1000
newrelic.agent.record_custom_event('SqliteOperation', {
'table': 'user_events',
'operation': 'insert',
'duration_ms': duration_ms,
'row_count': 1,
})
except sqlite3.Error as exc:
duration_ms = (time.time() - start) * 1000
newrelic.agent.record_custom_event('SqliteOperation', {
'table': 'user_events',
'operation': 'insert',
'duration_ms': duration_ms,
'row_count': 0,
'error_type': type(exc).__name__,
})
raise
这个封装的关键点有两个：一是用 with 语句管理事务，避免半提交状态；二是捕获 sqlite3.Error 后先上报再抛出，既不影响原有异常链路，又能留下可观测痕迹。对于查询操作，可以再封装一个 select 方法，按同样方式记录耗时和返回行数。这样后续无论是排查慢查询还是分析表写入趋势，都能直接基于事件数据展开。
此外，如果业务写入非常频繁，逐条上报并不划算。单行插入每秒上千次时，最好先在应用内存中聚合，再定期上报。聚合方案会在第五节详细说明。
四、NRQL 查询与告警实践
事件上报之后，在 New Relic 的查询界面中可以使用 NRQL 分析。最简单的查询是查看某个表最近一小时的平均耗时和最大耗时，并按时间序列展示。这样做可以直观看到写入耗时有没有随时间抬高，是否存在周期性的 I/O 或锁竞争。
下面是一个常用的 NRQL 查询，用于观察 user_events 表插入操作的趋势：
SELECT average(duration_ms) AS avg_ms, max(duration_ms) AS max_ms FROM SqliteOperation WHERE operation = 'insert' AND table = 'user_events' SINCE 1 day ago TIMESERIES 5 minutes
如果需要统计错误率，可以先分别查询总事件数和带 error_type 的事件数，再做除法；也可以直接使用 NRQL 的条件过滤。例如：
SELECT count(*) AS error_count FROM SqliteOperation WHERE error_type IS NOT NULL SINCE 10 minutes ago
告警条件可以基于类似查询创建。例如当最近 10 分钟错误事件数大于 5 时触发邮件或 PagerDuty 通知。如果关心慢操作，可以将条件设为 max(duration_ms) 超过某个阈值，或 average(duration_ms) 在 5 分钟窗口内持续高于 300 毫秒。NRQL 支持 COMPARE WITH 等语法，方便和过去同一时段做对比，减少误报。
需要注意的是，自定义事件默认保留期限有限，不适合做长期历史归档。如果需要对一个月甚至更久的数据做趋势分析，应在应用侧或数据管道中将聚合结果落库，而不是依赖事件保留策略。
五、高频场景下的聚合与避坑
高频 SQLite 操作如果逐条上报，事件配额会很快耗尽，同时也会增加应用 CPU 和网络开销。更合理的做法是本地聚合：按表名和操作类型分桶，累计请求次数、总耗时、最大耗时和错误次数，然后每 5 秒或 10 秒上报一个聚合事件。这样既保留了趋势，又大幅降低了事件量。
下面是一个简单的线程安全聚合器示例，可以在后台线程中定期调用 flush 方法：
import threading
import time
class EventAggregator:
def __init__(self, flush_interval=5):
self.flush_interval = flush_interval
self.buffer = {}
self.lock = threading.Lock()
self._stop = False
def add(self, table, operation, duration_ms, error_type=None):
key = (table, operation)
with self.lock:
item = self.buffer.setdefault(key, {
'count': 0,
'total_ms': 0.0,
'max_ms': 0.0,
'errors': 0,
})
item['count'] += 1
item['total_ms'] += duration_ms
if duration_ms > item['max_ms']:
item['max_ms'] = duration_ms
if error_type:
item['errors'] += 1
def flush(self):
with self.lock:
snapshot = self.buffer
self.buffer = {}
for (table, operation), item in snapshot.items():
newrelic.agent.record_custom_event('SqliteOperationAggregated', {
'table': table,
'operation': operation,
'count': item['count'],
'avg_ms': item['total_ms'] / item['count'],
'max_ms': item['max_ms'],
'error_count': item['errors'],
})
除了事件量，SQLite 自身也有几个容易踩的坑。写入模式建议启用 WAL，它能减少读写之间的锁竞争，但 WAL 文件会增长，需要在空闲时做 checkpoint。事务中不要做网络上报，因为 record_custom_event 虽然很快，但如果网络抖动仍可能拖慢事务提交。应把事件放入队列或聚合器，由后台线程统一 flush。最后，事件属性值保持类型一致，比如 duration_ms 如果不小心一处传了字符串，NRQL 的聚合结果就会出现类型错误，排查起来很麻烦。
从可观测性角度看，SQLite 轻量但不等于不需要监控。把关键操作转成结构化事件后，慢查询、锁等待、事务回滚和写入失败都不再依赖人工翻日志，而是可以通过仪表盘和告警主动暴露出来。配合合理的聚合策略，New Relic 自定义事件可以很好覆盖 SQLite 实战项目的监控需求。