Commit f84b7d59 by CaIon

refactor(log): simplify ClickHouse log deletion and add unit tests

Replace the per-batch ClickHouse mutation loop with a single
ALTER TABLE ... DELETE, since ClickHouse DELETE is a heavy mutation
that rewrites data parts and per-batch mutations are pathologically
slow. Add deterministic unit tests covering ClickHouse DSN handling,
main-database rejection, TTL DDL generation, log ordering, request_id
backfill, and display id assignment.
parent a162163b
package model
import (
"os"
"strings"
"testing"
"github.com/QuantumNous/new-api/common"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestIsClickHouseDSN(t *testing.T) {
cases := []struct {
dsn string
want bool
}{
{"clickhouse://default:pass@localhost:9000/logs", true},
{"tcp://localhost:9000/logs", true},
{"http://localhost:8123/logs", true},
{"https://localhost:8443/logs", true},
{"postgres://root:pass@localhost:5432/db", false},
{"postgresql://root:pass@localhost:5432/db", false},
{"root:pass@tcp(localhost:3306)/db", false},
{"local", false},
{"", false},
}
for _, c := range cases {
assert.Equalf(t, c.want, isClickHouseDSN(c.dsn), "dsn=%q", c.dsn)
}
}
func TestNormalizeClickHouseDSN(t *testing.T) {
// https without secure gets secure=true appended
normalized := normalizeClickHouseDSN("https://default:pass@localhost:8443/logs")
assert.Contains(t, normalized, "secure=true")
assert.True(t, strings.HasPrefix(normalized, "https://"))
// https that already specifies secure is left untouched
assert.Equal(t,
"https://localhost:8443/logs?secure=false",
normalizeClickHouseDSN("https://localhost:8443/logs?secure=false"),
)
// non-https schemes are returned verbatim
assert.Equal(t, "clickhouse://localhost:9000/logs", normalizeClickHouseDSN("clickhouse://localhost:9000/logs"))
assert.Equal(t, "tcp://localhost:9000/logs", normalizeClickHouseDSN("tcp://localhost:9000/logs"))
}
func TestChooseDBRejectsClickHouseForMainDatabase(t *testing.T) {
original, had := os.LookupEnv("SQL_DSN")
t.Cleanup(func() {
if had {
require.NoError(t, os.Setenv("SQL_DSN", original))
} else {
require.NoError(t, os.Unsetenv("SQL_DSN"))
}
})
require.NoError(t, os.Setenv("SQL_DSN", "clickhouse://default:pass@localhost:9000/logs"))
db, dbType, err := chooseDB("SQL_DSN", false)
require.Error(t, err)
assert.Nil(t, db)
assert.Equal(t, common.DatabaseType(""), dbType)
assert.Contains(t, err.Error(), "does not support ClickHouse")
}
func TestClickHouseLogTTLExpression(t *testing.T) {
assert.Equal(t, "", clickHouseLogTTLExpression(0))
assert.Equal(t, "", clickHouseLogTTLExpression(-5))
assert.Equal(t, "toDateTime(created_at) + INTERVAL 30 DAY DELETE", clickHouseLogTTLExpression(30))
}
func TestClickHouseLogTTLClause(t *testing.T) {
assert.Equal(t, "", clickHouseLogTTLClause(0))
assert.Equal(t, "\nTTL toDateTime(created_at) + INTERVAL 7 DAY DELETE", clickHouseLogTTLClause(7))
}
func TestClickHouseLogCreateTableSQL(t *testing.T) {
withoutTTL := clickHouseLogCreateTableSQL(0)
assert.Contains(t, withoutTTL, "CREATE TABLE IF NOT EXISTS logs")
assert.Contains(t, withoutTTL, "ENGINE = MergeTree()")
assert.Contains(t, withoutTTL, "PARTITION BY toYYYYMM(toDateTime(created_at))")
assert.Contains(t, withoutTTL, "ORDER BY (created_at, request_id)")
assert.NotContains(t, withoutTTL, "TTL ")
withTTL := clickHouseLogCreateTableSQL(30)
assert.Contains(t, withTTL, "ORDER BY (created_at, request_id)")
assert.Contains(t, withTTL, "TTL toDateTime(created_at) + INTERVAL 30 DAY DELETE")
}
func TestClickHouseCreateTableHasTTL(t *testing.T) {
assert.True(t, clickHouseCreateTableHasTTL("CREATE TABLE logs (...)\nTTL toDateTime(created_at) + INTERVAL 30 DAY DELETE"))
assert.True(t, clickHouseCreateTableHasTTL("CREATE TABLE logs (...) TTL toDateTime(created_at)"))
assert.False(t, clickHouseCreateTableHasTTL("CREATE TABLE logs (...)\nORDER BY (created_at, request_id)"))
}
func TestClickHouseLogOrder(t *testing.T) {
assert.Equal(t, "created_at desc, request_id desc", clickHouseLogOrder(""))
assert.Equal(t, "logs.created_at desc, logs.request_id desc", clickHouseLogOrder("logs."))
}
func TestEnsureLogRequestId(t *testing.T) {
empty := &Log{}
ensureLogRequestId(empty)
assert.NotEmpty(t, empty.RequestId, "empty request id should be backfilled")
existing := &Log{RequestId: "fixed-request-id"}
ensureLogRequestId(existing)
assert.Equal(t, "fixed-request-id", existing.RequestId, "existing request id must be preserved")
assert.NotPanics(t, func() { ensureLogRequestId(nil) })
}
func TestAssignDisplayLogIds(t *testing.T) {
logs := []*Log{{}, {}, {}}
assignDisplayLogIds(logs, 0)
assert.Equal(t, []int{1, 2, 3}, []int{logs[0].Id, logs[1].Id, logs[2].Id})
assignDisplayLogIds(logs, 20)
assert.Equal(t, []int{21, 22, 23}, []int{logs[0].Id, logs[1].Id, logs[2].Id})
assert.NotPanics(t, func() { assignDisplayLogIds(nil, 0) })
}
...@@ -683,33 +683,24 @@ func DeleteOldLogBatch(ctx context.Context, targetTimestamp int64, limit int) (i ...@@ -683,33 +683,24 @@ func DeleteOldLogBatch(ctx context.Context, targetTimestamp int64, limit int) (i
} }
if common.UsingLogDatabase(common.DatabaseTypeClickHouse) { if common.UsingLogDatabase(common.DatabaseTypeClickHouse) {
var batchCount int64 // ClickHouse DELETE is a heavy mutation that rewrites data parts, so
if err := LOG_DB.WithContext(ctx).Raw(` // per-batch mutations would be pathologically slow. Remove all matching
SELECT count() FROM ( // rows in a single synchronous mutation regardless of limit; the reported
SELECT created_at, request_id // count lets the caller's progress loop complete in one pass.
FROM logs total, err := CountOldLog(ctx, targetTimestamp)
WHERE created_at < ? if err != nil {
ORDER BY created_at ASC, request_id ASC
LIMIT ?
)`, targetTimestamp, limit).Scan(&batchCount).Error; err != nil {
return 0, err return 0, err
} }
if batchCount == 0 { if total == 0 {
return 0, nil return 0, nil
} }
if err := LOG_DB.WithContext(ctx).Exec(
if err := LOG_DB.WithContext(ctx).Exec(` "ALTER TABLE logs DELETE WHERE created_at < ? SETTINGS mutations_sync = 1",
ALTER TABLE logs DELETE WHERE (created_at, request_id) IN ( targetTimestamp,
SELECT created_at, request_id ).Error; err != nil {
FROM logs
WHERE created_at < ?
ORDER BY created_at ASC, request_id ASC
LIMIT ?
) SETTINGS mutations_sync = 1`, targetTimestamp, limit).Error; err != nil {
return 0, err return 0, err
} }
return total, nil
return batchCount, nil
} }
result := LOG_DB.WithContext(ctx).Where("created_at < ?", targetTimestamp).Limit(limit).Delete(&Log{}) result := LOG_DB.WithContext(ctx).Where("created_at < ?", targetTimestamp).Limit(limit).Delete(&Log{})
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment