|
| 1 | +package dbmaint |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "strings" |
| 6 | + "time" |
| 7 | + |
| 8 | + logging "github.com/ipfs/go-log/v2" |
| 9 | + "github.com/yugabyte/pgx/v5" |
| 10 | + "golang.org/x/mod/semver" |
| 11 | + "golang.org/x/xerrors" |
| 12 | + |
| 13 | + curiobuild "github.com/filecoin-project/curio/build" |
| 14 | + "github.com/filecoin-project/curio/harmony/harmonydb" |
| 15 | + "github.com/filecoin-project/curio/harmony/harmonytask" |
| 16 | + "github.com/filecoin-project/curio/harmony/resources" |
| 17 | + "github.com/filecoin-project/curio/harmony/taskhelp" |
| 18 | + "github.com/filecoin-project/curio/tasks/tasknames" |
| 19 | +) |
| 20 | + |
| 21 | +var log = logging.Logger("dbanalyze") |
| 22 | + |
| 23 | +const ( |
| 24 | + analyzeInterval = 24 * time.Hour |
| 25 | + analyzeGrowthThreshold = 0.10 |
| 26 | + minChurnDelta = 100 |
| 27 | + perTableAnalyzeTimeout = 10 * time.Minute |
| 28 | + upgradeQuietWindow = time.Hour |
| 29 | +) |
| 30 | + |
| 31 | +type DBAnalyzeTask struct { |
| 32 | + db *harmonydb.DB |
| 33 | +} |
| 34 | + |
| 35 | +func NewDBAnalyzeTask(db *harmonydb.DB) *DBAnalyzeTask { |
| 36 | + return &DBAnalyzeTask{db: db} |
| 37 | +} |
| 38 | + |
| 39 | +func (d *DBAnalyzeTask) Do(ctx context.Context, taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error) { |
| 40 | + skip, err := d.skipForRollingUpgrade(ctx) |
| 41 | + if err != nil { |
| 42 | + return false, xerrors.Errorf("checking rolling upgrade: %w", err) |
| 43 | + } |
| 44 | + if skip { |
| 45 | + return true, nil |
| 46 | + } |
| 47 | + |
| 48 | + var rows []struct { |
| 49 | + TableName string `db:"table_name"` |
| 50 | + Churn int64 `db:"churn"` |
| 51 | + LiveRows int64 `db:"live_rows"` |
| 52 | + ChurnAtAnalyze *int64 `db:"churn_at_analyze"` |
| 53 | + } |
| 54 | + err = d.db.Select(ctx, &rows, ` |
| 55 | + SELECT s.relname AS table_name, |
| 56 | + s.n_tup_ins + s.n_tup_upd + s.n_tup_del AS churn, |
| 57 | + s.n_live_tup AS live_rows, |
| 58 | + a.churn_at_analyze |
| 59 | + FROM pg_stat_user_tables s |
| 60 | + LEFT JOIN table_analyze_state a ON a.table_name = s.relname |
| 61 | + WHERE s.schemaname = current_schema() |
| 62 | + ORDER BY s.relname`) |
| 63 | + if err != nil { |
| 64 | + return false, xerrors.Errorf("listing table churn: %w", err) |
| 65 | + } |
| 66 | + |
| 67 | + var schema string |
| 68 | + if err := d.db.QueryRow(ctx, `SELECT current_schema()`).Scan(&schema); err != nil { |
| 69 | + return false, xerrors.Errorf("getting current schema: %w", err) |
| 70 | + } |
| 71 | + |
| 72 | + analyzed := 0 |
| 73 | + for _, r := range rows { |
| 74 | + if !stillOwned() { |
| 75 | + return false, nil |
| 76 | + } |
| 77 | + if !shouldAnalyze(r.Churn, r.ChurnAtAnalyze) { |
| 78 | + continue |
| 79 | + } |
| 80 | + |
| 81 | + tctx, cancel := context.WithTimeout(ctx, perTableAnalyzeTimeout) |
| 82 | + _, err = harmonydb.AdminQuery(tctx, d.db, "ANALYZE "+pgx.Identifier{schema, r.TableName}.Sanitize()) |
| 83 | + cancel() |
| 84 | + if err != nil { |
| 85 | + log.Warnw("ANALYZE failed", "table", r.TableName, "error", err) |
| 86 | + continue |
| 87 | + } |
| 88 | + |
| 89 | + _, err = d.db.Exec(ctx, ` |
| 90 | + INSERT INTO table_analyze_state (table_name, churn_at_analyze, rows_at_analyze, last_analyzed_at, analyze_count) |
| 91 | + VALUES ($1, $2, $3, NOW(), 1) |
| 92 | + ON CONFLICT (table_name) DO UPDATE SET |
| 93 | + churn_at_analyze = EXCLUDED.churn_at_analyze, |
| 94 | + rows_at_analyze = EXCLUDED.rows_at_analyze, |
| 95 | + last_analyzed_at = NOW(), |
| 96 | + analyze_count = table_analyze_state.analyze_count + 1`, |
| 97 | + r.TableName, r.Churn, r.LiveRows) |
| 98 | + if err != nil { |
| 99 | + log.Warnw("failed to record analyze state", "table", r.TableName, "error", err) |
| 100 | + continue |
| 101 | + } |
| 102 | + |
| 103 | + analyzed++ |
| 104 | + log.Infow("analyzed table", "table", r.TableName, "churn", r.Churn, "live_rows", r.LiveRows) |
| 105 | + } |
| 106 | + |
| 107 | + if _, err := d.db.Exec(ctx, ` |
| 108 | + DELETE FROM table_analyze_state |
| 109 | + WHERE table_name NOT IN ( |
| 110 | + SELECT relname FROM pg_stat_user_tables WHERE schemaname = current_schema() |
| 111 | + )`); err != nil { |
| 112 | + log.Warnw("failed to prune dropped table analyze state", "error", err) |
| 113 | + } |
| 114 | + |
| 115 | + log.Infow("DB analyze pass complete", "task_id", taskID, "tables", len(rows), "analyzed", analyzed) |
| 116 | + return true, nil |
| 117 | +} |
| 118 | + |
| 119 | +func shouldAnalyze(churn int64, churnAtAnalyze *int64) bool { |
| 120 | + if churnAtAnalyze == nil { |
| 121 | + return true |
| 122 | + } |
| 123 | + prev := *churnAtAnalyze |
| 124 | + if churn < prev { |
| 125 | + // Stats counters were reset; re-baseline. |
| 126 | + return true |
| 127 | + } |
| 128 | + delta := churn - prev |
| 129 | + if delta < minChurnDelta { |
| 130 | + return false |
| 131 | + } |
| 132 | + return float64(churn) >= float64(prev)*(1+analyzeGrowthThreshold) |
| 133 | +} |
| 134 | + |
| 135 | +func (d *DBAnalyzeTask) skipForRollingUpgrade(ctx context.Context) (bool, error) { |
| 136 | + var peers []struct { |
| 137 | + Version string `db:"version"` |
| 138 | + StartupTime time.Time `db:"startup_time"` |
| 139 | + MachineID int64 `db:"machine_id"` |
| 140 | + } |
| 141 | + err := d.db.Select(ctx, &peers, ` |
| 142 | + SELECT machine_id, version, startup_time |
| 143 | + FROM harmony_machine_details |
| 144 | + WHERE version IS NOT NULL |
| 145 | + AND startup_time > NOW() - $1::interval`, upgradeQuietWindow.String()) |
| 146 | + if err != nil { |
| 147 | + return false, err |
| 148 | + } |
| 149 | + |
| 150 | + myLabel := curiobuild.ClusterMachineVersionLabel() |
| 151 | + mine := analyzeSemver(myLabel) |
| 152 | + for _, p := range peers { |
| 153 | + if semver.Compare(analyzeSemver(p.Version), mine) > 0 { |
| 154 | + log.Infow("skipping DB analyze pass, newer machine booted recently", |
| 155 | + "machine_id", p.MachineID, |
| 156 | + "their_version", p.Version, |
| 157 | + "my_version", myLabel, |
| 158 | + "startup_time", p.StartupTime) |
| 159 | + return true, nil |
| 160 | + } |
| 161 | + } |
| 162 | + return false, nil |
| 163 | +} |
| 164 | + |
| 165 | +// analyzeSemver turns a harmony_machine_details.version label such as |
| 166 | +// "1.28.3 abcdef1" or "1.28.3-rc1" into a semver.Compare-able string. |
| 167 | +func analyzeSemver(label string) string { |
| 168 | + v, _, _ := strings.Cut(label, " ") // drop the git hash suffix |
| 169 | + if !strings.HasPrefix(v, "v") { |
| 170 | + v = "v" + v |
| 171 | + } |
| 172 | + if !semver.IsValid(v) { |
| 173 | + return "" |
| 174 | + } |
| 175 | + return v |
| 176 | +} |
| 177 | + |
| 178 | +func (d *DBAnalyzeTask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) ([]harmonytask.TaskID, error) { |
| 179 | + return ids, nil |
| 180 | +} |
| 181 | + |
| 182 | +func (d *DBAnalyzeTask) TypeDetails() harmonytask.TaskTypeDetails { |
| 183 | + return harmonytask.TaskTypeDetails{ |
| 184 | + Max: taskhelp.Max(1), |
| 185 | + Name: tasknames.DBAnalyze, |
| 186 | + Cost: resources.Resources{ |
| 187 | + Cpu: 0, |
| 188 | + Gpu: 0, |
| 189 | + Ram: 64 << 20, |
| 190 | + }, |
| 191 | + MaxFailures: 3, |
| 192 | + IAmBored: harmonytask.SingletonTaskAdder(analyzeInterval, d), |
| 193 | + } |
| 194 | +} |
| 195 | + |
| 196 | +func (d *DBAnalyzeTask) Adder(taskFunc harmonytask.AddTaskFunc) {} |
| 197 | + |
| 198 | +var _ = harmonytask.Reg(&DBAnalyzeTask{}) |
| 199 | +var _ harmonytask.TaskInterface = &DBAnalyzeTask{} |
0 commit comments