Files
OpenFlare/internal/repository/analytics/clickhouse_stats.go
T
ryan 58624db397 perf(clickhouse): Phase 2 legacy governance — TTL cleanup, unified pool, MV, ops API
- Replace retention ALTER DELETE with MATERIALIZE TTL; use TRUNCATE for delete-all
- Remove GORM ClickHouse pool; migrate user access log reads to ChConn
- Drop query-side trim(remote_addr); enable wait_for_async_insert=1
- Add of_node_traffic_hourly MV and dashboard traffic trend fallback
- Add GET /admin/status/clickhouse operational metrics endpoint
2026-07-02 15:47:38 +08:00

70 lines
2.2 KiB
Go

// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package analytics
import (
"context"
"fmt"
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/db"
)
// ClickHouseOperationalStats summarizes ClickHouse merge/mutation pressure.
type ClickHouseOperationalStats struct {
Database string `json:"database"`
ActiveParts int64 `json:"active_parts"`
TotalRows int64 `json:"total_rows"`
PendingMutations int64 `json:"pending_mutations"`
AsyncInsertQueue int64 `json:"async_insert_queue"`
AsyncInsertBytes int64 `json:"async_insert_bytes"`
}
// GetClickHouseOperationalStats returns operational metrics for the configured database.
func GetClickHouseOperationalStats(ctx context.Context) (*ClickHouseOperationalStats, error) {
if db.ChConn == nil {
return nil, fmt.Errorf("clickhouse native connection is not initialized")
}
database := config.Config.ClickHouse.Database
stats := &ClickHouseOperationalStats{Database: database}
partsSQL := `
SELECT
count() AS active_parts,
ifNull(sum(rows), 0) AS total_rows
FROM system.parts
WHERE active AND database = ?`
var activeParts, totalRows uint64
if err := db.ChConn.QueryRow(ctx, partsSQL, database).Scan(&activeParts, &totalRows); err != nil {
return nil, fmt.Errorf("query system.parts: %w", err)
}
stats.ActiveParts = safeInt64Count(activeParts)
stats.TotalRows = safeInt64Count(totalRows)
mutationsSQL := `
SELECT count()
FROM system.mutations
WHERE is_done = 0 AND database = ?`
if err := db.ChConn.QueryRow(ctx, mutationsSQL, database).Scan(&stats.PendingMutations); err != nil {
return nil, fmt.Errorf("query system.mutations: %w", err)
}
asyncSQL := `
SELECT
count() AS queue_entries,
ifNull(sum(bytes), 0) AS queue_bytes
FROM system.asynchronous_inserts
WHERE database = ?`
var queueEntries, queueBytes uint64
if err := db.ChConn.QueryRow(ctx, asyncSQL, database).Scan(&queueEntries, &queueBytes); err != nil {
// Older ClickHouse versions may not expose asynchronous_inserts; treat as optional.
stats.AsyncInsertQueue = 0
stats.AsyncInsertBytes = 0
} else {
stats.AsyncInsertQueue = safeInt64Count(queueEntries)
stats.AsyncInsertBytes = safeInt64Count(queueBytes)
}
return stats, nil
}