diff --git a/docs/changelog/index.md b/docs/changelog/index.md index bc312e8a..0572d4f7 100644 --- a/docs/changelog/index.md +++ b/docs/changelog/index.md @@ -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` 健康依赖。 ### 移除 diff --git a/openflare-server/internal/model/openflare_access_log.go b/openflare-server/internal/model/openflare_access_log.go index cea5bca5..ee93f442 100644 --- a/openflare-server/internal/model/openflare_access_log.go +++ b/openflare-server/internal/model/openflare_access_log.go @@ -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 { diff --git a/openflare-server/internal/model/openflare_access_log_store.go b/openflare-server/internal/model/openflare_access_log_store.go index cbe5d2cf..37532678 100644 --- a/openflare-server/internal/model/openflare_access_log_store.go +++ b/openflare-server/internal/model/openflare_access_log_store.go @@ -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 } \ No newline at end of file diff --git a/openflare-server/internal/model/openflare_observability.go b/openflare-server/internal/model/openflare_observability.go index 7963829a..d0d8125f 100644 --- a/openflare-server/internal/model/openflare_observability.go +++ b/openflare-server/internal/model/openflare_observability.go @@ -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"` diff --git a/openflare-server/internal/repository/analytics/node_access_log.go b/openflare-server/internal/repository/analytics/node_access_log.go new file mode 100644 index 00000000..bbc5380b --- /dev/null +++ b/openflare-server/internal/repository/analytics/node_access_log.go @@ -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 +} \ No newline at end of file diff --git a/openflare-server/internal/repository/analytics/node_access_log_delete.go b/openflare-server/internal/repository/analytics/node_access_log_delete.go new file mode 100644 index 00000000..40c3e7a8 --- /dev/null +++ b/openflare-server/internal/repository/analytics/node_access_log_delete.go @@ -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 +} \ No newline at end of file diff --git a/openflare-server/internal/repository/analytics/node_access_log_filter.go b/openflare-server/internal/repository/analytics/node_access_log_filter.go new file mode 100644 index 00000000..e3a240d9 --- /dev/null +++ b/openflare-server/internal/repository/analytics/node_access_log_filter.go @@ -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" +} \ No newline at end of file diff --git a/openflare-server/internal/repository/analytics/node_access_log_stats.go b/openflare-server/internal/repository/analytics/node_access_log_stats.go new file mode 100644 index 00000000..a43c9308 --- /dev/null +++ b/openflare-server/internal/repository/analytics/node_access_log_stats.go @@ -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 +} \ No newline at end of file diff --git a/openflare-server/internal/repository/analytics/node_access_log_test.go b/openflare-server/internal/repository/analytics/node_access_log_test.go new file mode 100644 index 00000000..6434490f --- /dev/null +++ b/openflare-server/internal/repository/analytics/node_access_log_test.go @@ -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]) +} \ No newline at end of file diff --git a/openflare-server/internal/repository/analytics/node_access_log_writer.go b/openflare-server/internal/repository/analytics/node_access_log_writer.go new file mode 100644 index 00000000..58fe6ef8 --- /dev/null +++ b/openflare-server/internal/repository/analytics/node_access_log_writer.go @@ -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 +} \ No newline at end of file