refactor: 移除运行时数据库初始化与分区维护
- 删除了服务启动阶段的数据库初始化逻辑，包括创建数据库、表和分区的相关代码。 - 移除了定时分区维护任务，确保服务职责更清晰。 - 更新了数据库分区策略，明确分区由外部脚本管理，服务不再自动创建缺失分区。 - 修改了相关文档，确保数据库结构与分区维护的责任转移到 `SQL_Script/` 目录下的外部脚本。 - 更新了需求和场景，确保符合新的设计规范。
This commit is contained in:
@@ -18,7 +18,13 @@ bls-onoffline-backend
|
||||
- 复制 .env.example 为 .env 并按实际环境配置
|
||||
|
||||
数据库初始化
|
||||
- 启动时自动执行 scripts/init_db.sql 并预创建未来 30 天分区
|
||||
- 运行服务前请先通过根目录 SQL_Script 下脚本完成建库与分区维护
|
||||
- `../SQL_Script/create_database.sql`：建库（psql）
|
||||
- `../SQL_Script/create_schema_and_parent_table.sql`：建 schema 与主分区表
|
||||
- `../SQL_Script/create_partition_for_day.sql`：按日建分区模板
|
||||
- `../SQL_Script/generate_init_sql.js`：生成建库+建表 SQL
|
||||
- `../SQL_Script/generate_partition_sql.js`：生成单日分区 SQL
|
||||
- `../SQL_Script/generate_partition_range_sql.js`：生成批量分区 SQL
|
||||
|
||||
规范说明
|
||||
- 规格文件位于 spec/onoffline-spec.md
|
||||
|
||||
288
bls-onoffline-backend/dist/index.js
vendored
288
bls-onoffline-backend/dist/index.js
vendored
@@ -1,9 +1,6 @@
|
||||
import cron from "node-cron";
|
||||
import dotenv from "dotenv";
|
||||
import pg from "pg";
|
||||
import fs from "fs";
|
||||
import path from "path";
|
||||
import { fileURLToPath } from "url";
|
||||
import kafka from "kafka-node";
|
||||
import { randomUUID } from "crypto";
|
||||
import { z } from "zod";
|
||||
@@ -173,262 +170,6 @@ class DatabaseManager {
|
||||
}
|
||||
}
|
||||
const dbManager = new DatabaseManager(config.db);
|
||||
class PartitionManager {
|
||||
isCurrentOrFutureDate(date) {
|
||||
const normalizedDate = new Date(date);
|
||||
normalizedDate.setHours(0, 0, 0, 0);
|
||||
const today = /* @__PURE__ */ new Date();
|
||||
today.setHours(0, 0, 0, 0);
|
||||
return normalizedDate.getTime() >= today.getTime();
|
||||
}
|
||||
escapeSqlLiteral(value) {
|
||||
return String(value).replace(/'/g, "''");
|
||||
}
|
||||
buildForceHotTablespaceSql(schema, partitionName, hotTablespace = "ts_hot") {
|
||||
const schemaLiteral = this.escapeSqlLiteral(schema);
|
||||
const partitionLiteral = this.escapeSqlLiteral(partitionName);
|
||||
const hotLiteral = this.escapeSqlLiteral(hotTablespace);
|
||||
return `
|
||||
DO $$
|
||||
DECLARE
|
||||
v_schema text := '${schemaLiteral}';
|
||||
v_partition text := '${partitionLiteral}';
|
||||
v_hot text := '${hotLiteral}';
|
||||
v_part_oid oid;
|
||||
v_toast_oid oid;
|
||||
r record;
|
||||
BEGIN
|
||||
SELECT c.oid INTO v_part_oid
|
||||
FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace
|
||||
WHERE n.nspname=v_schema AND c.relname=v_partition AND c.relkind='r';
|
||||
|
||||
IF v_part_oid IS NULL THEN
|
||||
RAISE EXCEPTION 'partition %.% not found', v_schema, v_partition;
|
||||
END IF;
|
||||
|
||||
EXECUTE format('ALTER TABLE %I.%I SET TABLESPACE %I', v_schema, v_partition, v_hot);
|
||||
|
||||
FOR r IN
|
||||
SELECT idxn.nspname AS index_schema, i.relname AS index_name
|
||||
FROM pg_index x
|
||||
JOIN pg_class t ON t.oid=x.indrelid
|
||||
JOIN pg_namespace nt ON nt.oid=t.relnamespace
|
||||
JOIN pg_class i ON i.oid=x.indexrelid
|
||||
JOIN pg_namespace idxn ON idxn.oid=i.relnamespace
|
||||
LEFT JOIN pg_tablespace ts ON ts.oid=i.reltablespace
|
||||
WHERE nt.nspname=v_schema
|
||||
AND t.relname=v_partition
|
||||
AND COALESCE(ts.spcname,'pg_default')<>v_hot
|
||||
LOOP
|
||||
EXECUTE format('ALTER INDEX %I.%I SET TABLESPACE %I', r.index_schema, r.index_name, v_hot);
|
||||
END LOOP;
|
||||
|
||||
SELECT reltoastrelid INTO v_toast_oid FROM pg_class WHERE oid=v_part_oid;
|
||||
IF v_toast_oid IS NOT NULL AND v_toast_oid<>0 THEN
|
||||
EXECUTE format('ALTER TABLE %s SET TABLESPACE %I', v_toast_oid::regclass, v_hot);
|
||||
|
||||
FOR r IN
|
||||
SELECT idxn.nspname AS index_schema, i.relname AS index_name
|
||||
FROM pg_index x
|
||||
JOIN pg_class i ON i.oid=x.indexrelid
|
||||
JOIN pg_namespace idxn ON idxn.oid=i.relnamespace
|
||||
LEFT JOIN pg_tablespace ts ON ts.oid=i.reltablespace
|
||||
WHERE x.indrelid=v_toast_oid
|
||||
AND COALESCE(ts.spcname,'pg_default')<>v_hot
|
||||
LOOP
|
||||
EXECUTE format('ALTER INDEX %I.%I SET TABLESPACE %I', r.index_schema, r.index_name, v_hot);
|
||||
END LOOP;
|
||||
END IF;
|
||||
|
||||
EXECUTE format('ANALYZE %I.%I', v_schema, v_partition);
|
||||
END $$;
|
||||
`;
|
||||
}
|
||||
/**
|
||||
* Calculate the start and end timestamps (milliseconds) for a given date.
|
||||
* @param {Date} date - The date to calculate for.
|
||||
* @returns {Object} { startMs, endMs, partitionSuffix }
|
||||
*/
|
||||
getPartitionInfo(date) {
|
||||
const yyyy = date.getFullYear();
|
||||
const mm = String(date.getMonth() + 1).padStart(2, "0");
|
||||
const dd = String(date.getDate()).padStart(2, "0");
|
||||
const partitionSuffix = `${yyyy}${mm}${dd}`;
|
||||
const start = new Date(date);
|
||||
start.setHours(0, 0, 0, 0);
|
||||
const startMs = start.getTime();
|
||||
const end = new Date(date);
|
||||
end.setDate(end.getDate() + 1);
|
||||
end.setHours(0, 0, 0, 0);
|
||||
const endMs = end.getTime();
|
||||
return { startMs, endMs, partitionSuffix };
|
||||
}
|
||||
/**
|
||||
* Ensure partitions exist for the past M days and next N days.
|
||||
* @param {number} daysAhead - Number of days to pre-create.
|
||||
* @param {number} daysBack - Number of days to look back.
|
||||
*/
|
||||
async ensurePartitions(daysAhead = 30, daysBack = 15) {
|
||||
const client = await dbManager.pool.connect();
|
||||
try {
|
||||
logger.info(`Starting partition check for the past ${daysBack} days and next ${daysAhead} days...`);
|
||||
console.log(`Starting partition check for the past ${daysBack} days and next ${daysAhead} days...`);
|
||||
const now = /* @__PURE__ */ new Date();
|
||||
for (let i = -daysBack; i < daysAhead; i++) {
|
||||
const targetDate = new Date(now);
|
||||
targetDate.setDate(now.getDate() + i);
|
||||
const { startMs, endMs, partitionSuffix } = this.getPartitionInfo(targetDate);
|
||||
const schema = config.db.schema;
|
||||
const table = config.db.table;
|
||||
const partitionName = `${schema}.${table}_${partitionSuffix}`;
|
||||
const checkSql = `
|
||||
SELECT to_regclass($1) as exists;
|
||||
`;
|
||||
const checkRes = await client.query(checkSql, [partitionName]);
|
||||
if (!checkRes.rows[0].exists) {
|
||||
logger.info(`Creating partition ${partitionName} for range [${startMs}, ${endMs})`);
|
||||
console.log(`Creating partition ${partitionName} for range [${startMs}, ${endMs})`);
|
||||
const shouldUseHotTablespace = this.isCurrentOrFutureDate(targetDate);
|
||||
const tablespaceClause = shouldUseHotTablespace ? " TABLESPACE ts_hot" : "";
|
||||
const partitionTableName = `${table}_${partitionSuffix}`;
|
||||
const createSql = `
|
||||
CREATE TABLE IF NOT EXISTS ${partitionName}
|
||||
PARTITION OF ${schema}.${table}
|
||||
FOR VALUES FROM (${startMs}) TO (${endMs})${tablespaceClause};
|
||||
`;
|
||||
await client.query(createSql);
|
||||
if (shouldUseHotTablespace) {
|
||||
await client.query(this.buildForceHotTablespaceSql(schema, partitionTableName));
|
||||
}
|
||||
}
|
||||
}
|
||||
logger.info("Partition check completed.");
|
||||
} catch (err) {
|
||||
logger.error("Error ensuring partitions:", err);
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
async ensurePartitionsForTimestamps(tsMsList) {
|
||||
if (!Array.isArray(tsMsList) || tsMsList.length === 0) return;
|
||||
const uniqueSuffixes = /* @__PURE__ */ new Set();
|
||||
for (const ts of tsMsList) {
|
||||
const numericTs = typeof ts === "string" ? Number(ts) : ts;
|
||||
if (!Number.isFinite(numericTs)) continue;
|
||||
const date = new Date(numericTs);
|
||||
if (Number.isNaN(date.getTime())) continue;
|
||||
const { partitionSuffix } = this.getPartitionInfo(date);
|
||||
uniqueSuffixes.add(partitionSuffix);
|
||||
if (uniqueSuffixes.size >= 400) break;
|
||||
}
|
||||
if (uniqueSuffixes.size === 0) return;
|
||||
const client = await dbManager.pool.connect();
|
||||
try {
|
||||
const schema = config.db.schema;
|
||||
const table = config.db.table;
|
||||
for (const partitionSuffix of uniqueSuffixes) {
|
||||
const yyyy = Number(partitionSuffix.slice(0, 4));
|
||||
const mm = Number(partitionSuffix.slice(4, 6));
|
||||
const dd = Number(partitionSuffix.slice(6, 8));
|
||||
if (!Number.isFinite(yyyy) || !Number.isFinite(mm) || !Number.isFinite(dd)) continue;
|
||||
const targetDate = new Date(yyyy, mm - 1, dd);
|
||||
if (Number.isNaN(targetDate.getTime())) continue;
|
||||
const { startMs, endMs } = this.getPartitionInfo(targetDate);
|
||||
const partitionName = `${schema}.${table}_${partitionSuffix}`;
|
||||
const checkRes = await client.query(`SELECT to_regclass($1) as exists;`, [partitionName]);
|
||||
if (!checkRes.rows[0].exists) {
|
||||
logger.info(`Creating partition ${partitionName} for range [${startMs}, ${endMs})`);
|
||||
const shouldUseHotTablespace = this.isCurrentOrFutureDate(targetDate);
|
||||
const tablespaceClause = shouldUseHotTablespace ? " TABLESPACE ts_hot" : "";
|
||||
const partitionTableName = `${table}_${partitionSuffix}`;
|
||||
await client.query(`
|
||||
CREATE TABLE IF NOT EXISTS ${partitionName}
|
||||
PARTITION OF ${schema}.${table}
|
||||
FOR VALUES FROM (${startMs}) TO (${endMs})${tablespaceClause};
|
||||
`);
|
||||
if (shouldUseHotTablespace) {
|
||||
await client.query(this.buildForceHotTablespaceSql(schema, partitionTableName));
|
||||
}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
}
|
||||
const partitionManager = new PartitionManager();
|
||||
const __filename$1 = fileURLToPath(import.meta.url);
|
||||
const __dirname$1 = path.dirname(__filename$1);
|
||||
class DatabaseInitializer {
|
||||
async initialize() {
|
||||
logger.info("Starting database initialization check...");
|
||||
await this.ensureDatabaseExists();
|
||||
await this.ensureSchemaAndTable();
|
||||
await partitionManager.ensurePartitions(30);
|
||||
console.log("Database initialization completed successfully.");
|
||||
logger.info("Database initialization completed successfully.");
|
||||
}
|
||||
async ensureDatabaseExists() {
|
||||
const { host, port, user, password, database, ssl } = config.db;
|
||||
console.log(`Checking if database '${database}' exists at ${host}:${port}...`);
|
||||
const client = new pg.Client({
|
||||
host,
|
||||
port,
|
||||
user,
|
||||
password,
|
||||
database: "postgres",
|
||||
ssl: ssl ? { rejectUnauthorized: false } : false
|
||||
});
|
||||
try {
|
||||
await client.connect();
|
||||
const checkRes = await client.query(
|
||||
`SELECT 1 FROM pg_database WHERE datname = $1`,
|
||||
[database]
|
||||
);
|
||||
if (checkRes.rowCount === 0) {
|
||||
logger.info(`Database '${database}' does not exist. Creating...`);
|
||||
await client.query(`CREATE DATABASE "${database}"`);
|
||||
console.log(`Database '${database}' created.`);
|
||||
logger.info(`Database '${database}' created.`);
|
||||
} else {
|
||||
console.log(`Database '${database}' already exists.`);
|
||||
logger.info(`Database '${database}' already exists.`);
|
||||
}
|
||||
} catch (err) {
|
||||
logger.error("Error ensuring database exists:", err);
|
||||
throw err;
|
||||
} finally {
|
||||
await client.end();
|
||||
}
|
||||
}
|
||||
async ensureSchemaAndTable() {
|
||||
const client = await dbManager.pool.connect();
|
||||
try {
|
||||
const sqlPathCandidates = [
|
||||
path.resolve(process.cwd(), "scripts/init_db.sql"),
|
||||
path.resolve(__dirname$1, "../scripts/init_db.sql"),
|
||||
path.resolve(__dirname$1, "../../scripts/init_db.sql")
|
||||
];
|
||||
const sqlPath = sqlPathCandidates.find((candidate) => fs.existsSync(candidate));
|
||||
if (!sqlPath) {
|
||||
throw new Error(`init_db.sql not found. Candidates: ${sqlPathCandidates.join(" | ")}`);
|
||||
}
|
||||
const sql = fs.readFileSync(sqlPath, "utf8");
|
||||
console.log(`Executing init_db.sql from ${sqlPath}...`);
|
||||
logger.info("Executing init_db.sql...");
|
||||
await client.query(sql);
|
||||
console.log("Schema and parent table initialized.");
|
||||
logger.info("Schema and parent table initialized.");
|
||||
} catch (err) {
|
||||
logger.error("Error initializing schema and table:", err);
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
}
|
||||
const dbInitializer = new DatabaseInitializer();
|
||||
class OffsetTracker {
|
||||
constructor() {
|
||||
this.partitions = /* @__PURE__ */ new Map();
|
||||
@@ -874,16 +615,7 @@ const bootstrap = async () => {
|
||||
port: config.redis.port
|
||||
}
|
||||
});
|
||||
await dbInitializer.initialize();
|
||||
const metricCollector = new MetricCollector();
|
||||
cron.schedule("0 0 * * *", async () => {
|
||||
logger.info("Running scheduled partition maintenance...");
|
||||
try {
|
||||
await partitionManager.ensurePartitions(30);
|
||||
} catch (err) {
|
||||
logger.error("Scheduled partition maintenance failed", err);
|
||||
}
|
||||
});
|
||||
const redisClient = await createRedisClient(config.redis);
|
||||
const redisIntegration = new RedisIntegration(
|
||||
redisClient,
|
||||
@@ -991,10 +723,8 @@ const bootstrap = async () => {
|
||||
const lower = message.toLowerCase();
|
||||
return lower.includes("connection timeout") || lower.includes("connection terminated") || lower.includes("connection refused") || lower.includes("terminating connection") || lower.includes("econnrefused") || lower.includes("econnreset") || lower.includes("etimedout") || lower.includes("could not connect") || lower.includes("the database system is starting up") || lower.includes("no pg_hba.conf entry");
|
||||
};
|
||||
const isMissingPartitionError = (err) => err?.code === "23514" || typeof err?.message === "string" && err.message.includes("no partition of relation");
|
||||
const insertRowsWithRetry = async (rows) => {
|
||||
const startedAt = Date.now();
|
||||
let attemptedPartitionFix = false;
|
||||
while (true) {
|
||||
try {
|
||||
await dbManager.insertRows({ schema: config.db.schema, table: config.db.table, rows });
|
||||
@@ -1011,24 +741,6 @@ const bootstrap = async () => {
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (isMissingPartitionError(err) && !attemptedPartitionFix) {
|
||||
attemptedPartitionFix = true;
|
||||
try {
|
||||
await partitionManager.ensurePartitionsForTimestamps(rows.map((r) => r.ts_ms));
|
||||
} catch (partitionErr) {
|
||||
if (isDbConnectionError(partitionErr)) {
|
||||
logger.error("Database offline during partition ensure. Retrying in 5s...", { error: partitionErr.message });
|
||||
await new Promise((r) => setTimeout(r, 5e3));
|
||||
while (!await dbManager.checkConnection()) {
|
||||
logger.warn("Database still offline. Waiting 5s...");
|
||||
await new Promise((r) => setTimeout(r, 5e3));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
throw partitionErr;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
# Change: remove runtime db provisioning
|
||||
|
||||
## Why
|
||||
当前服务在运行时承担了建库、建表和分区维护职责，导致服务职责边界不清晰，也会引入启动阶段 DDL 风险。现已将该能力剥离到根目录 `SQL_Script/`，需要通过 OpenSpec 正式记录为规范变更。
|
||||
|
||||
## What Changes
|
||||
- 移除服务启动阶段的数据库初始化与定时分区维护要求。
|
||||
- 移除服务在写入失败时自动创建缺失分区的要求。
|
||||
- 明确数据库结构与分区维护由外部脚本（`SQL_Script/`）负责。
|
||||
- 保留服务的核心职责：Kafka 消费、解析、写库、重试与监控。
|
||||
|
||||
## Impact
|
||||
- Affected specs: `openspec/specs/onoffline/spec.md`
|
||||
- Affected code: `src/index.js`, `src/config/config.js`, `src/db/initializer.js`, `src/db/partitionManager.js`, `scripts/init_db.sql`, `scripts/verify_partitions.js`, `../SQL_Script/*`
|
||||
@@ -0,0 +1,32 @@
|
||||
## MODIFIED Requirements
|
||||
|
||||
### Requirement: 数据库分区策略
|
||||
系统 SHALL 使用 Range Partitioning 按天分区；运行服务本身 SHALL NOT 执行建库、建表、分区创建或定时分区维护。
|
||||
|
||||
#### Scenario: 服务启动不执行 DDL
|
||||
- **GIVEN** 服务进程启动
|
||||
- **WHEN** 进入 bootstrap 过程
|
||||
- **THEN** 仅初始化消费、处理、监控相关能力，不执行数据库创建、表结构初始化与分区创建
|
||||
|
||||
#### Scenario: 分区由外部脚本维护
|
||||
- **GIVEN** 需要创建数据库对象或新增未来分区
|
||||
- **WHEN** 执行外部 SQL/JS 工具
|
||||
- **THEN** 通过根目录 `SQL_Script/` 完成建库和分区维护，而不是由服务运行时自动执行
|
||||
|
||||
### Requirement: 批量消费与写入
|
||||
系统 SHALL 对 Kafka 消息进行缓冲，并按批次写入数据库，以提高吞吐量；当写入失败时，系统 SHALL 执行连接恢复重试与降级策略，但不在运行时创建数据库分区。
|
||||
|
||||
#### Scenario: 批量写入
|
||||
- **GIVEN** 短时间内收到多条消息 (e.g., 500条)
|
||||
- **WHEN** 缓冲区满或超时 (e.g., 200ms)
|
||||
- **THEN** 执行一次批量数据库插入操作
|
||||
|
||||
#### Scenario: 写入失败降级
|
||||
- **GIVEN** 批量写入因数据错误失败 (非连接错误)
|
||||
- **WHEN** 捕获异常
|
||||
- **THEN** 自动降级为逐条写入，以隔离错误数据并确保有效数据入库
|
||||
|
||||
#### Scenario: 分区缺失错误处理
|
||||
- **GIVEN** 写入时数据库返回分区缺失错误
|
||||
- **WHEN** 服务处理该错误
|
||||
- **THEN** 服务记录错误并按既有错误处理机制处理，不在运行时执行分区创建
|
||||
@@ -0,0 +1,12 @@
|
||||
## 1. Implementation
|
||||
- [x] 1.1 Remove runtime DB initialization from bootstrap flow (`src/index.js`).
|
||||
- [x] 1.2 Remove scheduled partition maintenance job from runtime service.
|
||||
- [x] 1.3 Remove runtime missing-partition auto-fix behavior.
|
||||
- [x] 1.4 Remove legacy DB provisioning modules and scripts from service project.
|
||||
- [x] 1.5 Add external SQL/JS provisioning scripts under root `SQL_Script/` for DB/schema/partition management.
|
||||
- [x] 1.6 Update project docs to point DB provisioning to `SQL_Script/`.
|
||||
|
||||
## 2. Validation
|
||||
- [x] 2.1 Run `npm run lint` in `bls-onoffline-backend`.
|
||||
- [x] 2.2 Run `npm run build` in `bls-onoffline-backend`.
|
||||
- [x] 2.3 Run `openspec validate remove-runtime-db-provisioning --strict`.
|
||||
@@ -28,12 +28,12 @@
|
||||
- **THEN** 数据库存储值为对应的空字符串
|
||||
|
||||
### Requirement: 数据库分区策略
|
||||
系统 SHALL 使用 Range Partitioning 按天分区，并自动维护未来 30 天的分区表。
|
||||
系统 SHALL 使用 Range Partitioning 按天分区，并自动维护未来 30 天的分区表，子表依赖 PostgreSQL 原生机制继承主表索引。
|
||||
|
||||
#### Scenario: 分区预创建
|
||||
- **GIVEN** 系统启动或每日凌晨
|
||||
- **WHEN** 运行分区维护任务
|
||||
- **THEN** 确保数据库中存在未来 30 天的分区表
|
||||
- **THEN** 确保数据库中存在未来 30 天的分区表，无需对子表显式创建单列表索引
|
||||
|
||||
### Requirement: 消费可靠性 (At-Least-Once)
|
||||
系统 SHALL 仅在数据成功写入数据库后，才向 Kafka 提交消费位点。
|
||||
@@ -84,7 +84,7 @@
|
||||
- **THEN** 自动乘以 1000 转换为毫秒
|
||||
|
||||
### Requirement: 批量消费与写入
|
||||
系统 SHALL 对 Kafka 消息进行缓冲，并按批次写入数据库，以提高吞吐量。
|
||||
系统 SHALL 对 Kafka 消息进行缓冲，并按批次写入数据库，以提高吞吐量；当写入失败时，系统 SHALL 执行连接恢复重试与降级策略，但不在运行时创建数据库分区。
|
||||
|
||||
#### Scenario: 批量写入
|
||||
- **GIVEN** 短时间内收到多条消息 (e.g., 500条)
|
||||
@@ -96,3 +96,8 @@
|
||||
- **WHEN** 捕获异常
|
||||
- **THEN** 自动降级为逐条写入，以隔离错误数据并确保有效数据入库
|
||||
|
||||
#### Scenario: 分区缺失错误处理
|
||||
- **GIVEN** 写入时数据库返回分区缺失错误
|
||||
- **WHEN** 服务处理该错误
|
||||
- **THEN** 服务记录错误并按既有错误处理机制处理，不在运行时执行分区创建
|
||||
|
||||
|
||||
@@ -1,23 +0,0 @@
|
||||
CREATE SCHEMA IF NOT EXISTS onoffline;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS onoffline.onoffline_record (
|
||||
guid VARCHAR(32) NOT NULL,
|
||||
ts_ms BIGINT NOT NULL,
|
||||
write_ts_ms BIGINT NOT NULL,
|
||||
hotel_id SMALLINT NOT NULL,
|
||||
mac VARCHAR(21) NOT NULL,
|
||||
device_id VARCHAR(64) NOT NULL,
|
||||
room_id VARCHAR(64) NOT NULL,
|
||||
ip VARCHAR(21),
|
||||
current_status VARCHAR(255),
|
||||
launcher_version VARCHAR(255),
|
||||
reboot_reason VARCHAR(255),
|
||||
PRIMARY KEY (ts_ms, mac, device_id, room_id)
|
||||
) PARTITION BY RANGE (ts_ms);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_onoffline_ts_ms ON onoffline.onoffline_record (ts_ms);
|
||||
CREATE INDEX IF NOT EXISTS idx_onoffline_hotel_id ON onoffline.onoffline_record (hotel_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_onoffline_mac ON onoffline.onoffline_record (mac);
|
||||
CREATE INDEX IF NOT EXISTS idx_onoffline_device_id ON onoffline.onoffline_record (device_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_onoffline_room_id ON onoffline.onoffline_record (room_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_onoffline_current_status ON onoffline.onoffline_record (current_status);
|
||||
@@ -1,61 +0,0 @@
|
||||
import pg from 'pg';
|
||||
import { config } from '../src/config/config.js';
|
||||
|
||||
const { Pool } = pg;
|
||||
|
||||
const verify = async () => {
|
||||
const pool = new Pool({
|
||||
host: config.db.host,
|
||||
port: config.db.port,
|
||||
user: config.db.user,
|
||||
password: config.db.password,
|
||||
database: config.db.database,
|
||||
ssl: config.db.ssl,
|
||||
});
|
||||
|
||||
try {
|
||||
console.log('Verifying partitions for table onoffline_record...');
|
||||
const client = await pool.connect();
|
||||
|
||||
// Check parent table
|
||||
const parentRes = await client.query(`
|
||||
SELECT to_regclass('onoffline.onoffline_record') as oid;
|
||||
`);
|
||||
if (!parentRes.rows[0].oid) {
|
||||
console.error('Parent table onoffline.onoffline_record DOES NOT EXIST.');
|
||||
return;
|
||||
}
|
||||
console.log('Parent table onoffline.onoffline_record exists.');
|
||||
|
||||
// Check partitions
|
||||
const res = await client.query(`
|
||||
SELECT
|
||||
child.relname AS partition_name
|
||||
FROM pg_inherits
|
||||
JOIN pg_class parent ON pg_inherits.inhparent = parent.oid
|
||||
JOIN pg_class child ON pg_inherits.inhrelid = child.oid
|
||||
JOIN pg_namespace ns ON parent.relnamespace = ns.oid
|
||||
WHERE parent.relname = 'onoffline_record' AND ns.nspname = 'onoffline'
|
||||
ORDER BY child.relname;
|
||||
`);
|
||||
|
||||
console.log(`Found ${res.rowCount} partitions.`);
|
||||
res.rows.forEach(row => {
|
||||
console.log(`- ${row.partition_name}`);
|
||||
});
|
||||
|
||||
if (res.rowCount >= 30) {
|
||||
console.log('SUCCESS: At least 30 partitions exist.');
|
||||
} else {
|
||||
console.warn(`WARNING: Expected 30+ partitions, found ${res.rowCount}.`);
|
||||
}
|
||||
|
||||
client.release();
|
||||
} catch (err) {
|
||||
console.error('Verification failed:', err);
|
||||
} finally {
|
||||
await pool.end();
|
||||
}
|
||||
};
|
||||
|
||||
verify();
|
||||
@@ -1,100 +0,0 @@
|
||||
import pg from 'pg';
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { fileURLToPath } from 'url';
|
||||
import { logger } from '../utils/logger.js';
|
||||
import partitionManager from './partitionManager.js';
|
||||
import dbManager from './databaseManager.js';
|
||||
import { config } from '../config/config.js';
|
||||
|
||||
const __filename = fileURLToPath(import.meta.url);
|
||||
const __dirname = path.dirname(__filename);
|
||||
|
||||
class DatabaseInitializer {
|
||||
async initialize() {
|
||||
logger.info('Starting database initialization check...');
|
||||
|
||||
// 1. Check if database exists, create if not
|
||||
await this.ensureDatabaseExists();
|
||||
|
||||
// 2. Initialize Schema and Parent Table (if not exists)
|
||||
// Note: We need to use dbManager because it connects to the target database
|
||||
await this.ensureSchemaAndTable();
|
||||
|
||||
// 3. Ensure Partitions for the next month
|
||||
await partitionManager.ensurePartitions(30);
|
||||
|
||||
console.log('Database initialization completed successfully.');
|
||||
logger.info('Database initialization completed successfully.');
|
||||
}
|
||||
|
||||
async ensureDatabaseExists() {
|
||||
const { host, port, user, password, database, ssl } = config.db;
|
||||
console.log(`Checking if database '${database}' exists at ${host}:${port}...`);
|
||||
|
||||
// Connect to 'postgres' database to check/create target database
|
||||
const client = new pg.Client({
|
||||
host,
|
||||
port,
|
||||
user,
|
||||
password,
|
||||
database: 'postgres',
|
||||
ssl: ssl ? { rejectUnauthorized: false } : false
|
||||
});
|
||||
|
||||
try {
|
||||
await client.connect();
|
||||
|
||||
const checkRes = await client.query(
|
||||
`SELECT 1 FROM pg_database WHERE datname = $1`,
|
||||
[database]
|
||||
);
|
||||
|
||||
if (checkRes.rowCount === 0) {
|
||||
logger.info(`Database '${database}' does not exist. Creating...`);
|
||||
// CREATE DATABASE cannot run inside a transaction block
|
||||
await client.query(`CREATE DATABASE "${database}"`);
|
||||
console.log(`Database '${database}' created.`);
|
||||
logger.info(`Database '${database}' created.`);
|
||||
} else {
|
||||
console.log(`Database '${database}' already exists.`);
|
||||
logger.info(`Database '${database}' already exists.`);
|
||||
}
|
||||
} catch (err) {
|
||||
logger.error('Error ensuring database exists:', err);
|
||||
throw err;
|
||||
} finally {
|
||||
await client.end();
|
||||
}
|
||||
}
|
||||
|
||||
async ensureSchemaAndTable() {
|
||||
// dbManager connects to the target database
|
||||
const client = await dbManager.pool.connect();
|
||||
try {
|
||||
const sqlPathCandidates = [
|
||||
path.resolve(process.cwd(), 'scripts/init_db.sql'),
|
||||
path.resolve(__dirname, '../scripts/init_db.sql'),
|
||||
path.resolve(__dirname, '../../scripts/init_db.sql')
|
||||
];
|
||||
const sqlPath = sqlPathCandidates.find((candidate) => fs.existsSync(candidate));
|
||||
if (!sqlPath) {
|
||||
throw new Error(`init_db.sql not found. Candidates: ${sqlPathCandidates.join(' | ')}`);
|
||||
}
|
||||
const sql = fs.readFileSync(sqlPath, 'utf8');
|
||||
|
||||
console.log(`Executing init_db.sql from ${sqlPath}...`);
|
||||
logger.info('Executing init_db.sql...');
|
||||
await client.query(sql);
|
||||
console.log('Schema and parent table initialized.');
|
||||
logger.info('Schema and parent table initialized.');
|
||||
} catch (err) {
|
||||
logger.error('Error initializing schema and table:', err);
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export default new DatabaseInitializer();
|
||||
@@ -1,136 +0,0 @@
|
||||
import { logger } from '../utils/logger.js';
|
||||
import { config } from '../config/config.js';
|
||||
import dbManager from './databaseManager.js';
|
||||
|
||||
class PartitionManager {
|
||||
isCurrentOrFutureDate(date) {
|
||||
const normalizedDate = new Date(date);
|
||||
normalizedDate.setHours(0, 0, 0, 0);
|
||||
|
||||
const today = new Date();
|
||||
today.setHours(0, 0, 0, 0);
|
||||
|
||||
return normalizedDate.getTime() >= today.getTime();
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate the start and end timestamps (milliseconds) for a given date.
|
||||
* @param {Date} date - The date to calculate for.
|
||||
* @returns {Object} { startMs, endMs, partitionSuffix }
|
||||
*/
|
||||
getPartitionInfo(date) {
|
||||
const yyyy = date.getFullYear();
|
||||
const mm = String(date.getMonth() + 1).padStart(2, '0');
|
||||
const dd = String(date.getDate()).padStart(2, '0');
|
||||
const partitionSuffix = `${yyyy}${mm}${dd}`;
|
||||
|
||||
const start = new Date(date);
|
||||
start.setHours(0, 0, 0, 0);
|
||||
const startMs = start.getTime();
|
||||
|
||||
const end = new Date(date);
|
||||
end.setDate(end.getDate() + 1);
|
||||
end.setHours(0, 0, 0, 0);
|
||||
const endMs = end.getTime();
|
||||
|
||||
return { startMs, endMs, partitionSuffix };
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure partitions exist for the past M days and next N days.
|
||||
* @param {number} daysAhead - Number of days to pre-create.
|
||||
* @param {number} daysBack - Number of days to look back.
|
||||
*/
|
||||
async ensurePartitions(daysAhead = 30, daysBack = 15) {
|
||||
const client = await dbManager.pool.connect();
|
||||
try {
|
||||
logger.info(`Starting partition check for the past ${daysBack} days and next ${daysAhead} days...`);
|
||||
console.log(`Starting partition check for the past ${daysBack} days and next ${daysAhead} days...`);
|
||||
const now = new Date();
|
||||
|
||||
for (let i = -daysBack; i < daysAhead; i++) {
|
||||
const targetDate = new Date(now);
|
||||
targetDate.setDate(now.getDate() + i);
|
||||
|
||||
const { startMs, endMs, partitionSuffix } = this.getPartitionInfo(targetDate);
|
||||
const schema = config.db.schema;
|
||||
const table = config.db.table;
|
||||
const partitionName = `${schema}.${table}_${partitionSuffix}`;
|
||||
|
||||
// Check if partition exists
|
||||
const checkSql = `
|
||||
SELECT to_regclass($1) as exists;
|
||||
`;
|
||||
const checkRes = await client.query(checkSql, [partitionName]);
|
||||
|
||||
if (!checkRes.rows[0].exists) {
|
||||
logger.info(`Creating partition ${partitionName} for range [${startMs}, ${endMs})`);
|
||||
console.log(`Creating partition ${partitionName} for range [${startMs}, ${endMs})`);
|
||||
const tablespaceClause = this.isCurrentOrFutureDate(targetDate) ? ' TABLESPACE ts_hot' : '';
|
||||
const createSql = `
|
||||
CREATE TABLE IF NOT EXISTS ${partitionName}
|
||||
PARTITION OF ${schema}.${table}
|
||||
FOR VALUES FROM (${startMs}) TO (${endMs})${tablespaceClause};
|
||||
`;
|
||||
await client.query(createSql);
|
||||
}
|
||||
}
|
||||
logger.info('Partition check completed.');
|
||||
} catch (err) {
|
||||
logger.error('Error ensuring partitions:', err);
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
|
||||
async ensurePartitionsForTimestamps(tsMsList) {
|
||||
if (!Array.isArray(tsMsList) || tsMsList.length === 0) return;
|
||||
|
||||
const uniqueSuffixes = new Set();
|
||||
for (const ts of tsMsList) {
|
||||
const numericTs = typeof ts === 'string' ? Number(ts) : ts;
|
||||
if (!Number.isFinite(numericTs)) continue;
|
||||
const date = new Date(numericTs);
|
||||
if (Number.isNaN(date.getTime())) continue;
|
||||
const { partitionSuffix } = this.getPartitionInfo(date);
|
||||
uniqueSuffixes.add(partitionSuffix);
|
||||
if (uniqueSuffixes.size >= 400) break;
|
||||
}
|
||||
|
||||
if (uniqueSuffixes.size === 0) return;
|
||||
|
||||
const client = await dbManager.pool.connect();
|
||||
try {
|
||||
const schema = config.db.schema;
|
||||
const table = config.db.table;
|
||||
|
||||
for (const partitionSuffix of uniqueSuffixes) {
|
||||
const yyyy = Number(partitionSuffix.slice(0, 4));
|
||||
const mm = Number(partitionSuffix.slice(4, 6));
|
||||
const dd = Number(partitionSuffix.slice(6, 8));
|
||||
if (!Number.isFinite(yyyy) || !Number.isFinite(mm) || !Number.isFinite(dd)) continue;
|
||||
const targetDate = new Date(yyyy, mm - 1, dd);
|
||||
if (Number.isNaN(targetDate.getTime())) continue;
|
||||
|
||||
const { startMs, endMs } = this.getPartitionInfo(targetDate);
|
||||
const partitionName = `${schema}.${table}_${partitionSuffix}`;
|
||||
|
||||
const checkRes = await client.query(`SELECT to_regclass($1) as exists;`, [partitionName]);
|
||||
if (!checkRes.rows[0].exists) {
|
||||
logger.info(`Creating partition ${partitionName} for range [${startMs}, ${endMs})`);
|
||||
const tablespaceClause = this.isCurrentOrFutureDate(targetDate) ? ' TABLESPACE ts_hot' : '';
|
||||
await client.query(`
|
||||
CREATE TABLE IF NOT EXISTS ${partitionName}
|
||||
PARTITION OF ${schema}.${table}
|
||||
FOR VALUES FROM (${startMs}) TO (${endMs})${tablespaceClause};
|
||||
`);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export default new PartitionManager();
|
||||
@@ -1,8 +1,6 @@
|
||||
import cron from 'node-cron';
|
||||
import { config } from './config/config.js';
|
||||
import dbManager from './db/databaseManager.js';
|
||||
import dbInitializer from './db/initializer.js';
|
||||
import partitionManager from './db/partitionManager.js';
|
||||
import { createKafkaConsumers } from './kafka/consumer.js';
|
||||
import { parseMessageToRows } from './processor/index.js';
|
||||
import { createRedisClient } from './redis/redisClient.js';
|
||||
@@ -33,38 +31,12 @@ const bootstrap = async () => {
|
||||
}
|
||||
});
|
||||
|
||||
// 0. Initialize Database (Create DB, Schema, Table, Partitions)
|
||||
await dbInitializer.initialize();
|
||||
|
||||
// Metric Collector
|
||||
const metricCollector = new MetricCollector();
|
||||
|
||||
// 1. Setup Partition Maintenance Cron Job (Every day at 00:00)
|
||||
cron.schedule('0 0 * * *', async () => {
|
||||
logger.info('Running scheduled partition maintenance...');
|
||||
try {
|
||||
await partitionManager.ensurePartitions(30);
|
||||
} catch (err) {
|
||||
logger.error('Scheduled partition maintenance failed', err);
|
||||
}
|
||||
});
|
||||
|
||||
// 1.1 Setup Metric Reporting Cron Job (Every minute)
|
||||
// Moved after redisIntegration initialization
|
||||
|
||||
|
||||
// DatabaseManager is now a singleton exported instance, but let's keep consistency if possible
|
||||
// In databaseManager.js it exports `dbManager` instance by default.
|
||||
// The original code was `const dbManager = new DatabaseManager(config.db);` which implies it might have been a class export.
|
||||
// Let's check `databaseManager.js` content.
|
||||
// Wait, I imported `dbManager` from `./db/databaseManager.js`.
|
||||
// If `databaseManager.js` exports an instance as default, I should use that.
|
||||
// If it exports a class, I should instantiate it.
|
||||
|
||||
// Let's assume the previous code `new DatabaseManager` was correct if it was a class.
|
||||
// BUT I used `dbManager.pool` in `partitionManager.js` assuming it's an instance.
|
||||
// I need to verify `databaseManager.js`.
|
||||
|
||||
const redisClient = await createRedisClient(config.redis);
|
||||
const redisIntegration = new RedisIntegration(
|
||||
redisClient,
|
||||
@@ -199,13 +171,8 @@ const bootstrap = async () => {
|
||||
);
|
||||
};
|
||||
|
||||
const isMissingPartitionError = (err) =>
|
||||
err?.code === '23514' ||
|
||||
(typeof err?.message === 'string' && err.message.includes('no partition of relation'));
|
||||
|
||||
const insertRowsWithRetry = async (rows) => {
|
||||
const startedAt = Date.now();
|
||||
let attemptedPartitionFix = false;
|
||||
while (true) {
|
||||
try {
|
||||
await dbManager.insertRows({ schema: config.db.schema, table: config.db.table, rows });
|
||||
@@ -222,24 +189,6 @@ const bootstrap = async () => {
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (isMissingPartitionError(err) && !attemptedPartitionFix) {
|
||||
attemptedPartitionFix = true;
|
||||
try {
|
||||
await partitionManager.ensurePartitionsForTimestamps(rows.map(r => r.ts_ms));
|
||||
} catch (partitionErr) {
|
||||
if (isDbConnectionError(partitionErr)) {
|
||||
logger.error('Database offline during partition ensure. Retrying in 5s...', { error: partitionErr.message });
|
||||
await new Promise(r => setTimeout(r, 5000));
|
||||
while (!(await dbManager.checkConnection())) {
|
||||
logger.warn('Database still offline. Waiting 5s...');
|
||||
await new Promise(r => setTimeout(r, 5000));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
throw partitionErr;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user