Mr Chen
GitHub
commit
0ba89cbb9e
|
|
@ -0,0 +1,259 @@
|
|||
# coding:utf-8
|
||||
|
||||
import sys
|
||||
import io
|
||||
import os
|
||||
import time
|
||||
import json
|
||||
import fcntl
|
||||
|
||||
web_dir = os.getcwd() + "/web"
|
||||
if os.path.exists(web_dir):
|
||||
sys.path.append(web_dir)
|
||||
os.chdir(web_dir)
|
||||
|
||||
import core.mw as mw
|
||||
from utils.crontab import crontab as MwCrontab
|
||||
|
||||
app_debug = False
|
||||
if mw.isAppleSystem():
|
||||
app_debug = True
|
||||
|
||||
|
||||
def getPluginName():
|
||||
return 'webstats'
|
||||
|
||||
|
||||
def getPluginDir():
|
||||
return mw.getPluginDir() + '/' + getPluginName()
|
||||
|
||||
|
||||
def getServerDir():
|
||||
return mw.getServerDir() + '/' + getPluginName()
|
||||
|
||||
|
||||
def getConf():
|
||||
conf = getServerDir() + "/lua/config.json"
|
||||
return conf
|
||||
|
||||
|
||||
def getGlobalConf():
|
||||
conf = getConf()
|
||||
content = mw.readFile(conf)
|
||||
result = json.loads(content)
|
||||
return result
|
||||
|
||||
|
||||
def pSqliteDb(dbname='web_logs', site_name='unset', fn="logs"):
|
||||
db_dir = getServerDir() + '/logs/' + site_name
|
||||
if not os.path.exists(db_dir):
|
||||
mw.execShell('mkdir -p ' + db_dir)
|
||||
|
||||
name = fn
|
||||
file = db_dir + '/' + name + '.db'
|
||||
|
||||
if not os.path.exists(file):
|
||||
conn = mw.M(dbname).dbPos(db_dir, name)
|
||||
sql = mw.readFile(getPluginDir() + '/conf/init.sql')
|
||||
sql_list = sql.split(';')
|
||||
for index in range(len(sql_list)):
|
||||
conn.execute(sql_list[index], ())
|
||||
else:
|
||||
conn = mw.M(dbname).dbPos(db_dir, name)
|
||||
|
||||
conn.execute("PRAGMA synchronous = 0", ())
|
||||
conn.execute("PRAGMA page_size = 4096", ())
|
||||
conn.execute("PRAGMA journal_mode = wal", ())
|
||||
|
||||
conn.autoTextFactory()
|
||||
|
||||
return conn
|
||||
|
||||
|
||||
def migrateSiteHotLogs(site_name, query_date):
|
||||
start_time = time.time()
|
||||
print(f"[{site_name}] 开始迁移日志...")
|
||||
|
||||
migrating_flag = getServerDir() + "/logs/%s/migrating" % site_name
|
||||
hot_db = getServerDir() + "/logs/%s/logs.db" % site_name
|
||||
hot_db_tmp = getServerDir() + "/logs/%s/logs_tmp.db" % site_name
|
||||
history_logs_db = getServerDir() + "/logs/%s/history_logs.db" % site_name
|
||||
|
||||
if not os.path.exists(hot_db):
|
||||
print(f"[{site_name}] 热日志数据库不存在，跳过")
|
||||
return mw.returnMsg(True, f"{site_name} logs not exist, skip")
|
||||
|
||||
# 检查是否正在迁移
|
||||
if os.path.exists(migrating_flag):
|
||||
print(f"[{site_name}] 正在迁移中，跳过")
|
||||
return mw.returnMsg(True, f"{site_name} is migrating, skip")
|
||||
|
||||
todayTime = time.strftime('%Y-%m-%d 00:00:00', time.localtime())
|
||||
todayUt = int(time.mktime(time.strptime(
|
||||
todayTime, "%Y-%m-%d %H:%M:%S")))
|
||||
|
||||
# 1. 备份到临时文件（copy 期间短暂互斥，完成后立即解除）
|
||||
try:
|
||||
import shutil
|
||||
print(f"[{site_name}] 备份 {hot_db} -> {hot_db_tmp} ...")
|
||||
mw.writeFile(migrating_flag, "yes")
|
||||
time.sleep(0.5)
|
||||
shutil.copy(hot_db, hot_db_tmp)
|
||||
if not os.path.exists(hot_db_tmp):
|
||||
return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!")
|
||||
except Exception as e:
|
||||
return mw.returnMsg(False, f"{site_name} migrating fail: {e}")
|
||||
finally:
|
||||
if os.path.exists(migrating_flag):
|
||||
os.remove(migrating_flag)
|
||||
|
||||
# 2. 从临时备份中迁移热日志数据到历史日志（读 logs_tmp，不阻塞 live 写入）
|
||||
try:
|
||||
logs_conn = pSqliteDb('web_log', site_name, 'logs_tmp')
|
||||
history_logs_conn = pSqliteDb('web_log', site_name, 'history_logs')
|
||||
|
||||
hot_db_columns = logs_conn.originExecute(
|
||||
"PRAGMA table_info([web_logs])")
|
||||
_columns = [c[1] for c in hot_db_columns if c[1] != "id"]
|
||||
columns_str = ",".join(_columns)
|
||||
placeholders = ",".join(["?"] * len(_columns))
|
||||
|
||||
logs_sql = f"select {columns_str} from web_logs where time<{todayUt}"
|
||||
selector = logs_conn.originExecute(logs_sql)
|
||||
|
||||
# 批量插入配置
|
||||
batch_size = 10000
|
||||
batch_count = 0
|
||||
total_count = 0
|
||||
insert_sql = f"insert into web_logs({columns_str}) values({placeholders})"
|
||||
|
||||
while True:
|
||||
logs = selector.fetchmany(batch_size)
|
||||
if not logs:
|
||||
break
|
||||
|
||||
batch_count += 1
|
||||
total_count += len(logs)
|
||||
# 使用原生 executemany 批量插入
|
||||
history_logs_conn.executemany(insert_sql, logs)
|
||||
history_logs_conn.commit()
|
||||
|
||||
if batch_count % 10 == 0:
|
||||
elapsed = time.time() - start_time
|
||||
print(f"[{site_name}] 已迁移 {total_count} 条记录, 耗时 {elapsed:.2f}s")
|
||||
|
||||
print(f"[{site_name}] 共迁移 {total_count} 条记录")
|
||||
|
||||
# 清理历史数据
|
||||
gcfg = getGlobalConf()
|
||||
save_day = gcfg['global']["save_day"]
|
||||
print(f"[{site_name}] 删除 {save_day} 天前的历史数据...")
|
||||
|
||||
time_now = time.localtime()
|
||||
save_timestamp = time.mktime((time_now.tm_year, time_now.tm_mon, time_now.tm_mday - save_day, 0, 0, 0, 0, 0, 0))
|
||||
delete_sql = f"delete from web_logs where time <= {save_timestamp}"
|
||||
history_logs_conn.execute(delete_sql)
|
||||
history_logs_conn.commit()
|
||||
|
||||
history_logs_conn.execute("VACUUM;")
|
||||
history_logs_conn.commit()
|
||||
|
||||
except Exception as e:
|
||||
if os.path.exists(hot_db_tmp):
|
||||
os.remove(hot_db_tmp)
|
||||
print(f"[{site_name}] logs to history error: {e}")
|
||||
return mw.returnMsg(False, f"{site_name} logs migrate error: {e}")
|
||||
|
||||
# 3. 删除已迁移的数据并清理统计（仅 DELETE/VACUUM 期间互斥）
|
||||
try:
|
||||
mw.writeFile(migrating_flag, "yes")
|
||||
time.sleep(0.5)
|
||||
|
||||
hot_db_conn = pSqliteDb('web_logs', site_name)
|
||||
|
||||
# 分批删除热日志
|
||||
del_hot_log = f"delete from web_logs where time<{todayUt}"
|
||||
print(f"[{site_name}] 删除已迁移的热日志...")
|
||||
hot_db_conn.execute(del_hot_log)
|
||||
|
||||
# 删除过期统计数据
|
||||
print(f"[{site_name}] 删除180天前的统计数据...")
|
||||
save_time_key = time.strftime(
|
||||
'%Y%m%d00', time.localtime(time.time() - 180 * 86400))
|
||||
|
||||
del_request_stat_sql = f"delete from request_stat where time<={save_time_key}"
|
||||
hot_db_conn.execute(del_request_stat_sql)
|
||||
hot_db_conn.execute(f"delete from spider_stat where time<={save_time_key}")
|
||||
hot_db_conn.execute(f"delete from client_stat where time<={save_time_key}")
|
||||
hot_db_conn.execute(f"delete from referer_stat where time<={save_time_key}")
|
||||
|
||||
hot_db_conn.commit()
|
||||
print(f"[{site_name}] 压缩热数据库...")
|
||||
hot_db_conn.execute("VACUUM;")
|
||||
hot_db_conn.commit()
|
||||
|
||||
except Exception as e:
|
||||
print(f"[{site_name}] delete hot logs error: {e}")
|
||||
finally:
|
||||
if os.path.exists(migrating_flag):
|
||||
os.remove(migrating_flag)
|
||||
if os.path.exists(hot_db_tmp):
|
||||
os.remove(hot_db_tmp)
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
print(f"[{site_name}] 日志迁移完成，耗时 {elapsed:.2f}s")
|
||||
|
||||
if not mw.isAppleSystem():
|
||||
mw.execShell("chown -R www:www " + getServerDir())
|
||||
|
||||
return mw.returnMsg(True, f"{site_name} logs migrate ok")
|
||||
|
||||
|
||||
def migrateHotLogs(query_date="today"):
|
||||
# 使用文件锁防止并发迁移
|
||||
lock_file = getServerDir() + "/logs/migrate_hot_logs.lock"
|
||||
lock_fd = None
|
||||
|
||||
try:
|
||||
lock_fd = open(lock_file, 'w')
|
||||
fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
|
||||
print("开始迁移热日志...")
|
||||
sites = mw.M('sites').field('name').order("add_time").select()
|
||||
|
||||
unset_site = {"name": "unset"}
|
||||
sites.append(unset_site)
|
||||
|
||||
total_sites = len(sites)
|
||||
success_count = 0
|
||||
fail_count = 0
|
||||
|
||||
for i, site_info in enumerate(sites):
|
||||
site_name = site_info["name"]
|
||||
print(f"\n[{i+1}/{total_sites}] 处理站点: {site_name}")
|
||||
migrate_res = migrateSiteHotLogs(site_name, query_date)
|
||||
if migrate_res["status"]:
|
||||
success_count += 1
|
||||
else:
|
||||
fail_count += 1
|
||||
print(f"[{site_name}] 迁移失败: {migrate_res['msg']}")
|
||||
|
||||
print(f"\n迁移完成! 成功: {success_count}, 失败: {fail_count}")
|
||||
|
||||
return mw.returnMsg(True, f"logs migrate ok, success: {success_count}, fail: {fail_count}")
|
||||
|
||||
except BlockingIOError:
|
||||
print("正在迁移中，跳过")
|
||||
return mw.returnMsg(True, "正在迁移中，跳过")
|
||||
except Exception as e:
|
||||
print(f"migrateHotLogs error: {e}")
|
||||
return mw.returnMsg(False, f"migrateHotLogs error: {e}")
|
||||
finally:
|
||||
if lock_fd:
|
||||
try:
|
||||
fcntl.flock(lock_fd.fileno(), fcntl.LOCK_UN)
|
||||
lock_fd.close()
|
||||
if os.path.exists(lock_file):
|
||||
os.remove(lock_file)
|
||||
except:
|
||||
pass
|
||||
|
|
@ -117,15 +117,11 @@ end
|
|||
- journal_size_limit = 20GB: WAL 文件大小限制
|
||||
]]
|
||||
function _M.initDB(self, input_sn)
|
||||
log_dir = self.app_dir .. "/logs"
|
||||
if log_dir == "" then
|
||||
return nil
|
||||
end
|
||||
|
||||
local log_dir = self.app_dir .. "/logs"
|
||||
local path = log_dir .. '/' .. input_sn .. "/logs.db"
|
||||
local db, err = sqlite3.open(path)
|
||||
|
||||
if err then
|
||||
self:D("initDB failed for " .. input_sn .. ": " .. tostring(err))
|
||||
return nil
|
||||
end
|
||||
|
||||
|
|
@ -401,41 +397,43 @@ function _M.cronPre(self)
|
|||
for _, site_v in ipairs(sites) do
|
||||
local input_sn = site_v["name"]
|
||||
|
||||
-- 安全地初始化数据库连接
|
||||
local db_ok, db = pcall(function() return self:initDB(input_sn) end)
|
||||
if not db_ok or not db then
|
||||
self:D("cronPre initDB failed for " .. tostring(input_sn) .. ": " .. tostring(db))
|
||||
return false
|
||||
end
|
||||
|
||||
-- 开启事务
|
||||
db:exec([[BEGIN TRANSACTION]])
|
||||
|
||||
local success = true
|
||||
-- 预创建当前小时和下一小时的统计记录
|
||||
for _, ws_v in ipairs(wc_stat) do
|
||||
if not self:_update_stat_pre(db, ws_v, time_key) then
|
||||
success = false
|
||||
break
|
||||
if not self:is_migrating(input_sn) then
|
||||
-- 安全地初始化数据库连接
|
||||
local db_ok, db = pcall(function() return self:initDB(input_sn) end)
|
||||
if not db_ok or not db then
|
||||
self:D("cronPre initDB failed for " .. tostring(input_sn) .. ": " .. tostring(db))
|
||||
return false
|
||||
end
|
||||
if not self:_update_stat_pre(db, ws_v, time_key_next) then
|
||||
success = false
|
||||
break
|
||||
|
||||
-- 开启事务
|
||||
db:exec([[BEGIN TRANSACTION]])
|
||||
|
||||
local success = true
|
||||
-- 预创建当前小时和下一小时的统计记录
|
||||
for _, ws_v in ipairs(wc_stat) do
|
||||
if not self:_update_stat_pre(db, ws_v, time_key) then
|
||||
success = false
|
||||
break
|
||||
end
|
||||
if not self:_update_stat_pre(db, ws_v, time_key_next) then
|
||||
success = false
|
||||
break
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
-- 提交或回滚事务
|
||||
if success then
|
||||
pcall(function() db:execute([[COMMIT]]) end)
|
||||
else
|
||||
pcall(function() db:exec([[ROLLBACK]]) end)
|
||||
end
|
||||
-- 提交或回滚事务
|
||||
if success then
|
||||
pcall(function() db:execute([[COMMIT]]) end)
|
||||
else
|
||||
pcall(function() db:exec([[ROLLBACK]]) end)
|
||||
end
|
||||
|
||||
-- 安全关闭数据库
|
||||
pcall(function() db:close() end)
|
||||
-- 安全关闭数据库
|
||||
pcall(function() db:close() end)
|
||||
|
||||
if not success then
|
||||
return false
|
||||
if not success then
|
||||
return false
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
|
|
@ -519,45 +517,45 @@ function _M.cron(self)
|
|||
end
|
||||
|
||||
local ok, err = pcall(function()
|
||||
if self:is_working('cron_init_stat') then
|
||||
return
|
||||
end
|
||||
|
||||
for _, site_v in ipairs(sites) do
|
||||
local input_sn = site_v["name"]
|
||||
if self:is_migrating(input_sn) then
|
||||
return
|
||||
end
|
||||
|
||||
if self:is_working('cron_init_stat') then
|
||||
return
|
||||
end
|
||||
|
||||
local db_ok, db = pcall(function() return self:initDB(input_sn) end)
|
||||
if not db_ok or not db then
|
||||
self:D("initDB failed for " .. input_sn .. ": " .. tostring(db))
|
||||
return
|
||||
end
|
||||
|
||||
stat_fields[input_sn] = {}
|
||||
|
||||
dbs[input_sn] = db
|
||||
self:clean_stats(db, input_sn)
|
||||
|
||||
local stmt_ok, stmt = pcall(function()
|
||||
return db:prepare[[INSERT INTO web_logs(
|
||||
time, ip, scheme, domain, server_name, method, status_code, uri, body_length,
|
||||
referer, user_agent, protocol, request_time, is_spider, request_headers, ip_list, client_port)
|
||||
VALUES(:time, :ip, :scheme, :domain, :server_name, :method, :status_code, :uri,
|
||||
:body_length, :referer, :user_agent, :protocol, :request_time, :is_spider,
|
||||
:request_headers, :ip_list, :client_port)]]
|
||||
end)
|
||||
|
||||
if stmt_ok and stmt then
|
||||
stmts[input_sn] = stmt
|
||||
db:exec([[BEGIN TRANSACTION]])
|
||||
-- 该站点迁移中，跳过；其它站点继续写入
|
||||
else
|
||||
if db and db:isopen() then
|
||||
db:close()
|
||||
local db_ok, db = pcall(function() return self:initDB(input_sn) end)
|
||||
if not db_ok or not db then
|
||||
self:D("initDB failed for " .. input_sn .. ": " .. tostring(db))
|
||||
return
|
||||
end
|
||||
|
||||
stat_fields[input_sn] = {}
|
||||
|
||||
dbs[input_sn] = db
|
||||
self:clean_stats(db, input_sn)
|
||||
|
||||
local stmt_ok, stmt = pcall(function()
|
||||
return db:prepare[[INSERT INTO web_logs(
|
||||
time, ip, scheme, domain, server_name, method, status_code, uri, body_length,
|
||||
referer, user_agent, protocol, request_time, is_spider, request_headers, ip_list, client_port)
|
||||
VALUES(:time, :ip, :scheme, :domain, :server_name, :method, :status_code, :uri,
|
||||
:body_length, :referer, :user_agent, :protocol, :request_time, :is_spider,
|
||||
:request_headers, :ip_list, :client_port)]]
|
||||
end)
|
||||
|
||||
if stmt_ok and stmt then
|
||||
stmts[input_sn] = stmt
|
||||
db:exec([[BEGIN TRANSACTION]])
|
||||
else
|
||||
if db and db:isopen() then
|
||||
db:close()
|
||||
end
|
||||
dbs[input_sn] = nil
|
||||
return
|
||||
end
|
||||
dbs[input_sn] = nil
|
||||
return
|
||||
end
|
||||
end
|
||||
|
||||
|
|
@ -575,81 +573,79 @@ function _M.cron(self)
|
|||
end
|
||||
|
||||
local input_sn = info['server_name']
|
||||
local db = dbs[input_sn]
|
||||
if not db then
|
||||
if self:is_migrating(input_sn) then
|
||||
ngx.shared.mw_total:rpush(total_key, data)
|
||||
return
|
||||
end
|
||||
else
|
||||
local db = dbs[input_sn]
|
||||
local stmt = stmts[input_sn]
|
||||
if not db or not stmt then
|
||||
ngx.shared.mw_total:rpush(total_key, data)
|
||||
else
|
||||
local insert_ok = self:store_logs_line(db, stmt, input_sn, info)
|
||||
if not insert_ok then
|
||||
self:D("store_logs_line failed for " .. input_sn .. ": " .. tostring(insert_ok))
|
||||
ngx.shared.mw_total:rpush(total_key, data)
|
||||
rollback_sites[input_sn] = true
|
||||
else
|
||||
local log_kv = info["log_kv"]
|
||||
local excluded = log_kv['excluded']
|
||||
local stat_tmp_fields = info['stat_fields']
|
||||
|
||||
local stmt = stmts[input_sn]
|
||||
if not stmt then
|
||||
ngx.shared.mw_total:rpush(total_key, data)
|
||||
return
|
||||
end
|
||||
local stat_fields_is = stat_fields[input_sn]
|
||||
for stf_k, stf_v in pairs(stat_tmp_fields) do
|
||||
if excluded then
|
||||
if stf_k == "spider_stat_fields" or stf_k == "client_stat_fields" then
|
||||
break
|
||||
end
|
||||
end
|
||||
|
||||
local insert_ok = self:store_logs_line(db, stmt, input_sn, info)
|
||||
if not insert_ok then
|
||||
ngx.shared.mw_total:rpush(total_key, data)
|
||||
rollback_sites[input_sn] = true
|
||||
return
|
||||
end
|
||||
local stf_is = stat_fields_is[stf_k]
|
||||
if not stf_is then
|
||||
stf_is = {}
|
||||
stat_fields_is[stf_k] = stf_is
|
||||
end
|
||||
|
||||
local log_kv = info["log_kv"]
|
||||
local excluded = log_kv['excluded']
|
||||
local stat_tmp_fields = info['stat_fields']
|
||||
for sv_k, sv_v in pairs(stf_v) do
|
||||
stf_is[sv_k] = (stf_is[sv_k] or 0) + sv_v
|
||||
end
|
||||
end
|
||||
|
||||
local stat_fields_is = stat_fields[input_sn]
|
||||
for stf_k, stf_v in pairs(stat_tmp_fields) do
|
||||
if excluded then
|
||||
if stf_k == "spider_stat_fields" or stf_k == "client_stat_fields" then
|
||||
break
|
||||
if not excluded then
|
||||
local ip = log_kv['ip']
|
||||
local body_length = log_kv["body_length"]
|
||||
|
||||
local ip_stats_sn = ip_stats[input_sn]
|
||||
if not ip_stats_sn then
|
||||
ip_stats_sn = {}
|
||||
ip_stats[input_sn] = ip_stats_sn
|
||||
end
|
||||
|
||||
local ip_stat = ip_stats_sn[ip]
|
||||
if not ip_stat then
|
||||
ip_stats_sn[ip] = {ip_num = 1, body_length = body_length}
|
||||
else
|
||||
ip_stat.ip_num = ip_stat.ip_num + 1
|
||||
ip_stat.body_length = ip_stat.body_length + body_length
|
||||
end
|
||||
|
||||
local url_stats_sn = url_stats[input_sn]
|
||||
if not url_stats_sn then
|
||||
url_stats_sn = {}
|
||||
url_stats[input_sn] = url_stats_sn
|
||||
end
|
||||
|
||||
local request_uri = log_kv["request_uri"]
|
||||
local request_uri_md5 = ngx.md5(request_uri)
|
||||
local url_stat = url_stats_sn[request_uri_md5]
|
||||
if not url_stat then
|
||||
url_stats_sn[request_uri_md5] = {url_num = 1, uri = request_uri, body_length = body_length}
|
||||
else
|
||||
url_stat.url_num = url_stat.url_num + 1
|
||||
url_stat.body_length = url_stat.body_length + body_length
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local stf_is = stat_fields_is[stf_k]
|
||||
if not stf_is then
|
||||
stf_is = {}
|
||||
stat_fields_is[stf_k] = stf_is
|
||||
end
|
||||
|
||||
for sv_k, sv_v in pairs(stf_v) do
|
||||
stf_is[sv_k] = (stf_is[sv_k] or 0) + sv_v
|
||||
end
|
||||
end
|
||||
|
||||
if not excluded then
|
||||
local ip = log_kv['ip']
|
||||
local body_length = log_kv["body_length"]
|
||||
|
||||
local ip_stats_sn = ip_stats[input_sn]
|
||||
if not ip_stats_sn then
|
||||
ip_stats_sn = {}
|
||||
ip_stats[input_sn] = ip_stats_sn
|
||||
end
|
||||
|
||||
local ip_stat = ip_stats_sn[ip]
|
||||
if not ip_stat then
|
||||
ip_stats_sn[ip] = {ip_num = 1, body_length = body_length}
|
||||
else
|
||||
ip_stat.ip_num = ip_stat.ip_num + 1
|
||||
ip_stat.body_length = ip_stat.body_length + body_length
|
||||
end
|
||||
|
||||
local url_stats_sn = url_stats[input_sn]
|
||||
if not url_stats_sn then
|
||||
url_stats_sn = {}
|
||||
url_stats[input_sn] = url_stats_sn
|
||||
end
|
||||
|
||||
local request_uri = log_kv["request_uri"]
|
||||
local request_uri_md5 = ngx.md5(request_uri)
|
||||
local url_stat = url_stats_sn[request_uri_md5]
|
||||
if not url_stat then
|
||||
url_stats_sn[request_uri_md5] = {url_num = 1, uri = request_uri, body_length = body_length}
|
||||
else
|
||||
url_stat.url_num = url_stat.url_num + 1
|
||||
url_stat.body_length = url_stat.body_length + body_length
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
|
|
@ -960,10 +956,16 @@ function _M.D(self,msg)
|
|||
end
|
||||
|
||||
function _M.is_migrating(self, input_sn)
|
||||
local file = io.open(self.app_dir.."/migrating", "rb")
|
||||
if file then return true end
|
||||
local file = io.open(self.app_dir.."/logs/"..input_sn.."/migrating", "rb")
|
||||
if file then return true end
|
||||
local fp = io.open(self.app_dir.."/migrating", "rb")
|
||||
if fp then
|
||||
fp:close()
|
||||
return true
|
||||
end
|
||||
local fp = io.open(self.app_dir.."/logs/"..input_sn.."/migrating", "rb")
|
||||
if fp then
|
||||
fp:close()
|
||||
return true
|
||||
end
|
||||
return false
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -276,7 +276,11 @@ local function run_app()
|
|||
log_kv = kv,
|
||||
}
|
||||
|
||||
cache:rpush(total_key, json.encode(data))
|
||||
local len, err, forcible = cache:rpush(total_key, json.encode(data))
|
||||
if not len then
|
||||
C:D("webstats rpush failed: " .. tostring(err or "unknown")
|
||||
.. ", forcible=" .. tostring(forcible) .. ", queue=" .. total_key)
|
||||
end
|
||||
end
|
||||
|
||||
-- C:D("webstats_log run_app start, server_name=" .. tostring(server_name))
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import os
|
|||
import time
|
||||
import json
|
||||
import fcntl
|
||||
import shutil
|
||||
|
||||
web_dir = os.getcwd() + "/web"
|
||||
if os.path.exists(web_dir):
|
||||
|
|
@ -43,7 +44,6 @@ def getGlobalConf():
|
|||
result = json.loads(content)
|
||||
return result
|
||||
|
||||
|
||||
def pSqliteDb(dbname='web_logs', site_name='unset', fn="logs"):
|
||||
db_dir = getServerDir() + '/logs/' + site_name
|
||||
if not os.path.exists(db_dir):
|
||||
|
|
@ -70,6 +70,23 @@ def pSqliteDb(dbname='web_logs', site_name='unset', fn="logs"):
|
|||
return conn
|
||||
|
||||
|
||||
def batch_delete(conn, table, where_clause, batch_size=5000, pause_sec=0.05):
|
||||
"""分批删除，缩短单次持锁时间，WAL 模式下可与 OpenResty 并发写入。"""
|
||||
total = 0
|
||||
while True:
|
||||
sql = (
|
||||
f"DELETE FROM {table} WHERE rowid IN "
|
||||
f"(SELECT rowid FROM {table} WHERE {where_clause} LIMIT {batch_size})"
|
||||
)
|
||||
affected = conn.execute(sql)
|
||||
if not isinstance(affected, int) or affected == 0:
|
||||
break
|
||||
total += affected
|
||||
if pause_sec > 0:
|
||||
time.sleep(pause_sec)
|
||||
return total
|
||||
|
||||
|
||||
def migrateSiteHotLogs(site_name, query_date):
|
||||
start_time = time.time()
|
||||
print(f"[{site_name}] 开始迁移日志...")
|
||||
|
|
@ -88,21 +105,29 @@ def migrateSiteHotLogs(site_name, query_date):
|
|||
print(f"[{site_name}] 正在迁移中，跳过")
|
||||
return mw.returnMsg(True, f"{site_name} is migrating, skip")
|
||||
|
||||
# 1. 备份到临时文件（使用文件直接拷贝）
|
||||
todayTime = time.strftime('%Y-%m-%d 00:00:00', time.localtime())
|
||||
todayUt = int(time.mktime(time.strptime(
|
||||
todayTime, "%Y-%m-%d %H:%M:%S")))
|
||||
|
||||
# 1. 备份到临时文件（copy 期间短暂互斥，完成后立即解除）
|
||||
try:
|
||||
import shutil
|
||||
print(f"[{site_name}] 备份 {hot_db} -> {hot_db_tmp} ...")
|
||||
mw.writeFile(migrating_flag, "yes")
|
||||
time.sleep(0.5)
|
||||
shutil.copy(hot_db, hot_db_tmp)
|
||||
copy_start = time.time()
|
||||
# import shutil
|
||||
# shutil.copy(hot_db, hot_db_tmp)
|
||||
mw.fastCopy(hot_db, hot_db_tmp, 256 * 1024)
|
||||
print(f"[{site_name}] 备份完成，耗时 {time.time() - copy_start:.2f}s")
|
||||
if not os.path.exists(hot_db_tmp):
|
||||
return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!")
|
||||
except Exception as e:
|
||||
return mw.returnMsg(False, f"{site_name} migrating fail: {e}")
|
||||
finally:
|
||||
if os.path.exists(migrating_flag):
|
||||
os.remove(migrating_flag)
|
||||
return mw.returnMsg(False, f"{site_name} migrating fail: {e}")
|
||||
|
||||
# 2. 从临时备份中迁移热日志数据到历史日志（批量插入）
|
||||
# 2. 从临时备份中迁移热日志数据到历史日志（读 logs_tmp，不阻塞 live 写入）
|
||||
try:
|
||||
logs_conn = pSqliteDb('web_log', site_name, 'logs_tmp')
|
||||
history_logs_conn = pSqliteDb('web_log', site_name, 'history_logs')
|
||||
|
|
@ -113,10 +138,6 @@ def migrateSiteHotLogs(site_name, query_date):
|
|||
columns_str = ",".join(_columns)
|
||||
placeholders = ",".join(["?"] * len(_columns))
|
||||
|
||||
todayTime = time.strftime('%Y-%m-%d 00:00:00', time.localtime())
|
||||
todayUt = int(time.mktime(time.strptime(
|
||||
todayTime, "%Y-%m-%d %H:%M:%S")))
|
||||
|
||||
logs_sql = f"select {columns_str} from web_logs where time<{todayUt}"
|
||||
selector = logs_conn.originExecute(logs_sql)
|
||||
|
||||
|
|
@ -158,49 +179,40 @@ def migrateSiteHotLogs(site_name, query_date):
|
|||
history_logs_conn.commit()
|
||||
|
||||
except Exception as e:
|
||||
if site_name:
|
||||
print(f"[{site_name}] logs to history error: {e}")
|
||||
else:
|
||||
print(f"logs to history error: {e}")
|
||||
if os.path.exists(hot_db_tmp):
|
||||
os.remove(hot_db_tmp)
|
||||
print(f"[{site_name}] logs to history error: {e}")
|
||||
return mw.returnMsg(False, f"{site_name} logs migrate error: {e}")
|
||||
|
||||
# 3. 删除已迁移的数据并清理统计（批量删除）
|
||||
# 3. 分批删除已迁移的热日志并清理统计（不设 migrating 锁，不阻塞 OpenResty 写入）
|
||||
try:
|
||||
if os.path.exists(migrating_flag):
|
||||
os.remove(migrating_flag)
|
||||
|
||||
mw.writeFile(migrating_flag, "yes")
|
||||
|
||||
hot_db_conn = pSqliteDb('web_logs', site_name)
|
||||
hot_db_conn.execute("PRAGMA busy_timeout = 30000")
|
||||
|
||||
# 分批删除热日志
|
||||
del_hot_log = f"delete from web_logs where time<{todayUt}"
|
||||
print(f"[{site_name}] 删除已迁移的热日志...")
|
||||
hot_db_conn.execute(del_hot_log)
|
||||
print(f"[{site_name}] 分批删除已迁移的热日志...")
|
||||
deleted = batch_delete(hot_db_conn, 'web_logs', f"time<{todayUt}")
|
||||
print(f"[{site_name}] 已删除 {deleted} 条热日志")
|
||||
|
||||
# 删除过期统计数据
|
||||
print(f"[{site_name}] 删除180天前的统计数据...")
|
||||
print(f"[{site_name}] 分批删除180天前的统计数据...")
|
||||
save_time_key = time.strftime(
|
||||
'%Y%m%d00', time.localtime(time.time() - 180 * 86400))
|
||||
stat_where = f"time<='{save_time_key}'"
|
||||
for stat_table in ('request_stat', 'spider_stat', 'client_stat', 'referer_stat'):
|
||||
stat_deleted = batch_delete(hot_db_conn, stat_table, stat_where)
|
||||
print(f"[{site_name}] {stat_table} 已删除 {stat_deleted} 条")
|
||||
|
||||
del_request_stat_sql = f"delete from request_stat where time<={save_time_key}"
|
||||
hot_db_conn.execute(del_request_stat_sql)
|
||||
hot_db_conn.execute(f"delete from spider_stat where time<={save_time_key}")
|
||||
hot_db_conn.execute(f"delete from client_stat where time<={save_time_key}")
|
||||
hot_db_conn.execute(f"delete from referer_stat where time<={save_time_key}")
|
||||
|
||||
hot_db_conn.commit()
|
||||
print(f"[{site_name}] 压缩热数据库...")
|
||||
hot_db_conn.execute("VACUUM;")
|
||||
hot_db_conn.commit()
|
||||
print(f"[{site_name}] 执行 WAL checkpoint...")
|
||||
hot_db_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
|
||||
except Exception as e:
|
||||
print(f"[{site_name}] delete hot logs error: {e}")
|
||||
finally:
|
||||
if os.path.exists(migrating_flag):
|
||||
os.remove(migrating_flag)
|
||||
if os.path.exists(hot_db_tmp):
|
||||
os.remove(hot_db_tmp)
|
||||
if os.path.exists(hot_db_tmp+"-shm"):
|
||||
os.remove(hot_db_tmp+"-shm")
|
||||
if os.path.exists(hot_db_tmp+"-wal"):
|
||||
os.remove(hot_db_tmp+"-wal")
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
print(f"[{site_name}] 日志迁移完成，耗时 {elapsed:.2f}s")
|
||||
|
|
@ -242,7 +254,6 @@ def migrateHotLogs(query_date="today"):
|
|||
|
||||
print(f"\n迁移完成! 成功: {success_count}, 失败: {fail_count}")
|
||||
|
||||
mw.opWeb('restart')
|
||||
return mw.returnMsg(True, f"logs migrate ok, success: {success_count}, fail: {fail_count}")
|
||||
|
||||
except BlockingIOError:
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ import subprocess
|
|||
import glob
|
||||
import base64
|
||||
import re
|
||||
import shutil
|
||||
|
||||
from random import Random
|
||||
|
||||
|
|
@ -318,6 +319,11 @@ def toSize(size, middle='') -> str:
|
|||
s = u
|
||||
return str(round(size, 2)) + middle + u
|
||||
|
||||
def fastCopy(src, dst, buffer_size=256 * 1024): # 128MB 缓冲区
|
||||
with open(src, 'rb') as fsrc:
|
||||
with open(dst, 'wb') as fdst:
|
||||
shutil.copyfileobj(fsrc, fdst, length=buffer_size)
|
||||
|
||||
def returnData(status, msg, data=None):
|
||||
if data is None:
|
||||
return {'status': status, 'msg': msg}
|
||||
|
|
|
|||
Loading…
Reference in New Issue