mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-09-28 05:46:36 +08:00
refactor(clickhouse): route of_node_access_logs through analytics repository
Move all ClickHouse DML for OpenFlare node access logs into internal/repository/analytics; model layer keeps domain aggregation and in-memory test store via a thin adapter. Aligns with goose-managed DDL and openflare database startup dependency.
This commit is contained in:
ryan
@@ -20,6 +20,7 @@ sidebar: false
|
||||
|
||||
- `of_node_access_logs` 从 PostgreSQL/SQLite 迁移至 ClickHouse（数据库 `openflare`）；ClickHouse 表结构改由 goose 独立迁移管线管理（`goose_clickhouse_version`），启动时在 `migrator.MigrateClickHouse()` 中执行，移除 `clickhouse_schema.go` 手写 DDL。
|
||||
- ClickHouse 分析库接入 GORM（`db.ChDB`）与 `internal/repository/analytics/`：用户访问日志（`w_user_access_logs`）读写、风控批量写入与管理端日志统计 API 统一经 repository 访问；新增 `internal/model/analytics/` 分析表模型。
|
||||
- OpenFlare 节点访问日志（`of_node_access_logs`）ClickHouse 读写迁入 `internal/repository/analytics/`；`model` 层保留聚合编排与内存测试 store，生产路径经薄适配器调用 repository。
|
||||
- ClickHouse 默认启用（`clickhouse.enabled: true`），`docker-compose` 默认启动 `clickhouse` 服务并纳入 `wavelet` 健康依赖。
|
||||
|
||||
### 移除
|
||||
|
||||
@@ -345,68 +345,6 @@ func openFlareAccessLogQueryFromBucket(query OpenFlareAccessLogBucketQuery) Open
|
||||
}
|
||||
}
|
||||
|
||||
func buildOpenFlareAccessLogFilterClause(query OpenFlareAccessLogQuery) (string, []any) {
|
||||
parts := make([]string, 0, 6)
|
||||
args := make([]any, 0, 6)
|
||||
if trimmed := strings.TrimSpace(query.NodeID); trimmed != "" {
|
||||
parts = append(parts, "node_id = ?")
|
||||
args = append(args, trimmed)
|
||||
}
|
||||
if trimmed := strings.TrimSpace(query.RemoteAddr); trimmed != "" {
|
||||
parts = append(parts, "remote_addr LIKE ?")
|
||||
args = append(args, trimmed+"%")
|
||||
}
|
||||
if trimmed := strings.TrimSpace(query.Host); trimmed != "" {
|
||||
parts = append(parts, "host LIKE ?")
|
||||
args = append(args, trimmed+"%")
|
||||
}
|
||||
if trimmed := strings.TrimSpace(query.Path); trimmed != "" {
|
||||
parts = append(parts, "path LIKE ?")
|
||||
args = append(args, trimmed+"%")
|
||||
}
|
||||
if !query.Since.IsZero() {
|
||||
parts = append(parts, "logged_at >= ?")
|
||||
args = append(args, query.Since.UTC())
|
||||
}
|
||||
if !query.Until.IsZero() {
|
||||
parts = append(parts, "logged_at < ?")
|
||||
args = append(args, query.Until.UTC())
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return "1", nil
|
||||
}
|
||||
return strings.Join(parts, " AND "), args
|
||||
}
|
||||
|
||||
func combineOpenFlareAccessLogSQLClauses(left string, right string) string {
|
||||
if strings.TrimSpace(left) == "" || left == "TRUE" || left == "1" {
|
||||
return right
|
||||
}
|
||||
return left + " AND " + right
|
||||
}
|
||||
|
||||
func openFlareAccessLogOrderClause(sortBy string, sortOrder string) string {
|
||||
direction := "DESC"
|
||||
if openFlareAccessLogNormalizeSortOrder(sortOrder) == "asc" {
|
||||
direction = "ASC"
|
||||
}
|
||||
column := "logged_at"
|
||||
switch strings.TrimSpace(sortBy) {
|
||||
case "status_code":
|
||||
column = "status_code"
|
||||
case "remote_addr":
|
||||
column = "remote_addr"
|
||||
case "host":
|
||||
column = "host"
|
||||
case "path":
|
||||
column = "path"
|
||||
}
|
||||
if column == "logged_at" {
|
||||
return column + " " + direction + ", id " + direction
|
||||
}
|
||||
return column + " " + direction + ", logged_at " + direction + ", id " + direction
|
||||
}
|
||||
|
||||
func sortOpenFlareAccessLogBucketIPRows(items []*OpenFlareAccessLogBucketIPRow, sortBy string, sortOrder string) {
|
||||
desc := openFlareAccessLogNormalizeSortOrder(sortOrder) != "asc"
|
||||
sort.Slice(items, func(i, j int) bool {
|
||||
|
||||
@@ -5,19 +5,13 @@ package model
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
"github.com/Rain-kl/Wavelet/internal/db/idgen"
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
analyticsrepo "github.com/Rain-kl/Wavelet/internal/repository/analytics"
|
||||
)
|
||||
|
||||
const openFlareAccessLogTable = "of_node_access_logs"
|
||||
|
||||
type accessLogStore interface {
|
||||
InsertBatch(ctx context.Context, records []*OpenFlareAccessLog) error
|
||||
List(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error)
|
||||
@@ -69,397 +63,183 @@ func NewMemoryAccessLogStore() accessLogStore {
|
||||
|
||||
type clickhouseAccessLogStore struct{}
|
||||
|
||||
func (clickhouseAccessLogStore) conn() (driver.Conn, error) {
|
||||
if db.ChConn == nil {
|
||||
return nil, errors.New(errClickHouseNotInitialized)
|
||||
}
|
||||
return db.ChConn, nil
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) InsertBatch(ctx context.Context, records []*OpenFlareAccessLog) error {
|
||||
if len(records) == 0 {
|
||||
return nil
|
||||
}
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
batch, err := conn.PrepareBatch(ctx, fmt.Sprintf(
|
||||
"INSERT INTO %s (id, node_id, logged_at, remote_addr, region, host, path, status_code, created_at)",
|
||||
openFlareAccessLogTable,
|
||||
))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
logs := make([]analyticsmodel.NodeAccessLog, 0, len(records))
|
||||
for _, record := range records {
|
||||
if record == nil {
|
||||
continue
|
||||
}
|
||||
id := record.ID
|
||||
if id == 0 {
|
||||
id = uint(idgen.NextUint64ID())
|
||||
}
|
||||
createdAt := record.CreatedAt
|
||||
if createdAt.IsZero() {
|
||||
createdAt = now
|
||||
}
|
||||
if err := batch.Append(
|
||||
uint64(id),
|
||||
record.NodeID,
|
||||
record.LoggedAt.UTC(),
|
||||
record.RemoteAddr,
|
||||
record.Region,
|
||||
record.Host,
|
||||
record.Path,
|
||||
int32(record.StatusCode),
|
||||
createdAt.UTC(),
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
logs = append(logs, toAnalyticsNodeAccessLog(record))
|
||||
}
|
||||
return batch.Send()
|
||||
return analyticsrepo.BatchInsertNodeAccessLogs(ctx, logs)
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) List(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
rows, err := analyticsrepo.ListNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildOpenFlareAccessLogFilterClause(query)
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT id, node_id, logged_at, remote_addr, region, host, path, status_code, created_at
|
||||
FROM %s
|
||||
WHERE %s
|
||||
ORDER BY %s`, openFlareAccessLogTable, clause, openFlareAccessLogOrderClause(query.SortBy, query.SortOrder))
|
||||
if query.PageSize > 0 {
|
||||
if query.Page < 0 {
|
||||
query.Page = 0
|
||||
}
|
||||
sql += " LIMIT ? OFFSET ?"
|
||||
args = append(args, query.PageSize, query.Page*query.PageSize)
|
||||
}
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
return scanOpenFlareAccessLogRows(rows)
|
||||
}
|
||||
|
||||
func scanOpenFlareAccessLogRows(rows driver.Rows) ([]*OpenFlareAccessLog, error) {
|
||||
var result []*OpenFlareAccessLog
|
||||
for rows.Next() {
|
||||
var (
|
||||
id uint64
|
||||
nodeID string
|
||||
loggedAt time.Time
|
||||
remoteAddr string
|
||||
region string
|
||||
host string
|
||||
path string
|
||||
statusCode int32
|
||||
createdAt time.Time
|
||||
)
|
||||
if err := rows.Scan(&id, &nodeID, &loggedAt, &remoteAddr, ®ion, &host, &path, &statusCode, &createdAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result = append(result, &OpenFlareAccessLog{
|
||||
ID: uint(id),
|
||||
NodeID: nodeID,
|
||||
LoggedAt: loggedAt.UTC(),
|
||||
RemoteAddr: remoteAddr,
|
||||
Region: region,
|
||||
Host: host,
|
||||
Path: path,
|
||||
StatusCode: int(statusCode),
|
||||
CreatedAt: createdAt.UTC(),
|
||||
})
|
||||
}
|
||||
return result, nil
|
||||
return fromAnalyticsNodeAccessLogs(rows), nil
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) Count(ctx context.Context, query OpenFlareAccessLogQuery) (int64, int64, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
clause, args := buildOpenFlareAccessLogFilterClause(query)
|
||||
var totalRecords int64
|
||||
countSQL := fmt.Sprintf("SELECT count() FROM %s WHERE %s", openFlareAccessLogTable, clause)
|
||||
if err := conn.QueryRow(ctx, countSQL, args...).Scan(&totalRecords); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
ipSQL := fmt.Sprintf(`
|
||||
SELECT count() FROM (
|
||||
SELECT trim(remote_addr) AS remote_addr
|
||||
FROM %s
|
||||
WHERE %s AND remote_addr != ''
|
||||
GROUP BY trim(remote_addr)
|
||||
)`, openFlareAccessLogTable, clause)
|
||||
var totalIPs int64
|
||||
if err := conn.QueryRow(ctx, ipSQL, args...).Scan(&totalIPs); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
return totalRecords, totalIPs, nil
|
||||
return analyticsrepo.CountNodeAccessLogs(ctx, toNodeAccessLogFilter(query))
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) RegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*OpenFlareAccessLogRegionCount, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
rows, err := analyticsrepo.RegionCountsNodeAccessLogs(ctx, nodeID, since, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
filter := OpenFlareAccessLogQuery{NodeID: nodeID, Since: since}
|
||||
clause, args := buildOpenFlareAccessLogFilterClause(filter)
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT trim(region) AS region, count() AS count
|
||||
FROM %s
|
||||
WHERE %s AND trim(region) != ''
|
||||
GROUP BY trim(region)
|
||||
ORDER BY count DESC, region ASC`, openFlareAccessLogTable, clause)
|
||||
if limit > 0 {
|
||||
sql += " LIMIT ?"
|
||||
args = append(args, limit)
|
||||
}
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
var result []*OpenFlareAccessLogRegionCount
|
||||
for rows.Next() {
|
||||
var item OpenFlareAccessLogRegionCount
|
||||
if err := rows.Scan(&item.Region, &item.Count); err != nil {
|
||||
return nil, err
|
||||
result := make([]*OpenFlareAccessLogRegionCount, len(rows))
|
||||
for index, row := range rows {
|
||||
result[index] = &OpenFlareAccessLogRegionCount{
|
||||
Region: row.Region,
|
||||
Count: row.Count,
|
||||
}
|
||||
result = append(result, &item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) BucketAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogBucketAggregateRow, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
rows, err := analyticsrepo.BucketAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), bucketSeconds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildOpenFlareAccessLogFilterClause(filter)
|
||||
bucketExpr := openFlareAccessLogBucketEpochExpr(bucketSeconds)
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
%s AS bucket_epoch,
|
||||
count() AS request_count,
|
||||
countIf(status_code < 400) AS success_count,
|
||||
countIf(status_code >= 400 AND status_code < 500) AS client_error_count,
|
||||
countIf(status_code >= 500) AS server_error_count
|
||||
FROM %s
|
||||
WHERE %s
|
||||
GROUP BY bucket_epoch`, bucketExpr, openFlareAccessLogTable, clause)
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
var result []openFlareAccessLogBucketAggregateRow
|
||||
for rows.Next() {
|
||||
var item openFlareAccessLogBucketAggregateRow
|
||||
if err := rows.Scan(&item.BucketEpoch, &item.RequestCount, &item.SuccessCount, &item.ClientErrorCount, &item.ServerErrorCount); err != nil {
|
||||
return nil, err
|
||||
result := make([]openFlareAccessLogBucketAggregateRow, len(rows))
|
||||
for index, row := range rows {
|
||||
result[index] = openFlareAccessLogBucketAggregateRow{
|
||||
BucketEpoch: row.BucketEpoch,
|
||||
RequestCount: row.RequestCount,
|
||||
SuccessCount: row.SuccessCount,
|
||||
ClientErrorCount: row.ClientErrorCount,
|
||||
ServerErrorCount: row.ServerErrorCount,
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) BucketDimensions(ctx context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
rows, err := analyticsrepo.BucketDimensionsNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), column, bucketSeconds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildOpenFlareAccessLogFilterClause(filter)
|
||||
bucketExpr := openFlareAccessLogBucketEpochExpr(bucketSeconds)
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
%s AS bucket_epoch,
|
||||
trim(%s) AS value
|
||||
FROM %s
|
||||
WHERE %s AND trim(%s) != ''
|
||||
GROUP BY bucket_epoch, trim(%s)`, bucketExpr, column, openFlareAccessLogTable, clause, column, column)
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
var result []openFlareAccessLogBucketDimensionRow
|
||||
for rows.Next() {
|
||||
var item openFlareAccessLogBucketDimensionRow
|
||||
if err := rows.Scan(&item.BucketEpoch, &item.Value); err != nil {
|
||||
return nil, err
|
||||
result := make([]openFlareAccessLogBucketDimensionRow, len(rows))
|
||||
for index, row := range rows {
|
||||
result[index] = openFlareAccessLogBucketDimensionRow{
|
||||
BucketEpoch: row.BucketEpoch,
|
||||
Value: row.Value,
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) IPAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]openFlareAccessLogIPAggregateRow, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
rows, err := analyticsrepo.IPAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), exactRemoteAddr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildOpenFlareAccessLogFilterClause(filter)
|
||||
queryClause := clause
|
||||
queryArgs := append([]any{}, args...)
|
||||
if exactRemoteAddr {
|
||||
trimmed := strings.TrimSpace(filter.RemoteAddr)
|
||||
if trimmed == "" {
|
||||
return []openFlareAccessLogIPAggregateRow{}, nil
|
||||
result := make([]openFlareAccessLogIPAggregateRow, len(rows))
|
||||
for index, row := range rows {
|
||||
result[index] = openFlareAccessLogIPAggregateRow{
|
||||
RemoteAddr: row.RemoteAddr,
|
||||
RequestCount: row.RequestCount,
|
||||
SuccessCount: row.SuccessCount,
|
||||
ClientErrorCount: row.ClientErrorCount,
|
||||
ServerErrorCount: row.ServerErrorCount,
|
||||
LastSeenEpoch: row.LastSeenEpoch,
|
||||
}
|
||||
queryClause = combineOpenFlareAccessLogSQLClauses(queryClause, "trim(remote_addr) = ?")
|
||||
queryArgs = append(queryArgs, trimmed)
|
||||
}
|
||||
lastSeenExpr := openFlareAccessLogEpochExpr()
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
trim(remote_addr) AS remote_addr,
|
||||
count() AS request_count,
|
||||
countIf(status_code < 400) AS success_count,
|
||||
countIf(status_code >= 400 AND status_code < 500) AS client_error_count,
|
||||
countIf(status_code >= 500) AS server_error_count,
|
||||
max(%s) AS last_seen_epoch
|
||||
FROM %s
|
||||
WHERE %s AND trim(remote_addr) != ''
|
||||
GROUP BY trim(remote_addr)`, lastSeenExpr, openFlareAccessLogTable, queryClause)
|
||||
rows, err := conn.Query(ctx, sql, queryArgs...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
var result []openFlareAccessLogIPAggregateRow
|
||||
for rows.Next() {
|
||||
var item openFlareAccessLogIPAggregateRow
|
||||
if err := rows.Scan(&item.RemoteAddr, &item.RequestCount, &item.SuccessCount, &item.ClientErrorCount, &item.ServerErrorCount, &item.LastSeenEpoch); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) IPSummaries(ctx context.Context, filter OpenFlareAccessLogQuery, recentSince time.Time) ([]openFlareAccessLogIPSummaryRow, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
rows, err := analyticsrepo.IPSummariesNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), recentSince)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildOpenFlareAccessLogFilterClause(filter)
|
||||
lastSeenExpr := openFlareAccessLogEpochExpr()
|
||||
recentClause := "0"
|
||||
queryArgs := make([]any, 0, len(args)+1)
|
||||
if !recentSince.IsZero() {
|
||||
recentClause = "if(logged_at >= ?, 1, 0)"
|
||||
queryArgs = append(queryArgs, recentSince)
|
||||
}
|
||||
queryArgs = append(queryArgs, args...)
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
trim(remote_addr) AS remote_addr,
|
||||
count() AS total_requests,
|
||||
sum(%s) AS recent_requests,
|
||||
max(%s) AS last_seen_epoch
|
||||
FROM %s
|
||||
WHERE %s AND trim(remote_addr) != ''
|
||||
GROUP BY trim(remote_addr)`, recentClause, lastSeenExpr, openFlareAccessLogTable, clause)
|
||||
rows, err := conn.Query(ctx, sql, queryArgs...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
var result []openFlareAccessLogIPSummaryRow
|
||||
for rows.Next() {
|
||||
var item openFlareAccessLogIPSummaryRow
|
||||
if err := rows.Scan(&item.RemoteAddr, &item.TotalRequests, &item.RecentRequests, &item.LastSeenEpoch); err != nil {
|
||||
return nil, err
|
||||
result := make([]openFlareAccessLogIPSummaryRow, len(rows))
|
||||
for index, row := range rows {
|
||||
result[index] = openFlareAccessLogIPSummaryRow{
|
||||
RemoteAddr: row.RemoteAddr,
|
||||
TotalRequests: row.TotalRequests,
|
||||
RecentRequests: row.RecentRequests,
|
||||
LastSeenEpoch: row.LastSeenEpoch,
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) IPTrend(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogIPTrendRow, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
rows, err := analyticsrepo.IPTrendNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), bucketSeconds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildOpenFlareAccessLogFilterClause(filter)
|
||||
bucketExpr := openFlareAccessLogBucketEpochExpr(bucketSeconds)
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
%s AS bucket_epoch,
|
||||
count() AS request_count
|
||||
FROM %s
|
||||
WHERE %s
|
||||
GROUP BY bucket_epoch
|
||||
ORDER BY bucket_epoch ASC`, bucketExpr, openFlareAccessLogTable, clause)
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
var result []openFlareAccessLogIPTrendRow
|
||||
for rows.Next() {
|
||||
var item openFlareAccessLogIPTrendRow
|
||||
if err := rows.Scan(&item.BucketEpoch, &item.RequestCount); err != nil {
|
||||
return nil, err
|
||||
result := make([]openFlareAccessLogIPTrendRow, len(rows))
|
||||
for index, row := range rows {
|
||||
result[index] = openFlareAccessLogIPTrendRow{
|
||||
BucketEpoch: row.BucketEpoch,
|
||||
RequestCount: row.RequestCount,
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s clickhouseAccessLogStore) DeleteAll(ctx context.Context) (int64, error) {
|
||||
return s.deleteWithCount(ctx, "SELECT count() FROM "+openFlareAccessLogTable, nil, "ALTER TABLE "+openFlareAccessLogTable+" DELETE WHERE 1")
|
||||
func (clickhouseAccessLogStore) DeleteAll(ctx context.Context) (int64, error) {
|
||||
return analyticsrepo.DeleteAllNodeAccessLogs(ctx)
|
||||
}
|
||||
|
||||
func (s clickhouseAccessLogStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
return s.deleteWithCount(
|
||||
ctx,
|
||||
fmt.Sprintf("SELECT count() FROM %s WHERE logged_at < ?", openFlareAccessLogTable),
|
||||
[]any{cutoff.UTC()},
|
||||
fmt.Sprintf("ALTER TABLE %s DELETE WHERE logged_at < ?", openFlareAccessLogTable),
|
||||
cutoff.UTC(),
|
||||
)
|
||||
func (clickhouseAccessLogStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
return analyticsrepo.DeleteNodeAccessLogsBefore(ctx, cutoff)
|
||||
}
|
||||
|
||||
func (s clickhouseAccessLogStore) DeleteByNodeBefore(ctx context.Context, nodeID string, before time.Time) (int64, error) {
|
||||
return s.deleteWithCount(
|
||||
ctx,
|
||||
fmt.Sprintf("SELECT count() FROM %s WHERE node_id = ? AND logged_at < ?", openFlareAccessLogTable),
|
||||
[]any{nodeID, before.UTC()},
|
||||
fmt.Sprintf("ALTER TABLE %s DELETE WHERE node_id = ? AND logged_at < ?", openFlareAccessLogTable),
|
||||
nodeID, before.UTC(),
|
||||
)
|
||||
func (clickhouseAccessLogStore) DeleteByNodeBefore(ctx context.Context, nodeID string, before time.Time) (int64, error) {
|
||||
return analyticsrepo.DeleteNodeAccessLogsByNodeBefore(ctx, nodeID, before)
|
||||
}
|
||||
|
||||
func (clickhouseAccessLogStore) deleteWithCount(ctx context.Context, countSQL string, countArgs []any, deleteSQL string, deleteArgs ...any) (int64, error) {
|
||||
conn, err := clickhouseAccessLogStore{}.conn()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
func toNodeAccessLogFilter(query OpenFlareAccessLogQuery) analyticsrepo.NodeAccessLogFilter {
|
||||
return analyticsrepo.NodeAccessLogFilter{
|
||||
NodeID: query.NodeID,
|
||||
RemoteAddr: query.RemoteAddr,
|
||||
Host: query.Host,
|
||||
Path: query.Path,
|
||||
Since: query.Since,
|
||||
Until: query.Until,
|
||||
Page: query.Page,
|
||||
PageSize: query.PageSize,
|
||||
SortBy: query.SortBy,
|
||||
SortOrder: query.SortOrder,
|
||||
}
|
||||
var count int64
|
||||
if err := conn.QueryRow(ctx, countSQL, countArgs...).Scan(&count); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if count == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
if err := conn.Exec(ctx, deleteSQL, deleteArgs...); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
func openFlareAccessLogBucketEpochExpr(bucketSeconds int64) string {
|
||||
return fmt.Sprintf("toInt64(intDiv(toUnixTimestamp(logged_at), %d) * %d)", bucketSeconds, bucketSeconds)
|
||||
func toAnalyticsNodeAccessLog(record *OpenFlareAccessLog) analyticsmodel.NodeAccessLog {
|
||||
return analyticsmodel.NodeAccessLog{
|
||||
ID: uint64(record.ID),
|
||||
NodeID: record.NodeID,
|
||||
LoggedAt: record.LoggedAt,
|
||||
RemoteAddr: record.RemoteAddr,
|
||||
Region: record.Region,
|
||||
Host: record.Host,
|
||||
Path: record.Path,
|
||||
StatusCode: int32(record.StatusCode),
|
||||
CreatedAt: record.CreatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
func openFlareAccessLogEpochExpr() string {
|
||||
return "toInt64(toUnixTimestamp(logged_at))"
|
||||
func fromAnalyticsNodeAccessLogs(rows []analyticsmodel.NodeAccessLog) []*OpenFlareAccessLog {
|
||||
result := make([]*OpenFlareAccessLog, len(rows))
|
||||
for index, row := range rows {
|
||||
result[index] = &OpenFlareAccessLog{
|
||||
ID: uint(row.ID),
|
||||
NodeID: row.NodeID,
|
||||
LoggedAt: row.LoggedAt,
|
||||
RemoteAddr: row.RemoteAddr,
|
||||
Region: row.Region,
|
||||
Host: row.Host,
|
||||
Path: row.Path,
|
||||
StatusCode: int(row.StatusCode),
|
||||
CreatedAt: row.CreatedAt,
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -55,7 +55,8 @@ func (OpenFlareRequestReport) TableName() string {
|
||||
return "of_node_request_reports"
|
||||
}
|
||||
|
||||
// OpenFlareAccessLog stores a single access log row in ClickHouse (database: openflare).
|
||||
// OpenFlareAccessLog stores a single access log row in ClickHouse (database: openflare, table: of_node_access_logs).
|
||||
// ClickHouse DDL is managed by goose; reads/writes go through internal/repository/analytics.
|
||||
type OpenFlareAccessLog struct {
|
||||
ID uint `json:"id" gorm:"primaryKey;autoIncrement"`
|
||||
NodeID string `json:"node_id" gorm:"index;size:64;not null"`
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
)
|
||||
|
||||
// NodeAccessLogRegionCount aggregates access log regions.
|
||||
type NodeAccessLogRegionCount struct {
|
||||
Region string
|
||||
Count int64
|
||||
}
|
||||
|
||||
func nodeAccessLogConn() (driver.Conn, error) {
|
||||
if db.ChConn == nil {
|
||||
return nil, fmt.Errorf("clickhouse connection is not initialized")
|
||||
}
|
||||
return db.ChConn, nil
|
||||
}
|
||||
|
||||
// ListNodeAccessLogs returns access logs matching filter.
|
||||
func ListNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter) ([]analyticsmodel.NodeAccessLog, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildNodeAccessLogFilterClause(filter)
|
||||
tableName := nodeAccessLogTableName()
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT id, node_id, logged_at, remote_addr, region, host, path, status_code, created_at
|
||||
FROM %s
|
||||
WHERE %s
|
||||
ORDER BY %s`, tableName, clause, nodeAccessLogOrderClause(filter.SortBy, filter.SortOrder))
|
||||
if filter.PageSize > 0 {
|
||||
if filter.Page < 0 {
|
||||
filter.Page = 0
|
||||
}
|
||||
sql += " LIMIT ? OFFSET ?"
|
||||
args = append(args, filter.PageSize, filter.Page*filter.PageSize)
|
||||
}
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list node access logs: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
return scanNodeAccessLogRows(rows)
|
||||
}
|
||||
|
||||
func scanNodeAccessLogRows(rows driver.Rows) ([]analyticsmodel.NodeAccessLog, error) {
|
||||
var result []analyticsmodel.NodeAccessLog
|
||||
for rows.Next() {
|
||||
var item analyticsmodel.NodeAccessLog
|
||||
if err := rows.Scan(
|
||||
&item.ID,
|
||||
&item.NodeID,
|
||||
&item.LoggedAt,
|
||||
&item.RemoteAddr,
|
||||
&item.Region,
|
||||
&item.Host,
|
||||
&item.Path,
|
||||
&item.StatusCode,
|
||||
&item.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan node access log row: %w", err)
|
||||
}
|
||||
item.LoggedAt = item.LoggedAt.UTC()
|
||||
item.CreatedAt = item.CreatedAt.UTC()
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// CountNodeAccessLogs returns total records and distinct IPs matching filter.
|
||||
func CountNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter) (int64, int64, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
clause, args := buildNodeAccessLogFilterClause(filter)
|
||||
tableName := nodeAccessLogTableName()
|
||||
|
||||
var totalRecords int64
|
||||
countSQL := fmt.Sprintf("SELECT count() FROM %s WHERE %s", tableName, clause)
|
||||
if err := conn.QueryRow(ctx, countSQL, args...).Scan(&totalRecords); err != nil {
|
||||
return 0, 0, fmt.Errorf("count node access logs: %w", err)
|
||||
}
|
||||
|
||||
ipSQL := fmt.Sprintf(`
|
||||
SELECT count() FROM (
|
||||
SELECT trim(remote_addr) AS remote_addr
|
||||
FROM %s
|
||||
WHERE %s AND remote_addr != ''
|
||||
GROUP BY trim(remote_addr)
|
||||
)`, tableName, clause)
|
||||
var totalIPs int64
|
||||
if err := conn.QueryRow(ctx, ipSQL, args...).Scan(&totalIPs); err != nil {
|
||||
return 0, 0, fmt.Errorf("count node access log ips: %w", err)
|
||||
}
|
||||
return totalRecords, totalIPs, nil
|
||||
}
|
||||
|
||||
// RegionCountsNodeAccessLogs returns region counts for a node since a time.
|
||||
func RegionCountsNodeAccessLogs(ctx context.Context, nodeID string, since time.Time, limit int) ([]NodeAccessLogRegionCount, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
filter := NodeAccessLogFilter{NodeID: nodeID, Since: since}
|
||||
clause, args := buildNodeAccessLogFilterClause(filter)
|
||||
tableName := nodeAccessLogTableName()
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT trim(region) AS region, count() AS count
|
||||
FROM %s
|
||||
WHERE %s AND trim(region) != ''
|
||||
GROUP BY trim(region)
|
||||
ORDER BY count DESC, region ASC`, tableName, clause)
|
||||
if limit > 0 {
|
||||
sql += " LIMIT ?"
|
||||
args = append(args, limit)
|
||||
}
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("region counts node access logs: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var result []NodeAccessLogRegionCount
|
||||
for rows.Next() {
|
||||
var item NodeAccessLogRegionCount
|
||||
if err := rows.Scan(&item.Region, &item.Count); err != nil {
|
||||
return nil, fmt.Errorf("scan region count row: %w", err)
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DeleteAllNodeAccessLogs deletes all node access logs.
|
||||
func DeleteAllNodeAccessLogs(ctx context.Context) (int64, error) {
|
||||
tableName := nodeAccessLogTableName()
|
||||
return deleteNodeAccessLogsWithCount(ctx, "SELECT count() FROM "+tableName, nil, "ALTER TABLE "+tableName+" DELETE WHERE 1")
|
||||
}
|
||||
|
||||
// DeleteNodeAccessLogsBefore deletes logs older than cutoff.
|
||||
func DeleteNodeAccessLogsBefore(ctx context.Context, cutoff time.Time) (int64, error) {
|
||||
tableName := nodeAccessLogTableName()
|
||||
cutoff = cutoff.UTC()
|
||||
return deleteNodeAccessLogsWithCount(
|
||||
ctx,
|
||||
fmt.Sprintf("SELECT count() FROM %s WHERE logged_at < ?", tableName),
|
||||
[]any{cutoff},
|
||||
fmt.Sprintf("ALTER TABLE %s DELETE WHERE logged_at < ?", tableName),
|
||||
cutoff,
|
||||
)
|
||||
}
|
||||
|
||||
// DeleteNodeAccessLogsByNodeBefore deletes logs for a node older than cutoff.
|
||||
func DeleteNodeAccessLogsByNodeBefore(ctx context.Context, nodeID string, before time.Time) (int64, error) {
|
||||
tableName := nodeAccessLogTableName()
|
||||
before = before.UTC()
|
||||
return deleteNodeAccessLogsWithCount(
|
||||
ctx,
|
||||
fmt.Sprintf("SELECT count() FROM %s WHERE node_id = ? AND logged_at < ?", tableName),
|
||||
[]any{nodeID, before},
|
||||
fmt.Sprintf("ALTER TABLE %s DELETE WHERE node_id = ? AND logged_at < ?", tableName),
|
||||
nodeID, before,
|
||||
)
|
||||
}
|
||||
|
||||
func deleteNodeAccessLogsWithCount(ctx context.Context, countSQL string, countArgs []any, deleteSQL string, deleteArgs ...any) (int64, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var count int64
|
||||
if err := conn.QueryRow(ctx, countSQL, countArgs...).Scan(&count); err != nil {
|
||||
return 0, fmt.Errorf("count node access logs for delete: %w", err)
|
||||
}
|
||||
if count == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
if err := conn.Exec(ctx, deleteSQL, deleteArgs...); err != nil {
|
||||
return 0, fmt.Errorf("delete node access logs: %w", err)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// NodeAccessLogFilter scopes ClickHouse node access log queries.
|
||||
type NodeAccessLogFilter struct {
|
||||
NodeID string
|
||||
RemoteAddr string
|
||||
Host string
|
||||
Path string
|
||||
Since time.Time
|
||||
Until time.Time
|
||||
Page int
|
||||
PageSize int
|
||||
SortBy string
|
||||
SortOrder string
|
||||
}
|
||||
|
||||
func buildNodeAccessLogFilterClause(filter NodeAccessLogFilter) (string, []any) {
|
||||
parts := make([]string, 0, 6)
|
||||
args := make([]any, 0, 6)
|
||||
if trimmed := strings.TrimSpace(filter.NodeID); trimmed != "" {
|
||||
parts = append(parts, "node_id = ?")
|
||||
args = append(args, trimmed)
|
||||
}
|
||||
if trimmed := strings.TrimSpace(filter.RemoteAddr); trimmed != "" {
|
||||
parts = append(parts, "remote_addr LIKE ?")
|
||||
args = append(args, trimmed+"%")
|
||||
}
|
||||
if trimmed := strings.TrimSpace(filter.Host); trimmed != "" {
|
||||
parts = append(parts, "host LIKE ?")
|
||||
args = append(args, trimmed+"%")
|
||||
}
|
||||
if trimmed := strings.TrimSpace(filter.Path); trimmed != "" {
|
||||
parts = append(parts, "path LIKE ?")
|
||||
args = append(args, trimmed+"%")
|
||||
}
|
||||
if !filter.Since.IsZero() {
|
||||
parts = append(parts, "logged_at >= ?")
|
||||
args = append(args, filter.Since.UTC())
|
||||
}
|
||||
if !filter.Until.IsZero() {
|
||||
parts = append(parts, "logged_at < ?")
|
||||
args = append(args, filter.Until.UTC())
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return "1", nil
|
||||
}
|
||||
return strings.Join(parts, " AND "), args
|
||||
}
|
||||
|
||||
func combineNodeAccessLogSQLClauses(left string, right string) string {
|
||||
if strings.TrimSpace(left) == "" || left == "TRUE" || left == "1" {
|
||||
return right
|
||||
}
|
||||
return left + " AND " + right
|
||||
}
|
||||
|
||||
func nodeAccessLogOrderClause(sortBy string, sortOrder string) string {
|
||||
direction := "DESC"
|
||||
if normalizeNodeAccessLogSortOrder(sortOrder) == "asc" {
|
||||
direction = "ASC"
|
||||
}
|
||||
column := "logged_at"
|
||||
switch strings.TrimSpace(sortBy) {
|
||||
case "status_code":
|
||||
column = "status_code"
|
||||
case "remote_addr":
|
||||
column = "remote_addr"
|
||||
case "host":
|
||||
column = "host"
|
||||
case "path":
|
||||
column = "path"
|
||||
}
|
||||
if column == "logged_at" {
|
||||
return column + " " + direction + ", id " + direction
|
||||
}
|
||||
return column + " " + direction + ", logged_at " + direction + ", id " + direction
|
||||
}
|
||||
|
||||
func normalizeNodeAccessLogSortOrder(sortOrder string) string {
|
||||
if strings.EqualFold(strings.TrimSpace(sortOrder), "asc") {
|
||||
return "asc"
|
||||
}
|
||||
return "desc"
|
||||
}
|
||||
|
||||
func nodeAccessLogBucketEpochExpr(bucketSeconds int64) string {
|
||||
return fmt.Sprintf("toInt64(intDiv(toUnixTimestamp(logged_at), %d) * %d)", bucketSeconds, bucketSeconds)
|
||||
}
|
||||
|
||||
func nodeAccessLogEpochExpr() string {
|
||||
return "toInt64(toUnixTimestamp(logged_at))"
|
||||
}
|
||||
|
||||
func nodeAccessLogTableName() string {
|
||||
return "of_node_access_logs"
|
||||
}
|
||||
@@ -0,0 +1,242 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// NodeAccessLogBucketAggregate is a folded bucket aggregate row.
|
||||
type NodeAccessLogBucketAggregate struct {
|
||||
BucketEpoch int64
|
||||
RequestCount int64
|
||||
SuccessCount int64
|
||||
ClientErrorCount int64
|
||||
ServerErrorCount int64
|
||||
}
|
||||
|
||||
// NodeAccessLogBucketDimension is a bucket dimension value.
|
||||
type NodeAccessLogBucketDimension struct {
|
||||
BucketEpoch int64
|
||||
Value string
|
||||
}
|
||||
|
||||
// NodeAccessLogIPAggregate is an IP aggregate row.
|
||||
type NodeAccessLogIPAggregate struct {
|
||||
RemoteAddr string
|
||||
RequestCount int64
|
||||
SuccessCount int64
|
||||
ClientErrorCount int64
|
||||
ServerErrorCount int64
|
||||
LastSeenEpoch int64
|
||||
}
|
||||
|
||||
// NodeAccessLogIPSummary is an IP summary row.
|
||||
type NodeAccessLogIPSummary struct {
|
||||
RemoteAddr string
|
||||
TotalRequests int64
|
||||
RecentRequests int64
|
||||
LastSeenEpoch int64
|
||||
}
|
||||
|
||||
// NodeAccessLogIPTrend is an IP trend bucket row.
|
||||
type NodeAccessLogIPTrend struct {
|
||||
BucketEpoch int64
|
||||
RequestCount int64
|
||||
}
|
||||
|
||||
// BucketAggregatesNodeAccessLogs returns folded bucket aggregates.
|
||||
func BucketAggregatesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, bucketSeconds int64) ([]NodeAccessLogBucketAggregate, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildNodeAccessLogFilterClause(filter)
|
||||
bucketExpr := nodeAccessLogBucketEpochExpr(bucketSeconds)
|
||||
tableName := nodeAccessLogTableName()
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
%s AS bucket_epoch,
|
||||
count() AS request_count,
|
||||
countIf(status_code < 400) AS success_count,
|
||||
countIf(status_code >= 400 AND status_code < 500) AS client_error_count,
|
||||
countIf(status_code >= 500) AS server_error_count
|
||||
FROM %s
|
||||
WHERE %s
|
||||
GROUP BY bucket_epoch`, bucketExpr, tableName, clause)
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("bucket aggregates node access logs: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var result []NodeAccessLogBucketAggregate
|
||||
for rows.Next() {
|
||||
var item NodeAccessLogBucketAggregate
|
||||
if err := rows.Scan(&item.BucketEpoch, &item.RequestCount, &item.SuccessCount, &item.ClientErrorCount, &item.ServerErrorCount); err != nil {
|
||||
return nil, fmt.Errorf("scan bucket aggregate row: %w", err)
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// BucketDimensionsNodeAccessLogs returns bucket dimension values.
|
||||
func BucketDimensionsNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, column string, bucketSeconds int64) ([]NodeAccessLogBucketDimension, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildNodeAccessLogFilterClause(filter)
|
||||
bucketExpr := nodeAccessLogBucketEpochExpr(bucketSeconds)
|
||||
tableName := nodeAccessLogTableName()
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
%s AS bucket_epoch,
|
||||
trim(%s) AS value
|
||||
FROM %s
|
||||
WHERE %s AND trim(%s) != ''
|
||||
GROUP BY bucket_epoch, trim(%s)`, bucketExpr, column, tableName, clause, column, column)
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("bucket dimensions node access logs: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var result []NodeAccessLogBucketDimension
|
||||
for rows.Next() {
|
||||
var item NodeAccessLogBucketDimension
|
||||
if err := rows.Scan(&item.BucketEpoch, &item.Value); err != nil {
|
||||
return nil, fmt.Errorf("scan bucket dimension row: %w", err)
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// IPAggregatesNodeAccessLogs returns IP aggregate rows.
|
||||
func IPAggregatesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, exactRemoteAddr bool) ([]NodeAccessLogIPAggregate, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildNodeAccessLogFilterClause(filter)
|
||||
queryClause := clause
|
||||
queryArgs := append([]any{}, args...)
|
||||
if exactRemoteAddr {
|
||||
trimmed := strings.TrimSpace(filter.RemoteAddr)
|
||||
if trimmed == "" {
|
||||
return []NodeAccessLogIPAggregate{}, nil
|
||||
}
|
||||
queryClause = combineNodeAccessLogSQLClauses(queryClause, "trim(remote_addr) = ?")
|
||||
queryArgs = append(queryArgs, trimmed)
|
||||
}
|
||||
lastSeenExpr := nodeAccessLogEpochExpr()
|
||||
tableName := nodeAccessLogTableName()
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
trim(remote_addr) AS remote_addr,
|
||||
count() AS request_count,
|
||||
countIf(status_code < 400) AS success_count,
|
||||
countIf(status_code >= 400 AND status_code < 500) AS client_error_count,
|
||||
countIf(status_code >= 500) AS server_error_count,
|
||||
max(%s) AS last_seen_epoch
|
||||
FROM %s
|
||||
WHERE %s AND trim(remote_addr) != ''
|
||||
GROUP BY trim(remote_addr)`, lastSeenExpr, tableName, queryClause)
|
||||
rows, err := conn.Query(ctx, sql, queryArgs...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ip aggregates node access logs: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var result []NodeAccessLogIPAggregate
|
||||
for rows.Next() {
|
||||
var item NodeAccessLogIPAggregate
|
||||
if err := rows.Scan(&item.RemoteAddr, &item.RequestCount, &item.SuccessCount, &item.ClientErrorCount, &item.ServerErrorCount, &item.LastSeenEpoch); err != nil {
|
||||
return nil, fmt.Errorf("scan ip aggregate row: %w", err)
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// IPSummariesNodeAccessLogs returns IP summary rows.
|
||||
func IPSummariesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, recentSince time.Time) ([]NodeAccessLogIPSummary, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildNodeAccessLogFilterClause(filter)
|
||||
lastSeenExpr := nodeAccessLogEpochExpr()
|
||||
recentClause := "0"
|
||||
queryArgs := make([]any, 0, len(args)+1)
|
||||
if !recentSince.IsZero() {
|
||||
recentClause = "if(logged_at >= ?, 1, 0)"
|
||||
queryArgs = append(queryArgs, recentSince)
|
||||
}
|
||||
queryArgs = append(queryArgs, args...)
|
||||
tableName := nodeAccessLogTableName()
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
trim(remote_addr) AS remote_addr,
|
||||
count() AS total_requests,
|
||||
sum(%s) AS recent_requests,
|
||||
max(%s) AS last_seen_epoch
|
||||
FROM %s
|
||||
WHERE %s AND trim(remote_addr) != ''
|
||||
GROUP BY trim(remote_addr)`, recentClause, lastSeenExpr, tableName, clause)
|
||||
rows, err := conn.Query(ctx, sql, queryArgs...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ip summaries node access logs: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var result []NodeAccessLogIPSummary
|
||||
for rows.Next() {
|
||||
var item NodeAccessLogIPSummary
|
||||
if err := rows.Scan(&item.RemoteAddr, &item.TotalRequests, &item.RecentRequests, &item.LastSeenEpoch); err != nil {
|
||||
return nil, fmt.Errorf("scan ip summary row: %w", err)
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// IPTrendNodeAccessLogs returns IP trend bucket rows.
|
||||
func IPTrendNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, bucketSeconds int64) ([]NodeAccessLogIPTrend, error) {
|
||||
conn, err := nodeAccessLogConn()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clause, args := buildNodeAccessLogFilterClause(filter)
|
||||
bucketExpr := nodeAccessLogBucketEpochExpr(bucketSeconds)
|
||||
tableName := nodeAccessLogTableName()
|
||||
sql := fmt.Sprintf(`
|
||||
SELECT
|
||||
%s AS bucket_epoch,
|
||||
count() AS request_count
|
||||
FROM %s
|
||||
WHERE %s
|
||||
GROUP BY bucket_epoch
|
||||
ORDER BY bucket_epoch ASC`, bucketExpr, tableName, clause)
|
||||
rows, err := conn.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ip trend node access logs: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var result []NodeAccessLogIPTrend
|
||||
for rows.Next() {
|
||||
var item NodeAccessLogIPTrend
|
||||
if err := rows.Scan(&item.BucketEpoch, &item.RequestCount); err != nil {
|
||||
return nil, fmt.Errorf("scan ip trend row: %w", err)
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestBatchInsertNodeAccessLogs_Empty(t *testing.T) {
|
||||
err := BatchInsertNodeAccessLogs(context.Background(), nil)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestBatchInsertNodeAccessLogs_UsesModelBatchSQL(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
mockBatch := &mockBatch{}
|
||||
mockConn := &mockConn{
|
||||
batch: mockBatch,
|
||||
batchQuery: analyticsmodel.NodeAccessLog{}.BatchInsertSQL(),
|
||||
}
|
||||
db.SetChConnForTest(mockConn)
|
||||
t.Cleanup(func() { db.SetChConnForTest(nil) })
|
||||
|
||||
loggedAt := time.Now().UTC()
|
||||
err := BatchInsertNodeAccessLogs(ctx, []analyticsmodel.NodeAccessLog{
|
||||
{
|
||||
NodeID: "node-a",
|
||||
LoggedAt: loggedAt,
|
||||
RemoteAddr: "1.1.1.1",
|
||||
Region: "US",
|
||||
Host: "example.com",
|
||||
Path: "/alpha",
|
||||
StatusCode: 200,
|
||||
CreatedAt: loggedAt,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, mockConn.prepareCalled)
|
||||
assert.Equal(t, analyticsmodel.NodeAccessLog{}.BatchInsertSQL(), mockConn.preparedQuery)
|
||||
assert.True(t, mockBatch.sendCalled)
|
||||
require.Len(t, mockBatch.rows, 1)
|
||||
assert.Equal(t, "node-a", mockBatch.rows[0][1])
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package analytics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
"github.com/Rain-kl/Wavelet/internal/db/idgen"
|
||||
analyticsmodel "github.com/Rain-kl/Wavelet/internal/model/analytics"
|
||||
)
|
||||
|
||||
// BatchInsertNodeAccessLogs writes node access logs to ClickHouse using the native batch API.
|
||||
func BatchInsertNodeAccessLogs(ctx context.Context, logs []analyticsmodel.NodeAccessLog) error {
|
||||
if len(logs) == 0 {
|
||||
return nil
|
||||
}
|
||||
if db.ChConn == nil {
|
||||
return fmt.Errorf("clickhouse connection is not initialized")
|
||||
}
|
||||
|
||||
batch, err := db.ChConn.PrepareBatch(ctx, analyticsmodel.NodeAccessLog{}.BatchInsertSQL())
|
||||
if err != nil {
|
||||
return fmt.Errorf("prepare clickhouse batch: %w", err)
|
||||
}
|
||||
|
||||
now := time.Now().UTC()
|
||||
for _, logItem := range logs {
|
||||
id := logItem.ID
|
||||
if id == 0 {
|
||||
id = idgen.NextUint64ID()
|
||||
}
|
||||
createdAt := logItem.CreatedAt
|
||||
if createdAt.IsZero() {
|
||||
createdAt = now
|
||||
}
|
||||
if err := batch.Append(
|
||||
id,
|
||||
logItem.NodeID,
|
||||
logItem.LoggedAt.UTC(),
|
||||
logItem.RemoteAddr,
|
||||
logItem.Region,
|
||||
logItem.Host,
|
||||
logItem.Path,
|
||||
logItem.StatusCode,
|
||||
createdAt.UTC(),
|
||||
); err != nil {
|
||||
return fmt.Errorf("append node access log to batch: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := batch.Send(); err != nil {
|
||||
return fmt.Errorf("send clickhouse batch: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user