From 28bf5f102e96bd4878e45334c0630b640bade7a8 Mon Sep 17 00:00:00 2001 From: liunyl Date: Sat, 29 Aug 2026 12:32:39 +0000 Subject: [PATCH 1/3] feat(metrics): add cache and checkpoint observability --- docs/07-durability-and-recovery.md | 43 +++- eloq_metrics/include/meter.h | 9 +- eloq_metrics/include/metrics.h | 18 +- eloq_metrics/include/metrics_registry_impl.h | 3 +- eloq_metrics/src/metrics_registry_impl.cpp | 6 +- eloq_metrics/src/prometheus_collector.cc | 6 +- eloq_metrics/tests/metrics_collector_test.cc | 72 +++++++ tx_service/include/checkpoint_metrics_state.h | 180 ++++++++++++++++ tx_service/include/checkpointer.h | 73 +++---- tx_service/include/data_sync_task.h | 94 +++++++- tx_service/include/tx_service_metrics.h | 20 +- tx_service/src/cc/cc_shard.cpp | 11 + tx_service/src/cc/local_cc_shards.cpp | 157 ++++++++++---- tx_service/src/checkpointer.cpp | 203 ++++++++++++++++-- tx_service/src/data_sync_task.cpp | 105 ++++++++- tx_service/src/fault/cc_node.cpp | 29 +++ tx_service/src/remote/cc_node_service.cpp | 3 +- tx_service/src/store/snapshot_manager.cpp | 6 +- tx_service/src/tx_index_operation.cpp | 13 +- tx_service/tests/CMakeLists.txt | 1 + .../tests/CheckpointMetricsState-Test.cpp | 66 ++++++ 21 files changed, 977 insertions(+), 141 deletions(-) create mode 100644 tx_service/include/checkpoint_metrics_state.h create mode 100644 tx_service/tests/CheckpointMetricsState-Test.cpp diff --git a/docs/07-durability-and-recovery.md b/docs/07-durability-and-recovery.md index 0d9e38b39..dc702fcc0 100644 --- a/docs/07-durability-and-recovery.md +++ b/docs/07-durability-and-recovery.md @@ -97,10 +97,10 @@ Notification sources: tx processors that find nothing clean to evict (`CcShard:: For each local node group (`checkpointer.cpp:141-431`): 1. Special cases first: a **candidate standby** doesn't checkpoint — it (re-)requests a storage snapshot from the primary via `RequestStorageSnapshotSync` (`:147-200`, §7); a synced **standby on shared storage** skips entirely (`:203-213`); a standby on private storage checkpoints its own kv store. -2. Skip the ng unless leader/candidate-leader (term from `Sharder`); compute `ckpt_ts` (§3.2); skip if `ckpt_ts <= GetNodeGroupCkptTs(ng)`. Consecutive skips are counted per ng: after `ckpt_stall_warn_rounds` of them (default 3, 0 disables; runtime-mutable) a rate-limited (60 s) `WARNING` names the transaction pinning the timestamp — `CkptTsCc::GetPinningTx()`, filled by `ActiveTxMinTs` with the tx whose `wlock_ts_` produced the minimum — plus its core and write-lock age, so a wedged or leaked write-lock holder is diagnosable from the log alone (`Checkpointer::WarnIfCkptStalled`). The same count is exported as the `checkpoint_stall_rounds` gauge (per `ng_id`, node meter) by `Checkpointer::CollectCkptStallMetric`, called on every round: from `WarnIfCkptStalled` before `ckpt_stall_warn_rounds` and the 60 s log rate limit are consulted, so neither can suppress it, and with 0 from the advancing path in `Ckpt()`, so the gauge falls back to 0 instead of flatlining at its last stalled value. The stall is therefore alertable without scraping logs. Note that a stall returns before any data sync task is enqueued, so it raises no checkpoint failure and never shows up in `is_continuous_checkpoint_failures`; that gauge stays 0 throughout. +2. Skip the ng unless leader/candidate-leader (term from `Sharder`). Once eligibility is established, record a checkpoint attempt, compute `ckpt_ts` (§3.2), and skip if `ckpt_ts <= GetNodeGroupCkptTs(ng)`. Consecutive skips are counted per ng: after `ckpt_stall_warn_rounds` of them (default 3, 0 disables; runtime-mutable) a rate-limited (60 s) `WARNING` names the transaction pinning the timestamp — `CkptTsCc::GetPinningTx()`, filled by `ActiveTxMinTs` with the tx whose `wlock_ts_` produced the minimum — plus its core and write-lock age, so a wedged or leaked write-lock holder is diagnosable from the log alone (`Checkpointer::WarnIfCkptStalled`). A stall returns before any data sync task is enqueued: it contributes to the attempt-interval distribution but raises no checkpoint failure and does not change `is_continuous_checkpoint_failures`. There is intentionally no checkpoint-stall metric; the warning is the diagnostic signal. 3. `GetCatalogTableNameSnapshot(ng, ckpt_ts)` → map table → `is_dirty`. For each non-meta table: if not dirty and `GetTableLastCommitTsCc < last_ckpt_ts`, skip; else `EnqueueDataSyncTaskForTable(..., can_be_skipped = !is_last_ckpt, status)` (`:274-329`). The smallest valid per-table `last synced ts` may already allow an early `UpdateNodeGroupCkptTs` + `NotifyLogOfCkptTs` + `BrocastPrimaryCkptTs` (`:338-358`). -4. Mark `status->all_task_started_`; if all scans finished but flushes are pending, force-flush the buffer (`FlushCurrentFlushBuffer`). On the last ckpt, block on `status->cv_` until `unfinished_tasks_ == 0` (`:360-375`). -5. When all tasks are done without error and `need_truncate_log_`, truncate using `status->truncate_log_ts_` (which may exceed this round's `ckpt_ts` — see §8): `UpdateNodeGroupCkptTs` → `NotifyLogOfCkptTs` (→ `TxLog::UpdateCheckpointTs`, skipped under `txservice_skip_wal`) → `BrocastPrimaryCkptTs` to standbys (`:377-425`). +4. Mark `status->all_task_started_`; if all scans finished but flushes are pending, force-flush the buffer (`FlushCurrentFlushBuffer`). On the last ckpt, block on `status->cv_` until `unfinished_tasks_ == 0` (`:360-375`). The zero-task path explicitly runs the same checkpoint-outcome finalizer used by asynchronous completion. +5. When all tasks are done without error and `need_truncate_log_`, truncate using `status->truncate_log_ts_` (which may exceed this round's `ckpt_ts` — see §8): `UpdateNodeGroupCkptTs` → `NotifyLogOfCkptTs` (→ `TxLog::UpdateCheckpointTs`, skipped under `txservice_skip_wal`) → `BrocastPrimaryCkptTs` to standbys (`:377-425`). A local `UpdateNodeGroupCkptTs` that actually changes the atomic checkpoint timestamp records a checkpoint-advance event; recovery and timestamps received from another node do not. `LogAgent::UpdateCheckpointTs` (`log_agent.cpp:437-469`) fires the `UpdateCheckpointTsRequest{ng, term, ckpt_ts}` at **every** log group (100 ms timeout, fire-and-forget — a missed update just delays truncation until the next round). @@ -111,7 +111,7 @@ Machinery: `tx_service/include/data_sync_task.h` + `tx_service/include/cc/local_ | Object | Granularity | Purpose | |---|---|---| | `DataSyncTask` | per core (hash-partitioned) or per range (range-partitioned) | one scan+flush unit; carries `data_sync_ts_`, ng id/term, flags (`is_dirty_`, `forward_cache_`, `is_standby_node_ckpt_`, `sync_ts_adjustable_`) | -| `DataSyncStatus` | per ckpt round per ng | counts `unfinished_tasks_` / `unfinished_scan_tasks_`, accumulates `truncate_log_ts_`, `need_truncate_log_`, error code | +| `DataSyncStatus` | per data-sync operation per ng | carries an immutable origin; counts `unfinished_tasks_` / `unfinished_scan_tasks_`; accumulates `truncate_log_ts_`, no-truncate reason, first checkpoint failure stage, and error code; finalizes checkpoint observability exactly once | | `FlushTaskEntry` | per scan batch | `data_sync_vec_` (base rows), `archive_vec_` (MVCC versions), `mv_base_vec_` (base→archive moves), schema, owning task | | `FlushDataTask` | per data-sync worker | buffer of `FlushTaskEntry`s keyed by kv table name; flushed when > `max_pending_flush_size_` (default 100 MB, `data_sync_task.h:301`) | @@ -126,20 +126,48 @@ Flow: 4. `PersistKV` if `store_hd_->NeedPersistKV()` (e.g. EloqStore) — batched fsync-equivalent; 5. `UpdateCceCkptTsCc` per shard — stamp `cce->SetCkptTs(commit_ts)` on every flushed entry (only when `need_update_ckpt_ts_`); 6. `WaitableCc` → `CcShard::OnDirtyDataFlushed()` — re-arm kickout requests blocked on dirty data. -4. **Completion & truncation** — `DataSyncTask::SetFinish/SetError` (`tx_service/src/data_sync_task.cpp:113-198`) maintain `truncate_log_ts_ = min(data_sync_ts_)` over the round's tasks. The last task to finish (or `Ckpt()` itself) performs `UpdateNodeGroupCkptTs` + `UpdateCheckpointTs` + `BrocastPrimaryCkptTs` (`tx_service/src/standby.cpp:107`, the `UpdateStandbyCkptTs` RPC). **Truncation contract: never report a ckpt ts unless every entry with `commit_ts <= ts` of this ng is durable in the kv store.** +4. **Completion & truncation** — `DataSyncTask::SetFinish/SetError` maintain `truncate_log_ts_ = min(data_sync_ts_)` over the operation's tasks. Once `all_task_started_` is true, the last task to finish (or `Ckpt()` for a zero-task round) performs exactly-once outcome finalization. Only `Origin::Checkpoint` can affect checkpoint failure metrics. Deduplicated/skipped success is neutral, a remaining task error is still a failure, and `NG_TERM_CHANGED` / `REQUESTED_NODE_NOT_LEADER` is cancellation. A successful truncatable operation performs `UpdateNodeGroupCkptTs` + `UpdateCheckpointTs` + `BrocastPrimaryCkptTs` (`tx_service/src/standby.cpp:107`, the `UpdateStandbyCkptTs` RPC). **Truncation contract: never report a ckpt ts unless every entry with `commit_ts <= ts` of this ng is durable in the kv store.** ### 3.5 ckpt_ts on entries, eviction, dirty-memory trigger - `CkptTs()` / monotonic `SetCkptTs()` live in `VersionedLruEntry`'s entry info (`tx_service/include/cc/cc_entry.h:580-662`). `IsDirty()` = `CommitTs > CkptTs` (versioned) or flush-bit unset (non-versioned); `IsFree()` (no locks ∧ not dirty) gates eviction — **only checkpointed entries can be kicked out** (`LocalCcShards::KickoutPage`, `local_cc_shards.h:1566`, additionally consults range `last_sync_ts`/dirty-range version for range tables). When eviction finds nothing free, the tx processor calls `ckpter_->Notify()` — memory pressure drives checkpointing. - `CcShard::CheckAndTriggerCkptByDirtyMemory` (`tx_service/src/cc/cc_shard.cpp:484-521`), checked every `dirty_memory_check_interval` (1000) key-stat updates: `dirty_memory = allocated_heap × dirty_key_ratio`; if it exceeds `dirty_memory_size_threshold_mb` (default 0 → 10% of the per-shard memory limit, min 1 MB, `cc_shard.cpp:126-141`), call `NotifyCkpt(true)`. +### 3.6 Checkpoint and cache observability + +Cache gauges are sampled on each shard with the existing memory-metric cadence. +`resident_data_key_count` exports the incrementally maintained number of +entries in every non-meta CCMap, including deleted or transient entries until +physical removal. `dirty_data_key_count` exports the subset still requiring a +durable checkpoint. Consumers derive the node dirty-key ratio as the ratio of +the sums, not the average of shard ratios. + +Checkpoint timing state is maintained per node group and leadership term but +exported through a node meter without `ng_id` or `core_id`: + +| Metric | Meaning | +|---|---| +| `checkpoint_attempt_interval_seconds` | Start-to-start time between eligible attempts for the same NG; the first event establishes an anchor. Stalls, no-work rounds, and coalesced rounds remain attempts. | +| `checkpoint_advance_interval_seconds` | Time between successful local durable checkpoint-ts advances for the same NG; the first advance establishes an anchor. | +| `checkpoint_failures_total{reason}` | Exactly one increment for a terminal checkpoint-origin failure, attributed to `scan`, `copy_base`, `put_base`, `put_archive`, `persist`, `metadata`, or `unknown`. | +| `is_continuous_checkpoint_failures` | Existing node alert signal: 1 if any locally tracked NG reaches three consecutive terminal failures, otherwise 0. | + +Successful and genuine no-work checkpoints clear only their NG's consecutive +failure streak. Coalesced/skipped outcomes, stalls, and term cancellations are +neutral. `CcNode::OnLeaderStop` erases that NG's streak and timing anchors after +invalidating its term, while the cumulative failure counter remains. Metric +callbacks revalidate the term while holding the same state mutex used for +cleanup, so a callback from the old term cannot recreate erased state. + ## 4. Data sync beyond checkpointing (overview) -The same task/scan/flush pipeline serves (all with `can_be_skipped=false`, usually waited on via a `CcHandlerResult`): +The same task/scan/flush pipeline serves operations with explicit non-checkpoint +origins (all with `can_be_skipped=false`, usually waited on via a +`CcHandlerResult`): - **Range split** — sync a subrange before ownership changes (`EnqueueDataSyncTaskForSplittingRange`, `local_cc_shards.cpp:2573`; the second `DataSyncTask` constructor with `start_key/end_key/export_base_table_items`). See [08-range-and-bucket-management.md](08-range-and-bucket-management.md). - **Bucket migration / cluster scale** — `EnqueueDataSyncTaskForBucket` (`local_cc_shards.cpp:2905`) flushes bucket data and forwards cache to the new owner. See [06-distribution-and-clustering.md](06-distribution-and-clustering.md) and [08-range-and-bucket-management.md](08-range-and-bucket-management.md). -- **Standby bootstrap & backup** — `SnapshotManager::RunOneRoundCheckpoint` (§7), `FlushDataAll` / `NotifyShutdownCkpt` RPCs. +- **Standby bootstrap & backup** — `SnapshotManager::RunOneRoundCheckpoint` (§7), plus explicit `FlushDataAll` / `NotifyShutdownCkpt` RPC work. ## 5. Recovery @@ -260,6 +288,7 @@ processed by the `replay_notify` thread. `ProcessRecoverTxTask` (`log_replay_ser - **Unknown log result ⇒ locks stay.** After exhausting retries the coordinator leaves write locks in place (status `Unknown`); correctness relies on `CheckTxStatus`/`RecoverTx`, never on lock timeouts. - **Eviction needs checkpointing.** Only entries with `CommitTs <= CkptTs` are evictable. With `skip_kv` there is no checkpointer at all; with cache replacement disabled, `RestoreTxCache` reloads the entire store on leader start. - **Standby checkpoints** happen only on non-shared storage (each replica owns its kv store) and never call `UpdateCheckpointTs` (`is_standby_node_ckpt_`, `data_sync_task.cpp:145-155`); on shared storage only the primary flushes and broadcasts its ckpt ts so standbys can advance entry `ckpt_ts` and evict. +- **Checkpoint metric state is leadership-tenure scoped.** Attempt/advance anchors and consecutive-failure streaks are keyed by NG and term, and are erased when that NG leaves the node. Cumulative failure counters are process-lifetime history and are not reset by failover. - **bthread caveat**: `CkptTsCc::Wait` / `WaitableCc::Wait` poll atomics with `bthread_usleep` backoff (`cc_req_misc.cpp:1138-1149`) instead of bthread condition variables — see the wake-routing deadlock pattern in [02-threading-model.md](02-threading-model.md) and `CLAUDE.md` before adding any new waitable cc request to this module. ## Appendix A — Configuration knobs diff --git a/eloq_metrics/include/meter.h b/eloq_metrics/include/meter.h index ca5de0b8e..6acc8693a 100644 --- a/eloq_metrics/include/meter.h +++ b/eloq_metrics/include/meter.h @@ -141,6 +141,8 @@ class Meter * @param type The type of the metric. * @param label_groups The deque of label groups. Each group contains a * label name and a vector of label types. + * @param histogram_buckets Explicit upper bounds for this histogram. An + * empty vector preserves the collector's default buckets. * * This function registers a metric with the provided name and type, along * with the specified label groups. The label groups represent different @@ -157,7 +159,8 @@ class Meter */ void Register(const Name &name, const Type &type, - std::vector &&dynamic_label_groups = {}) + std::vector &&dynamic_label_groups = {}, + const HistogramBuckets &histogram_buckets = {}) { std::vector all_labels; @@ -172,8 +175,8 @@ class Meter auto name_str = name.GetName(); for (const auto &labels : all_labels) { - auto metric_handle = - metrics_registry_->Register(name_str, type, labels); + auto metric_handle = metrics_registry_->Register( + name_str, type, labels, histogram_buckets); auto meter_key = Hash(name_str); for (size_t i = common_label_groups_.size(); i < labels.size(); ++i) { diff --git a/eloq_metrics/include/metrics.h b/eloq_metrics/include/metrics.h index 456848b7d..0c29fb670 100644 --- a/eloq_metrics/include/metrics.h +++ b/eloq_metrics/include/metrics.h @@ -30,6 +30,7 @@ #include #include #include +#include #include namespace metrics @@ -54,6 +55,9 @@ class Name using Clock = std::chrono::steady_clock; using Labels = std::vector>; +/** Explicit upper bounds for one histogram; empty selects collector defaults. + */ +using HistogramBuckets = std::vector; using TimePoint = decltype(Clock::now()); using MetricKey = size_t; @@ -126,11 +130,18 @@ struct Metric std::string name_; Type type_; Labels labels_; + HistogramBuckets histogram_buckets_; Metric() = delete; - Metric(const std::string &name, metrics::Type type, const Labels &labels) - : name_(name), type_(type), labels_(labels) + Metric(const std::string &name, + metrics::Type type, + const Labels &labels, + HistogramBuckets histogram_buckets = {}) + : name_(name), + type_(type), + labels_(labels), + histogram_buckets_(std::move(histogram_buckets)) { } @@ -184,7 +195,8 @@ class MetricsRegistry virtual MetricsErrors Open() = 0; virtual MetricHandle Register(const Name &, metrics::Type, - const Labels &) = 0; + const Labels &, + const HistogramBuckets & = {}) = 0; virtual void Collect(const MetricHandle &, const Value &) = 0; virtual ~MetricsRegistry() = default; }; diff --git a/eloq_metrics/include/metrics_registry_impl.h b/eloq_metrics/include/metrics_registry_impl.h index a908a542c..8900e5fc6 100644 --- a/eloq_metrics/include/metrics_registry_impl.h +++ b/eloq_metrics/include/metrics_registry_impl.h @@ -45,7 +45,8 @@ class MetricsRegistryImpl : public metrics::MetricsRegistry metrics::MetricsErrors Open() override; metrics::MetricHandle Register(const metrics::Name &, metrics::Type, - const metrics::Labels &) override; + const metrics::Labels &, + const metrics::HistogramBuckets &) override; void Collect(const metrics::MetricHandle &, const metrics::Value &) override; diff --git a/eloq_metrics/src/metrics_registry_impl.cpp b/eloq_metrics/src/metrics_registry_impl.cpp index 651efbd94..d8fdf3811 100644 --- a/eloq_metrics/src/metrics_registry_impl.cpp +++ b/eloq_metrics/src/metrics_registry_impl.cpp @@ -59,9 +59,11 @@ metrics::MetricsErrors MetricsRegistryImpl::Open() metrics::MetricHandle MetricsRegistryImpl::Register( const metrics::Name &name, metrics::Type type, - const metrics::Labels &labels) + const metrics::Labels &labels, + const metrics::HistogramBuckets &histogram_buckets) { - auto metric = metrics::Metric(name.GetName(), type, labels); + auto metric = + metrics::Metric(name.GetName(), type, labels, histogram_buckets); return metrics_mgr_result_.mgr_->MetricsRegistry( std::make_unique(metric)); diff --git a/eloq_metrics/src/prometheus_collector.cc b/eloq_metrics/src/prometheus_collector.cc index f04b443a1..f359bf657 100644 --- a/eloq_metrics/src/prometheus_collector.cc +++ b/eloq_metrics/src/prometheus_collector.cc @@ -145,8 +145,10 @@ MetricHandle PrometheusCollector::SetMetric(std::unique_ptr &metric_ptr) auto &histogram_family = prometheus::BuildHistogram() .Name(metric_ptr->name_) .Register(*registry_); - auto &histogram = histogram_family.Add( - prometheus_labels, PROMETHEUS_HISTOGRAM_DEF_BUCKETS); + const auto &buckets = metric_ptr->histogram_buckets_.empty() + ? PROMETHEUS_HISTOGRAM_DEF_BUCKETS + : metric_ptr->histogram_buckets_; + auto &histogram = histogram_family.Add(prometheus_labels, buckets); data = std::make_shared(histogram); break; } diff --git a/eloq_metrics/tests/metrics_collector_test.cc b/eloq_metrics/tests/metrics_collector_test.cc index 7da023f1e..034df6e26 100644 --- a/eloq_metrics/tests/metrics_collector_test.cc +++ b/eloq_metrics/tests/metrics_collector_test.cc @@ -21,9 +21,39 @@ */ #include #include +#include +#include "meter.h" #include "prometheus_collector.h" +namespace +{ +class RecordingRegistry : public metrics::MetricsRegistry +{ +public: + metrics::MetricsErrors Open() override + { + return metrics::MetricsErrors::Success; + } + + metrics::MetricHandle Register( + const metrics::Name &, + metrics::Type type, + const metrics::Labels &, + const metrics::HistogramBuckets &histogram_buckets) override + { + histogram_buckets_.push_back(histogram_buckets); + return metrics::MetricHandle(histogram_buckets_.size(), type); + } + + void Collect(const metrics::MetricHandle &, const metrics::Value &) override + { + } + + std::vector histogram_buckets_; +}; +} // namespace + SCENARIO("Metrics Collector no open", "[MCNoOpen]") { INFO("Run unit test MCNoOpen"); @@ -136,3 +166,45 @@ SCENARIO("Metrics collector call Open several times", } } } + +SCENARIO("Histograms can override default buckets", "[HistogramBuckets]") +{ + RecordingRegistry registry; + metrics::Meter meter(®istry, {}); + meter.Register(metrics::Name{"custom_bucket_histogram"}, + metrics::Type::Histogram, + {}, + {1.0, 5.0, 3600.0}); + meter.Register(metrics::Name{"default_bucket_histogram"}, + metrics::Type::Histogram); + REQUIRE(registry.histogram_buckets_.size() == 2); + REQUIRE(registry.histogram_buckets_[0] == + (metrics::HistogramBuckets{1.0, 5.0, 3600.0})); + REQUIRE(registry.histogram_buckets_[1].empty()); + + metrics::PrometheusCollector collector{"0.0.0.0", 18083}; + REQUIRE(collector.Open()); + + metrics::Metric custom_metric{"custom_bucket_histogram", + metrics::Type::Histogram, + {}, + {1.0, 5.0, 3600.0}}; + auto custom_metric_ptr = std::make_unique(custom_metric); + auto custom_handle = collector.SetMetric(custom_metric_ptr); + REQUIRE(collector.Collect(custom_handle, metrics::Value{6.0})); + + auto custom_sample = collector.CollectClientMetrics(custom_handle); + // prometheus-cpp appends its implicit +Inf bucket. + REQUIRE(custom_sample.histogram.bucket.size() == 4); + REQUIRE(custom_sample.histogram.bucket[0].upper_bound == 1.0); + REQUIRE(custom_sample.histogram.bucket[1].upper_bound == 5.0); + REQUIRE(custom_sample.histogram.bucket[2].upper_bound == 3600.0); + + metrics::Metric default_metric{ + "default_bucket_histogram", metrics::Type::Histogram, {}}; + auto default_metric_ptr = std::make_unique(default_metric); + auto default_handle = collector.SetMetric(default_metric_ptr); + auto default_sample = collector.CollectClientMetrics(default_handle); + REQUIRE(default_sample.histogram.bucket.size() == + metrics::PROMETHEUS_HISTOGRAM_DEF_BUCKETS.size() + 1); +} diff --git a/tx_service/include/checkpoint_metrics_state.h b/tx_service/include/checkpoint_metrics_state.h new file mode 100644 index 000000000..1e9bdc901 --- /dev/null +++ b/tx_service/include/checkpoint_metrics_state.h @@ -0,0 +1,180 @@ +/** + * Copyright (C) 2025 EloqData Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under either of the following two licenses: + * 1. GNU Affero General Public License, version 3, as published by the Free + * Software Foundation. + * 2. GNU General Public License as published by the Free Software + * Foundation; version 2 of the License. + */ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "type.h" + +namespace txservice +{ +/** + * Pure state machine behind node-aggregated checkpoint metrics. + * + * Checkpointer serializes calls and validates leadership terms before entering + * this class. Keeping metric I/O and Sharder access outside makes every per-NG + * transition deterministic and independently testable. + */ +class CheckpointMetricsState +{ +public: + using TimePoint = std::chrono::steady_clock::time_point; + + struct Update + { + std::optional interval_seconds_; + std::optional continuous_failure_gauge_; + }; + + explicit CheckpointMetricsState(size_t failure_threshold) + : failure_threshold_(failure_threshold) + { + assert(failure_threshold_ > 0); + } + + CheckpointMetricsState(const CheckpointMetricsState &) = delete; + CheckpointMetricsState &operator=(const CheckpointMetricsState &) = delete; + CheckpointMetricsState(CheckpointMetricsState &&) = delete; + CheckpointMetricsState &operator=(CheckpointMetricsState &&) = delete; + + /** Records an eligible attempt and returns an interval after its anchor. */ + Update RecordAttempt(NodeGroupId node_group_id, int64_t term, TimePoint now) + { + Update update; + NodeGroupState &state = StateForTerm(node_group_id, term, update); + if (state.last_attempt_.has_value()) + { + update.interval_seconds_ = + std::chrono::duration(now - *state.last_attempt_) + .count(); + } + state.last_attempt_ = now; + return update; + } + + /** Records a durable advance and returns an interval after its anchor. */ + Update RecordAdvance(NodeGroupId node_group_id, int64_t term, TimePoint now) + { + Update update; + NodeGroupState &state = StateForTerm(node_group_id, term, update); + if (state.last_advance_.has_value()) + { + update.interval_seconds_ = + std::chrono::duration(now - *state.last_advance_) + .count(); + } + state.last_advance_ = now; + return update; + } + + /** Clears this NG's failure streak without changing another NG. */ + Update RecordSuccess(NodeGroupId node_group_id, int64_t term) + { + Update update; + NodeGroupState &state = StateForTerm(node_group_id, term, update); + state.consecutive_failures_ = 0; + if (state.continuous_failure_) + { + state.continuous_failure_ = false; + assert(continuous_failure_ng_count_ > 0); + --continuous_failure_ng_count_; + update.continuous_failure_gauge_ = continuous_failure_ng_count_ > 0; + } + return update; + } + + /** Increments this NG's streak and raises the node signal at threshold. */ + Update RecordFailure(NodeGroupId node_group_id, int64_t term) + { + Update update; + NodeGroupState &state = StateForTerm(node_group_id, term, update); + ++state.consecutive_failures_; + if (!state.continuous_failure_ && + state.consecutive_failures_ >= failure_threshold_) + { + state.continuous_failure_ = true; + ++continuous_failure_ng_count_; + update.continuous_failure_gauge_ = true; + } + return update; + } + + /** Erases all leadership-tenure state for an NG. */ + Update Erase(NodeGroupId node_group_id) + { + Update update; + auto it = states_.find(node_group_id); + if (it == states_.end()) + { + return update; + } + if (it->second.continuous_failure_) + { + assert(continuous_failure_ng_count_ > 0); + --continuous_failure_ng_count_; + update.continuous_failure_gauge_ = continuous_failure_ng_count_ > 0; + } + states_.erase(it); + return update; + } + + size_t ConsecutiveFailures(NodeGroupId node_group_id) const + { + auto it = states_.find(node_group_id); + return it == states_.end() ? 0 : it->second.consecutive_failures_; + } + + bool Contains(NodeGroupId node_group_id) const + { + return states_.find(node_group_id) != states_.end(); + } + +private: + struct NodeGroupState + { + int64_t term_{-1}; + size_t consecutive_failures_{0}; + bool continuous_failure_{false}; + std::optional last_attempt_; + std::optional last_advance_; + }; + + NodeGroupState &StateForTerm(NodeGroupId node_group_id, + int64_t term, + Update &update) + { + auto [it, inserted] = states_.try_emplace(node_group_id); + NodeGroupState &state = it->second; + if (!inserted && state.term_ != term) + { + if (state.continuous_failure_) + { + assert(continuous_failure_ng_count_ > 0); + --continuous_failure_ng_count_; + update.continuous_failure_gauge_ = + continuous_failure_ng_count_ > 0; + } + state = NodeGroupState{}; + } + state.term_ = term; + return state; + } + + const size_t failure_threshold_; + std::unordered_map states_; + size_t continuous_failure_ng_count_{0}; +}; +} // namespace txservice diff --git a/tx_service/include/checkpointer.h b/tx_service/include/checkpointer.h index 195b12629..a486c4346 100644 --- a/tx_service/include/checkpointer.h +++ b/tx_service/include/checkpointer.h @@ -33,6 +33,7 @@ #include "cc/cc_request.h" #include "cc/local_cc_shards.h" +#include "checkpoint_metrics_state.h" #include "metrics.h" #include "sharder.h" #include "txlog.h" @@ -106,38 +107,24 @@ class Checkpointer void Join(); - void CollectCkptMetric(bool success) - { - if (metrics::enable_metrics) - { - if (success) - { - if (consecutive_fail_cnt_ > 0) - { - Sharder::Instance() - .GetLocalCcShards() - ->GetNodeMeter() - ->Collect( - metrics::NAME_IS_CONTINUOUS_CHECKPOINT_FAILURES, 0); - } - // reset value - consecutive_fail_cnt_.store(0, std::memory_order_relaxed); - } - else - { - size_t fail_cnt = consecutive_fail_cnt_.fetch_add( - 1, std::memory_order_relaxed); - if (fail_cnt + 1 >= continuous_ckpt_fail_threshold) - { - Sharder::Instance() - .GetLocalCcShards() - ->GetNodeMeter() - ->Collect( - metrics::NAME_IS_CONTINUOUS_CHECKPOINT_FAILURES, 1); - } - } - } - } + /** Records one eligible checkpoint attempt for interval telemetry. */ + void RecordCheckpointAttempt(NodeGroupId node_group_id, int64_t term); + + /** Records a successful local durable checkpoint-ts advance. */ + void RecordCheckpointAdvance(NodeGroupId node_group_id, int64_t term); + + /** Applies one terminal checkpoint outcome to per-NG failure state. */ + void ReportCheckpointOutcome( + NodeGroupId node_group_id, + int64_t term, + DataSyncStatus::CheckpointOutcome outcome, + DataSyncStatus::CheckpointFailureReason failure_reason); + + /** + * Drops leadership-tenure metrics state after this node loses an NG. + * Cumulative failure counters are intentionally retained. + */ + void ClearCheckpointMetricsForNodeGroup(NodeGroupId node_group_id); void IncrementOngoingDataSyncCnt() { @@ -180,7 +167,12 @@ class Checkpointer TxService *tx_service_; TxLog *log_agent_; - std::atomic consecutive_fail_cnt_{0}; + // Checkpoint callbacks and raft leadership callbacks run on different + // threads. The term is revalidated while holding this mutex so a callback + // from a failed-over term cannot recreate erased NG state. + std::mutex checkpoint_metrics_mux_; + CheckpointMetricsState checkpoint_metrics_state_{ + continuous_ckpt_fail_threshold}; // Per-node-group checkpoint-stall tracking: how many consecutive Ckpt() // rounds the checkpoint ts failed to advance (reset to 0 whenever it @@ -198,14 +190,6 @@ class Checkpointer void NotifyLogOfCkptTs(uint32_t node_group, int64_t term, uint64_t ckpt_ts); - /** - * @brief Exports @p rounds as the checkpoint_stall_rounds gauge (per - * ng_id, node meter). Called on every round, stalled or not: a gauge that - * skips rounds reads as a flatline at its last value, which is exactly - * the silence this diagnoses. - */ - void CollectCkptStallMetric(uint32_t node_group, uint32_t rounds); - /** * @brief Tracks consecutive rounds in which @p node_group's checkpoint ts * failed to advance and, past a threshold, logs a rate-limited warning @@ -215,5 +199,12 @@ class Checkpointer uint64_t ckpt_ts, uint64_t last_ckpt_ts, const CkptTsCc::PinningTxInfo &pinning_tx); + + bool IsCurrentCheckpointTerm(NodeGroupId node_group_id, int64_t term) const; + void ApplyCheckpointMetricsUpdateLocked( + const CheckpointMetricsState::Update &update, + const metrics::Name *interval_metric = nullptr); + static const char *CheckpointFailureReasonLabel( + DataSyncStatus::CheckpointFailureReason reason); }; } // namespace txservice diff --git a/tx_service/include/data_sync_task.h b/tx_service/include/data_sync_task.h index 821e8f337..a6d49f387 100644 --- a/tx_service/include/data_sync_task.h +++ b/tx_service/include/data_sync_task.h @@ -29,6 +29,7 @@ #include #include #include +#include #include #include @@ -48,16 +49,71 @@ struct DataSyncTask; struct DataSyncStatus { + /** Identifies which subsystem owns this shared data-sync pipeline run. */ + enum class Origin + { + Checkpoint, + Snapshot, + FlushData, + CreateIndex, + RangeSplit, + Migration + }; + + /** Explains why successful work cannot advance the checkpoint watermark. */ + enum class NoTruncateReason + { + None, + Deduplicated, + EntriesSkipped, + NotCheckpoint + }; + + /** Terminal checkpoint classification consumed by observability state. */ + enum class CheckpointOutcome + { + Success, + Failure, + Neutral, + Canceled + }; + + /** Stable low-cardinality stage attribution for checkpoint failures. */ + enum class CheckpointFailureReason + { + None, + Scan, + CopyBase, + PutBase, + PutArchive, + Persist, + Metadata, + Unknown + }; + + /** Exactly-once terminal result returned by checkpoint finalization. */ + struct CheckpointResult + { + CheckpointOutcome outcome_; + CheckpointFailureReason failure_reason_; + }; + explicit DataSyncStatus(NodeGroupId node_group_id, int64_t node_group_term, - bool need_truncate_log); + bool need_truncate_log, + Origin origin); ~DataSyncStatus(); - void SetNoTruncateLog() + void SetNoTruncateLog( + NoTruncateReason reason = NoTruncateReason::Deduplicated) { std::lock_guard lk(mux_); need_truncate_log_ = false; + if (no_truncate_reason_ == NoTruncateReason::None) + { + no_truncate_reason_ = reason; + } } void SetEntriesSkippedAndNoTruncateLog() @@ -65,8 +121,18 @@ struct DataSyncStatus std::lock_guard lk(mux_); has_skipped_entries_ = true; need_truncate_log_ = false; + no_truncate_reason_ = NoTruncateReason::EntriesSkipped; } + /** Records the first checkpoint failure stage observed by this sync. */ + void RecordCheckpointFailure(CheckpointFailureReason reason); + + /** + * Finalizes checkpoint observability exactly once after all tasks finish. + * The caller must hold mux_. Non-checkpoint data syncs return nullopt. + */ + std::optional TryFinalizeCheckpointLocked(); + void MarkDataStoreWrite() { has_data_store_write_.store(true, std::memory_order_release); @@ -79,6 +145,7 @@ struct DataSyncStatus NodeGroupId node_group_id_; int64_t node_group_term_; + const Origin origin_; // Number of unfinished scan tasks. We keep track of this separately since // we need to flush the flush data buffer when all scan tasks are finished. int32_t unfinished_scan_tasks_{0}; @@ -90,6 +157,7 @@ struct DataSyncStatus CcErrorCode err_code_{CcErrorCode::NO_ERROR}; // True if need to truncate redo log when all tasks succeed. bool need_truncate_log_{true}; + NoTruncateReason no_truncate_reason_{NoTruncateReason::None}; uint64_t truncate_log_ts_{0}; // Collect from each data sync task. size_t total_entry_cnt_{0}; @@ -98,6 +166,10 @@ struct DataSyncStatus // entries with buffer commands might be skipped. bool has_skipped_entries_{false}; + CheckpointFailureReason checkpoint_failure_reason_{ + CheckpointFailureReason::None}; + bool checkpoint_finalized_{false}; + // Whether there is any data written to datastore in this DataSync round. std::atomic has_data_store_write_{false}; @@ -174,7 +246,10 @@ struct DataSyncTask void SetFinish(); - void SetError(CcErrorCode err_code = CcErrorCode::DATA_STORE_ERR); + void SetError( + CcErrorCode err_code = CcErrorCode::DATA_STORE_ERR, + DataSyncStatus::CheckpointFailureReason checkpoint_failure_reason = + DataSyncStatus::CheckpointFailureReason::Unknown); // Decrease unfinished_scan_tasks_ by 1. If all scan tasks are finished, // and there are still unfinished data sync tasks, that means there might @@ -188,10 +263,21 @@ struct DataSyncTask // completes. void ResetRangeSplittingStatus(); - void SetErrorCode(CcErrorCode err_code) + void SetErrorCode( + CcErrorCode err_code, + DataSyncStatus::CheckpointFailureReason checkpoint_failure_reason = + DataSyncStatus::CheckpointFailureReason::Unknown) { std::unique_lock lk(status_->mux_); status_->err_code_ = err_code; + if (status_->origin_ == DataSyncStatus::Origin::Checkpoint && + err_code != CcErrorCode::NG_TERM_CHANGED && + err_code != CcErrorCode::REQUESTED_NODE_NOT_LEADER && + status_->checkpoint_failure_reason_ == + DataSyncStatus::CheckpointFailureReason::None) + { + status_->checkpoint_failure_reason_ = checkpoint_failure_reason; + } } bool SyncTsAdjustable() const diff --git a/tx_service/include/tx_service_metrics.h b/tx_service/include/tx_service_metrics.h index 0789b0ec1..e3b8d2344 100644 --- a/tx_service/include/tx_service_metrics.h +++ b/tx_service/include/tx_service_metrics.h @@ -40,6 +40,9 @@ inline const metrics::Name NAME_MEMORY_LIMIT{"memory_limit"}; inline const metrics::Name NAME_CACHE_HIT_OR_MISS_TOTAL{ "cache_hit_or_miss_total"}; inline const metrics::Name NAME_MEMORY_USAGE{"memory_usage"}; +inline const metrics::Name NAME_RESIDENT_DATA_KEY_COUNT{ + "resident_data_key_count"}; +inline const metrics::Name NAME_DIRTY_DATA_KEY_COUNT{"dirty_data_key_count"}; inline const metrics::Name NAME_FRAGMENT_RATIO{"memory_fragment_ratio"}; @@ -63,13 +66,18 @@ inline const metrics::Name NAME_STANDBY_OUT_OF_SYNC_COUNT{ inline const metrics::Name NAME_IS_CONTINUOUS_CHECKPOINT_FAILURES{ "is_continuous_checkpoint_failures"}; +inline const metrics::Name NAME_CHECKPOINT_ATTEMPT_INTERVAL_SECONDS{ + "checkpoint_attempt_interval_seconds"}; +inline const metrics::Name NAME_CHECKPOINT_ADVANCE_INTERVAL_SECONDS{ + "checkpoint_advance_interval_seconds"}; +inline const metrics::Name NAME_CHECKPOINT_FAILURES_TOTAL{ + "checkpoint_failures_total"}; -// Consecutive checkpoint rounds in which the node group's checkpoint ts failed -// to advance; 0 while it advances. A stall never reaches the data sync stage, -// so it raises no failure and is invisible in -// is_continuous_checkpoint_failures. -inline const metrics::Name NAME_CHECKPOINT_STALL_ROUNDS{ - "checkpoint_stall_rounds"}; +// Checkpoint intervals are seconds-to-minutes operational signals. These +// buckets retain useful resolution at the low end without losing visibility +// into checkpoints delayed for up to one hour. +inline const metrics::HistogramBuckets CHECKPOINT_INTERVAL_SECONDS_BUCKETS{ + 1, 2, 5, 10, 15, 30, 60, 120, 300, 600, 900, 1800, 3600}; inline const metrics::Name NAME_LEADER_CHANGES{"leader_changes_seen_total"}; inline const metrics::Name NAME_IS_LEADER{"is_leader"}; diff --git a/tx_service/src/cc/cc_shard.cpp b/tx_service/src/cc/cc_shard.cpp index 4def78986..84a9818ea 100644 --- a/tx_service/src/cc/cc_shard.cpp +++ b/tx_service/src/cc/cc_shard.cpp @@ -220,6 +220,10 @@ CcShard::CcShard( { meter_->Register(metrics::NAME_MEMORY_USAGE, metrics::Type::Gauge); meter_->Register(metrics::NAME_FRAGMENT_RATIO, metrics::Type::Gauge); + meter_->Register(metrics::NAME_RESIDENT_DATA_KEY_COUNT, + metrics::Type::Gauge); + meter_->Register(metrics::NAME_DIRTY_DATA_KEY_COUNT, + metrics::Type::Gauge); } if (metrics::enable_standby_metrics) @@ -686,6 +690,13 @@ size_t CcShard::ProcessRequests() meter_->Collect(metrics::NAME_FRAGMENT_RATIO, (100 * (static_cast(committed - allocated) / committed))); + // These counters are maintained incrementally by all non-meta + // CCMaps on this shard and include deleted/unknown resident + // entries. Sampling them here avoids a full cache traversal. + meter_->Collect(metrics::NAME_RESIDENT_DATA_KEY_COUNT, + static_cast(data_key_count_)); + meter_->Collect(metrics::NAME_DIRTY_DATA_KEY_COUNT, + static_cast(dirty_data_key_count_)); } } diff --git a/tx_service/src/cc/local_cc_shards.cpp b/tx_service/src/cc/local_cc_shards.cpp index 9434a7fcd..96e72a2cb 100644 --- a/tx_service/src/cc/local_cc_shards.cpp +++ b/tx_service/src/cc/local_cc_shards.cpp @@ -100,6 +100,23 @@ void WaitForStackCcRequestFree(const CcRequestBase &req) interval_us = std::min(interval_us << 1, max_interval_us); } } + +/** + * Validates a data-sync task against the term for the role that started it. + * Standby checkpoint work is valid only for the native NG's active standby + * term; treating a matching leader term as equivalent could let stale work + * advance checkpoint state after a role transition. + */ +bool IsCurrentDataSyncTerm(const DataSyncTask &task) +{ + Sharder &sharder = Sharder::Instance(); + if (task.is_standby_node_ckpt_) + { + return task.node_group_id_ == sharder.NativeNodeGroup() && + sharder.StandbyNodeTerm() == task.node_group_term_; + } + return sharder.CheckLeaderTerm(task.node_group_id_, task.node_group_term_); +} } // namespace std::atomic LocalCcShards::local_clock(0); @@ -238,6 +255,9 @@ LocalCcShards::LocalCcShards( ? conf.at("dirty_memory_size_threshold_mb") : 0; uint16_t core_cnt = conf.at("core_num"); + // Per-node metrics must not inherit the last core_id assigned below. + // Keep an immutable copy of the labels supplied by the embedding server. + const metrics::CommonLabels node_common_labels = common_labels; for (uint16_t thd_idx = 0; thd_idx < core_cnt; ++thd_idx) { common_labels["core_id"] = std::to_string(thd_idx); @@ -277,8 +297,8 @@ LocalCcShards::LocalCcShards( if (metrics::enable_metrics) { - node_meter_ = - std::make_unique(metrics_registry, common_labels); + node_meter_ = std::make_unique(metrics_registry, + node_common_labels); node_meter_->Register(metrics::NAME_IS_CONTINUOUS_CHECKPOINT_FAILURES, metrics::Type::Gauge); @@ -300,19 +320,30 @@ LocalCcShards::LocalCcShards( leader_changes_metric_labels.emplace_back( "ng_id", std::move(ng_id_label_values)); auto is_leader_metric_labels = leader_changes_metric_labels; - // The checkpointer reports stalls for the node groups it iterates - // (Sharder::LocalNodeGroups, plus the native ng on a standby), which - // is the same membership these labels are built from. - auto ckpt_stall_metric_labels = leader_changes_metric_labels; node_meter_->Register(metrics::NAME_LEADER_CHANGES, metrics::Type::Counter, std::move(leader_changes_metric_labels)); node_meter_->Register(metrics::NAME_IS_LEADER, metrics::Type::Gauge, std::move(is_leader_metric_labels)); - node_meter_->Register(metrics::NAME_CHECKPOINT_STALL_ROUNDS, - metrics::Type::Gauge, - std::move(ckpt_stall_metric_labels)); + node_meter_->Register(metrics::NAME_CHECKPOINT_ATTEMPT_INTERVAL_SECONDS, + metrics::Type::Histogram, + {}, + metrics::CHECKPOINT_INTERVAL_SECONDS_BUCKETS); + node_meter_->Register(metrics::NAME_CHECKPOINT_ADVANCE_INTERVAL_SECONDS, + metrics::Type::Histogram, + {}, + metrics::CHECKPOINT_INTERVAL_SECONDS_BUCKETS); + node_meter_->Register(metrics::NAME_CHECKPOINT_FAILURES_TOTAL, + metrics::Type::Counter, + {{"reason", + {"scan", + "copy_base", + "put_base", + "put_archive", + "persist", + "metadata", + "unknown"}}}); } } @@ -2587,6 +2618,13 @@ bool LocalCcShards::EnqueueRangeDataSyncTask( // data before data sync ts is flushed. std::lock_guard status_lk(status->mux_); status->err_code_ = CcErrorCode::PIN_RANGE_SLICE_FAILED; + if (status->origin_ == DataSyncStatus::Origin::Checkpoint && + status->checkpoint_failure_reason_ == + DataSyncStatus::CheckpointFailureReason::None) + { + status->checkpoint_failure_reason_ = + DataSyncStatus::CheckpointFailureReason::Metadata; + } break; } } @@ -2607,8 +2645,8 @@ void LocalCcShards::EnqueueDataSyncTaskForSplittingRange( uint64_t txn, CcHandlerResult *hres) { - std::shared_ptr status = - std::make_shared(ng_id, ng_term, false); + std::shared_ptr status = std::make_shared( + ng_id, ng_term, false, DataSyncStatus::Origin::RangeSplit); const std::vector *new_keys = range_entry->GetRangeInfo()->NewKey(); const std::vector *new_range_ids = range_entry->GetRangeInfo()->NewPartitionId(); @@ -2940,8 +2978,8 @@ void LocalCcShards::EnqueueDataSyncTaskForBucket( CcHandlerResult *hres) { std::shared_lock meta_lk(meta_data_mux_); - std::shared_ptr status = - std::make_shared(ng_id, ng_term, false); + std::shared_ptr status = std::make_shared( + ng_id, ng_term, false, DataSyncStatus::Origin::Migration); uint32_t unfinished_task_cnt = 0; for (auto &[range_table_name, range_ids] : ranges_in_bucket_snapshot) { @@ -3047,8 +3085,8 @@ void LocalCcShards::CreateSplitRangeDataSyncTask(const TableName &table_name, return; }); std::shared_lock meta_lk(meta_data_mux_); - std::shared_ptr status = - std::make_shared(ng_id, ng_term, false); + std::shared_ptr status = std::make_shared( + ng_id, ng_term, false, DataSyncStatus::Origin::RangeSplit); TableName range_table_name(table_name.StringView(), TableType::RangePartition, table_name.Engine()); @@ -3468,7 +3506,9 @@ void LocalCcShards::PostProcessFlushTaskEntries( err_code = CcErrorCode::REQUESTED_NODE_NOT_LEADER; } - task->SetError(err_code); + task->SetError( + err_code, + DataSyncStatus::CheckpointFailureReason::Metadata); task->ResetRangeSplittingStatus(); } } @@ -3789,12 +3829,16 @@ void LocalCcShards::DataSyncForRangePartition( switch (outcome) { case PreCheck::TableDropped: - data_sync_task->SetError(CcErrorCode::REQUESTED_TABLE_NOT_EXISTS); + data_sync_task->SetError( + CcErrorCode::REQUESTED_TABLE_NOT_EXISTS, + DataSyncStatus::CheckpointFailureReason::Metadata); data_sync_task->SetScanTaskFinished(); data_sync_task->ResetRangeSplittingStatus(); for (auto &task : detached_tasks) { - task->SetError(CcErrorCode::REQUESTED_TABLE_NOT_EXISTS); + task->SetError( + CcErrorCode::REQUESTED_TABLE_NOT_EXISTS, + DataSyncStatus::CheckpointFailureReason::Metadata); task->SetScanTaskFinished(); } return; @@ -4028,7 +4072,9 @@ void LocalCcShards::DataSyncForRangePartition( // Use AbortTxRequest to release read lock. txservice::AbortTx(data_sync_txm); - data_sync_task->SetError(); + data_sync_task->SetError( + CcErrorCode::DATA_STORE_ERR, + DataSyncStatus::CheckpointFailureReason::Metadata); data_sync_task->SetScanTaskFinished(); data_sync_task->ResetRangeSplittingStatus(); PopPendingTask(ng_id, expected_ng_term, table_name, range_id); @@ -4044,7 +4090,9 @@ void LocalCcShards::DataSyncForRangePartition( << "DataSync range version mismatch with data sync ts: " << data_sync_task->data_sync_ts_; txservice::AbortTx(data_sync_txm); - data_sync_task->SetError(CcErrorCode::GET_RANGE_ID_ERR); + data_sync_task->SetError( + CcErrorCode::GET_RANGE_ID_ERR, + DataSyncStatus::CheckpointFailureReason::Metadata); data_sync_task->SetScanTaskFinished(); data_sync_task->ResetRangeSplittingStatus(); PopPendingTask(ng_id, expected_ng_term, table_name, range_id); @@ -4190,7 +4238,9 @@ void LocalCcShards::DataSyncForRangePartition( LOG(ERROR) << "Calculate subranges key failed on table " << table_name.StringView(); - data_sync_task->SetError(); + data_sync_task->SetError( + CcErrorCode::DATA_STORE_ERR, + DataSyncStatus::CheckpointFailureReason::Scan); data_sync_task->SetScanTaskFinished(); data_sync_task->ResetRangeSplittingStatus(); // Handle the pending tasks for the same range @@ -4547,8 +4597,7 @@ void LocalCcShards::PostProcessHashPartitionDataSyncTask( meta_data_mux_); // Make sure that the term has not changed so that catalog entry // is still valid. - if (!Sharder::Instance().CheckLeaderTerm( - task->node_group_id_, task->node_group_term_)) + if (!IsCurrentDataSyncTerm(*task)) { err_code = CcErrorCode::NG_TERM_CHANGED; } @@ -4617,21 +4666,9 @@ void LocalCcShards::PostProcessHashPartitionDataSyncTask( { assert(task_ckpt_err == DataSyncTask::CkptErrorCode::FLUSH_ERROR); CcErrorCode err_code = CcErrorCode::DATA_STORE_ERR; - if (task->is_standby_node_ckpt_) + if (!IsCurrentDataSyncTerm(*task)) { - if (Sharder::Instance().LeaderTerm(task->node_group_id_) != - task->node_group_term_) - { - err_code = CcErrorCode::NG_TERM_CHANGED; - } - } - else - { - if (Sharder::Instance().StandbyNodeTerm() != - task->node_group_term_) - { - err_code = CcErrorCode::NG_TERM_CHANGED; - } + err_code = CcErrorCode::NG_TERM_CHANGED; } txservice::AbortTx(data_sync_txm); @@ -4738,11 +4775,14 @@ void LocalCcShards::DataSyncForHashPartition( switch (outcome) { case PreCheck::TableDropped: - data_sync_task->SetError(CcErrorCode::REQUESTED_TABLE_NOT_EXISTS); + data_sync_task->SetError( + CcErrorCode::REQUESTED_TABLE_NOT_EXISTS, + DataSyncStatus::CheckpointFailureReason::Metadata); data_sync_task->SetScanTaskFinished(); for (auto &task : detached_tasks) { - task->SetError(CcErrorCode::REQUESTED_TABLE_NOT_EXISTS); + task->SetError(CcErrorCode::REQUESTED_TABLE_NOT_EXISTS, + DataSyncStatus::CheckpointFailureReason::Metadata); task->SetScanTaskFinished(); } return; @@ -4851,7 +4891,9 @@ void LocalCcShards::DataSyncForHashPartition( // If table is deleted(!Normal), skip the table. Return finish // directly. - data_sync_task->SetError(CcErrorCode::REQUESTED_TABLE_NOT_EXISTS); + data_sync_task->SetError( + CcErrorCode::REQUESTED_TABLE_NOT_EXISTS, + DataSyncStatus::CheckpointFailureReason::Metadata); data_sync_task->SetScanTaskFinished(); ClearAllPendingTasks( @@ -5527,7 +5569,8 @@ void LocalCcShards::ClearAllPendingTasks(NodeGroupId ng_id, while (!iter->second->pending_tasks_.empty()) { auto &task = iter->second->pending_tasks_.front(); - task->SetError(CcErrorCode::REQUESTED_TABLE_NOT_EXISTS); + task->SetError(CcErrorCode::REQUESTED_TABLE_NOT_EXISTS, + DataSyncStatus::CheckpointFailureReason::Metadata); task->SetScanTaskFinished(); iter->second->pending_tasks_.pop_front(); } @@ -6041,6 +6084,8 @@ void LocalCcShards::FlushDataImpl(FlushDataTask *cur_work, { auto &flush_task_entries = cur_work->flush_task_entries_; bool succ = true; + DataSyncStatus::CheckpointFailureReason checkpoint_failure_reason = + DataSyncStatus::CheckpointFailureReason::None; if (EnableMvcc()) { @@ -6048,6 +6093,8 @@ void LocalCcShards::FlushDataImpl(FlushDataTask *cur_work, flush_task_entries, &yield_fn, &resume_fn); if (!succ) { + checkpoint_failure_reason = + DataSyncStatus::CheckpointFailureReason::CopyBase; LOG(ERROR) << "DataSync CopyBaseToArchive flush to kv " "storage failed"; } @@ -6059,6 +6106,8 @@ void LocalCcShards::FlushDataImpl(FlushDataTask *cur_work, flush_task_entries, &yield_fn, &resume_fn, &sync_yield_func); if (!succ) { + checkpoint_failure_reason = + DataSyncStatus::CheckpointFailureReason::PutBase; LOG(ERROR) << "DataSync PutAll flush to kv " "storage failed"; } @@ -6070,6 +6119,8 @@ void LocalCcShards::FlushDataImpl(FlushDataTask *cur_work, flush_task_entries, &yield_fn, &resume_fn); if (!succ) { + checkpoint_failure_reason = + DataSyncStatus::CheckpointFailureReason::PutArchive; LOG(ERROR) << "DataSync PutArchivesAll flush to " "kv storage failed"; } @@ -6083,6 +6134,30 @@ void LocalCcShards::FlushDataImpl(FlushDataTask *cur_work, kv_table_names.push_back(table_name.data()); } succ = store_hd_->PersistKV(kv_table_names, &yield_fn, &resume_fn); + if (!succ) + { + checkpoint_failure_reason = + DataSyncStatus::CheckpointFailureReason::Persist; + LOG(ERROR) << "DataSync PersistKV failed"; + } + } + + if (!succ) + { + // A flush batch can contain work from several checkpoint attempts. + // Preserve each status's first failing stage; terminal finalization + // later emits at most one failure for that checkpoint status. + for (auto &[_, entries] : flush_task_entries) + { + for (auto &entry : entries) + { + if (entry->data_sync_task_ && entry->data_sync_task_->status_) + { + entry->data_sync_task_->status_->RecordCheckpointFailure( + checkpoint_failure_reason); + } + } + } } // Record that data was written in DataSyncStatus if flush succeeded. diff --git a/tx_service/src/checkpointer.cpp b/tx_service/src/checkpointer.cpp index a3e5b3d61..2a2161c42 100644 --- a/tx_service/src/checkpointer.cpp +++ b/tx_service/src/checkpointer.cpp @@ -155,20 +155,171 @@ std::pair Checkpointer::GetNewCheckpointTs( return {ckpt_ts, ckpt_req.GetMemUsage()}; } -void Checkpointer::CollectCkptStallMetric(uint32_t node_group, uint32_t rounds) +bool Checkpointer::IsCurrentCheckpointTerm(NodeGroupId node_group_id, + int64_t term) const +{ + if (term < 0) + { + return false; + } + + Sharder &sharder = Sharder::Instance(); + if (sharder.LeaderTerm(node_group_id) == term || + sharder.CandidateLeaderTerm(node_group_id) == term) + { + return true; + } + + return node_group_id == sharder.NativeNodeGroup() && + (sharder.StandbyNodeTerm() == term || + sharder.CandidateStandbyNodeTerm() == term); +} + +void Checkpointer::ApplyCheckpointMetricsUpdateLocked( + const CheckpointMetricsState::Update &update, + const metrics::Name *interval_metric) +{ + metrics::Meter *meter = local_shards_.GetNodeMeter(); + if (meter == nullptr) + { + return; + } + if (interval_metric != nullptr && update.interval_seconds_.has_value()) + { + meter->Collect(*interval_metric, *update.interval_seconds_); + } + if (update.continuous_failure_gauge_.has_value()) + { + meter->Collect(metrics::NAME_IS_CONTINUOUS_CHECKPOINT_FAILURES, + *update.continuous_failure_gauge_ ? 1 : 0); + } +} + +const char *Checkpointer::CheckpointFailureReasonLabel( + DataSyncStatus::CheckpointFailureReason reason) +{ + switch (reason) + { + case DataSyncStatus::CheckpointFailureReason::Scan: + return "scan"; + case DataSyncStatus::CheckpointFailureReason::CopyBase: + return "copy_base"; + case DataSyncStatus::CheckpointFailureReason::PutBase: + return "put_base"; + case DataSyncStatus::CheckpointFailureReason::PutArchive: + return "put_archive"; + case DataSyncStatus::CheckpointFailureReason::Persist: + return "persist"; + case DataSyncStatus::CheckpointFailureReason::Metadata: + return "metadata"; + case DataSyncStatus::CheckpointFailureReason::None: + case DataSyncStatus::CheckpointFailureReason::Unknown: + return "unknown"; + } + return "unknown"; +} + +void Checkpointer::RecordCheckpointAttempt(NodeGroupId node_group_id, + int64_t term) { if (!metrics::enable_metrics) { return; } - metrics::Meter *meter = local_shards_.GetNodeMeter(); - if (meter != nullptr) + auto now = std::chrono::steady_clock::now(); + std::lock_guard lk(checkpoint_metrics_mux_); + // Validate after taking the state mutex. OnLeaderStop invalidates the term + // before taking this mutex to erase the NG, so a delayed callback cannot + // recreate state after failover cleanup. + if (!IsCurrentCheckpointTerm(node_group_id, term)) { - meter->Collect(metrics::NAME_CHECKPOINT_STALL_ROUNDS, - static_cast(rounds), - std::to_string(node_group)); + return; } + + auto update = + checkpoint_metrics_state_.RecordAttempt(node_group_id, term, now); + ApplyCheckpointMetricsUpdateLocked( + update, &metrics::NAME_CHECKPOINT_ATTEMPT_INTERVAL_SECONDS); +} + +void Checkpointer::RecordCheckpointAdvance(NodeGroupId node_group_id, + int64_t term) +{ + if (!metrics::enable_metrics) + { + return; + } + + auto now = std::chrono::steady_clock::now(); + std::lock_guard lk(checkpoint_metrics_mux_); + if (!IsCurrentCheckpointTerm(node_group_id, term)) + { + return; + } + + auto update = + checkpoint_metrics_state_.RecordAdvance(node_group_id, term, now); + ApplyCheckpointMetricsUpdateLocked( + update, &metrics::NAME_CHECKPOINT_ADVANCE_INTERVAL_SECONDS); +} + +void Checkpointer::ReportCheckpointOutcome( + NodeGroupId node_group_id, + int64_t term, + DataSyncStatus::CheckpointOutcome outcome, + DataSyncStatus::CheckpointFailureReason failure_reason) +{ + if (!metrics::enable_metrics || + outcome == DataSyncStatus::CheckpointOutcome::Neutral || + outcome == DataSyncStatus::CheckpointOutcome::Canceled) + { + return; + } + + std::lock_guard lk(checkpoint_metrics_mux_); + // The counter is process-lifetime history, so a terminal failure that was + // finalized just before failover must not disappear merely because its + // reporting callback acquired this mutex after failover cleanup. The + // tenure-scoped streak below still requires a current term. + if (outcome == DataSyncStatus::CheckpointOutcome::Failure) + { + metrics::Meter *meter = local_shards_.GetNodeMeter(); + if (meter != nullptr) + { + meter->Collect(metrics::NAME_CHECKPOINT_FAILURES_TOTAL, + 1, + CheckpointFailureReasonLabel(failure_reason)); + } + } + + if (!IsCurrentCheckpointTerm(node_group_id, term)) + { + return; + } + + if (outcome == DataSyncStatus::CheckpointOutcome::Success) + { + ApplyCheckpointMetricsUpdateLocked( + checkpoint_metrics_state_.RecordSuccess(node_group_id, term)); + return; + } + + assert(outcome == DataSyncStatus::CheckpointOutcome::Failure); + ApplyCheckpointMetricsUpdateLocked( + checkpoint_metrics_state_.RecordFailure(node_group_id, term)); +} + +void Checkpointer::ClearCheckpointMetricsForNodeGroup(NodeGroupId node_group_id) +{ + if (!metrics::enable_metrics) + { + return; + } + + std::lock_guard lk(checkpoint_metrics_mux_); + ApplyCheckpointMetricsUpdateLocked( + checkpoint_metrics_state_.Erase(node_group_id)); } void Checkpointer::WarnIfCkptStalled(uint32_t node_group, @@ -176,13 +327,9 @@ void Checkpointer::WarnIfCkptStalled(uint32_t node_group, uint64_t last_ckpt_ts, const CkptTsCc::PinningTxInfo &pinning_tx) { - // Counted (and exported) before the flag is consulted: ckpt_stall_warn_ - // rounds gates the log line only. Gating the count on it would make - // setting the flag to 0 freeze the gauge at 0 for the whole stall. auto stall_it = ckpt_stall_states_.try_emplace(node_group).first; CkptStallState &stall = stall_it->second; uint32_t rounds = ++stall.stall_rounds_; - CollectCkptStallMetric(node_group, rounds); if (FLAGS_ckpt_stall_warn_rounds <= 0) { @@ -338,6 +485,10 @@ void Checkpointer::Ckpt(bool is_last_ckpt) continue; } + const int64_t checkpoint_term = + is_standby_node ? standby_node_term : leader_term; + RecordCheckpointAttempt(node_group, checkpoint_term); + CkptTsCc::PinningTxInfo pinning_tx; auto [ckpt_ts, mem_usage] = GetNewCheckpointTs(node_group, is_last_ckpt, pinning_tx); @@ -349,7 +500,6 @@ void Checkpointer::Ckpt(bool is_last_ckpt) WarnIfCkptStalled(node_group, ckpt_ts, last_ckpt_ts, pinning_tx); continue; } - CollectCkptStallMetric(node_group, 0); auto stall_it = ckpt_stall_states_.find(node_group); if (stall_it != ckpt_stall_states_.end()) { @@ -370,7 +520,8 @@ void Checkpointer::Ckpt(bool is_last_ckpt) std::make_shared( node_group, is_standby_node ? standby_node_term : leader_term, - true); + true, + DataSyncStatus::Origin::Checkpoint); uint64_t last_succ_ckpt_ts = UINT64_MAX; bool can_be_skipped = !is_last_ckpt; @@ -449,8 +600,12 @@ void Checkpointer::Ckpt(bool is_last_ckpt) << "Checkpoint of node group #" << node_group << " succeeded with timestamp: " << last_succ_ckpt_ts; - Sharder::Instance().UpdateNodeGroupCkptTs(node_group, - last_succ_ckpt_ts); + bool advanced = Sharder::Instance().UpdateNodeGroupCkptTs( + node_group, last_succ_ckpt_ts); + if (advanced) + { + RecordCheckpointAdvance(node_group, checkpoint_term); + } if (!is_standby_node) { @@ -464,6 +619,7 @@ void Checkpointer::Ckpt(bool is_last_ckpt) } } + std::optional checkpoint_result; { std::unique_lock task_sender_lk(status->mux_); status->all_task_started_ = true; @@ -513,8 +669,14 @@ void Checkpointer::Ckpt(bool is_last_ckpt) status->truncate_log_ts_ > last_ckpt_ts) { assert(status->truncate_log_ts_ >= ckpt_ts); - Sharder::Instance().UpdateNodeGroupCkptTs( - node_group, status->truncate_log_ts_); + bool advanced = + Sharder::Instance().UpdateNodeGroupCkptTs( + node_group, status->truncate_log_ts_); + if (advanced) + { + RecordCheckpointAdvance(node_group, + checkpoint_term); + } if (!is_standby_node) { @@ -531,9 +693,16 @@ void Checkpointer::Ckpt(bool is_last_ckpt) } } - CollectCkptMetric(status->err_code_ == CcErrorCode::NO_ERROR); + checkpoint_result = status->TryFinalizeCheckpointLocked(); } } + if (checkpoint_result.has_value()) + { + ReportCheckpointOutcome(node_group, + checkpoint_term, + checkpoint_result->outcome_, + checkpoint_result->failure_reason_); + } } } diff --git a/tx_service/src/data_sync_task.cpp b/tx_service/src/data_sync_task.cpp index 1c8e31999..d74c2386d 100644 --- a/tx_service/src/data_sync_task.cpp +++ b/tx_service/src/data_sync_task.cpp @@ -37,12 +37,36 @@ namespace txservice { +namespace +{ +bool IsCheckpointCancellation(CcErrorCode err_code) +{ + return err_code == CcErrorCode::NG_TERM_CHANGED || + err_code == CcErrorCode::REQUESTED_NODE_NOT_LEADER; +} + +void ReportCheckpointResult( + const DataSyncStatus &status, + const std::optional &result) +{ + if (result.has_value()) + { + Sharder::Instance().GetCheckpointer()->ReportCheckpointOutcome( + status.node_group_id_, + status.node_group_term_, + result->outcome_, + result->failure_reason_); + } +} +} // namespace DataSyncStatus::DataSyncStatus(NodeGroupId node_group_id, int64_t node_group_term, - bool need_truncate_log) + bool need_truncate_log, + Origin origin) : node_group_id_(node_group_id), node_group_term_(node_group_term), + origin_(origin), need_truncate_log_(need_truncate_log) { Sharder::Instance().GetCheckpointer()->IncrementOngoingDataSyncCnt(); @@ -53,6 +77,54 @@ DataSyncStatus::~DataSyncStatus() Sharder::Instance().GetCheckpointer()->DecrementOngoingDataSyncCnt(); } +void DataSyncStatus::RecordCheckpointFailure(CheckpointFailureReason reason) +{ + if (origin_ != Origin::Checkpoint || + reason == CheckpointFailureReason::None) + { + return; + } + + std::lock_guard lk(mux_); + if (checkpoint_failure_reason_ == CheckpointFailureReason::None) + { + checkpoint_failure_reason_ = reason; + } +} + +std::optional +DataSyncStatus::TryFinalizeCheckpointLocked() +{ + if (origin_ != Origin::Checkpoint || checkpoint_finalized_ || + !all_task_started_ || unfinished_tasks_ != 0) + { + return std::nullopt; + } + + checkpoint_finalized_ = true; + if (IsCheckpointCancellation(err_code_)) + { + return CheckpointResult{CheckpointOutcome::Canceled, + CheckpointFailureReason::None}; + } + if (err_code_ != CcErrorCode::NO_ERROR) + { + return CheckpointResult{ + CheckpointOutcome::Failure, + checkpoint_failure_reason_ == CheckpointFailureReason::None + ? CheckpointFailureReason::Unknown + : checkpoint_failure_reason_}; + } + if (no_truncate_reason_ == NoTruncateReason::Deduplicated || + no_truncate_reason_ == NoTruncateReason::EntriesSkipped) + { + return CheckpointResult{CheckpointOutcome::Neutral, + CheckpointFailureReason::None}; + } + return CheckpointResult{CheckpointOutcome::Success, + CheckpointFailureReason::None}; +} + DataSyncTask::DataSyncTask(const TableName &table_name, uint32_t ng_id, int64_t ng_term, @@ -114,6 +186,7 @@ DataSyncTask::DataSyncTask(const TableName &table_name, void DataSyncTask::SetFinish() { std::unique_lock task_sender_lk(status_->mux_); + std::optional checkpoint_result; status_->unfinished_tasks_--; // The default value of `truncate_log_ts_` is `0`. if (status_->truncate_log_ts_ == 0) @@ -140,8 +213,16 @@ void DataSyncTask::SetFinish() << status_->truncate_log_ts_; if (status_->truncate_log_ts_ != UINT64_MAX) { - Sharder::Instance().UpdateNodeGroupCkptTs( + bool advanced = Sharder::Instance().UpdateNodeGroupCkptTs( node_group_id_, status_->truncate_log_ts_); + if (advanced && + status_->origin_ == DataSyncStatus::Origin::Checkpoint) + { + Sharder::Instance() + .GetCheckpointer() + ->RecordCheckpointAdvance(node_group_id_, + node_group_term_); + } if (!txservice_skip_wal) { @@ -180,8 +261,7 @@ void DataSyncTask::SetFinish() } } - Sharder::Instance().GetCheckpointer()->CollectCkptMetric( - status_->err_code_ == CcErrorCode::NO_ERROR); + checkpoint_result = status_->TryFinalizeCheckpointLocked(); if (task_res_) { @@ -196,13 +276,25 @@ void DataSyncTask::SetFinish() } status_->cv_.notify_all(); } + task_sender_lk.unlock(); + ReportCheckpointResult(*status_, checkpoint_result); } -void DataSyncTask::SetError(CcErrorCode err_code) +void DataSyncTask::SetError( + CcErrorCode err_code, + DataSyncStatus::CheckpointFailureReason checkpoint_failure_reason) { std::unique_lock task_sender_lk(status_->mux_); + std::optional checkpoint_result; status_->unfinished_tasks_--; status_->err_code_ = err_code; + if (status_->origin_ == DataSyncStatus::Origin::Checkpoint && + !IsCheckpointCancellation(err_code) && + status_->checkpoint_failure_reason_ == + DataSyncStatus::CheckpointFailureReason::None) + { + status_->checkpoint_failure_reason_ = checkpoint_failure_reason; + } // The default value of `truncate_log_ts_` is `0`. if (status_->truncate_log_ts_ == 0) { @@ -218,12 +310,15 @@ void DataSyncTask::SetError(CcErrorCode err_code) if (status_->unfinished_tasks_ == 0 && status_->all_task_started_) { + checkpoint_result = status_->TryFinalizeCheckpointLocked(); if (task_res_) { task_res_->SetError(status_->err_code_); } status_->cv_.notify_all(); } + task_sender_lk.unlock(); + ReportCheckpointResult(*status_, checkpoint_result); } void DataSyncTask::SetScanTaskFinished() diff --git a/tx_service/src/fault/cc_node.cpp b/tx_service/src/fault/cc_node.cpp index e0858aa40..675ae02a9 100644 --- a/tx_service/src/fault/cc_node.cpp +++ b/tx_service/src/fault/cc_node.cpp @@ -35,6 +35,7 @@ #include "cc_node_service.h" #include "cc_req_misc.h" #include "cc_request.pb.h" +#include "checkpointer.h" #include "local_cc_shards.h" #include "metrics.h" #include "sharder.h" @@ -363,6 +364,12 @@ bool CcNode::OnLeaderStart(int64_t term, // no longer subscribed to previous term Sharder::Instance().SetStandbyNodeTerm(-1); Sharder::Instance().SetCandidateStandbyNodeTerm(-1); + // Invalidate both standby term caches before erasing the native NG. + // A delayed checkpoint callback then revalidates to false while + // holding the metrics-state mutex and cannot recreate this tenure. + Sharder::Instance() + .GetCheckpointer() + ->ClearCheckpointMetricsForNodeGroup(ng_id_); Sharder::Instance().SetStandbyBecomingLeaderNodeTerm(-1); } @@ -498,6 +505,12 @@ bool CcNode::OnLeaderStop(int64_t term) Sharder::Instance().SetLeaderTerm(ng_id_, -1); Sharder::Instance().SetCandidateTerm(ng_id_, -1); } + // The term caches are invalidated before the per-NG state is erased. A + // delayed checkpoint callback revalidates the term while holding the same + // metrics-state mutex and therefore cannot recreate this leadership + // tenure after cleanup. + Sharder::Instance().GetCheckpointer()->ClearCheckpointMetricsForNodeGroup( + ng_id_); if (!txservice_skip_kv) { @@ -861,8 +874,16 @@ void CcNode::SubscribePrimaryNode(uint32_t leader_node_id, // clean old term ccm cache since this node was following on an // older term + bool had_active_standby_term = + Sharder::Instance().StandbyNodeTerm() > 0; Sharder::Instance().SetStandbyNodeTerm(-1); Sharder::Instance().SetCandidateStandbyNodeTerm(-1); + if (had_active_standby_term) + { + Sharder::Instance() + .GetCheckpointer() + ->ClearCheckpointMetricsForNodeGroup(ng_id_); + } if (resubscribe) { @@ -1179,14 +1200,22 @@ void CcNode::SubscribePrimaryNode(uint32_t leader_node_id, return false; } + bool cleared_active_standby_term = false; if (Sharder::Instance().StandbyNodeTerm() == standby_term) { Sharder::Instance().SetStandbyNodeTerm(-1); + cleared_active_standby_term = true; } if (Sharder::Instance().CandidateStandbyNodeTerm() == standby_term) { Sharder::Instance().SetCandidateStandbyNodeTerm(-1); } + if (cleared_active_standby_term) + { + Sharder::Instance() + .GetCheckpointer() + ->ClearCheckpointMetricsForNodeGroup(ng_id_); + } return true; }; diff --git a/tx_service/src/remote/cc_node_service.cpp b/tx_service/src/remote/cc_node_service.cpp index 9aa340871..6e820044f 100644 --- a/tx_service/src/remote/cc_node_service.cpp +++ b/tx_service/src/remote/cc_node_service.cpp @@ -803,7 +803,8 @@ void CcNodeService::FlushDataAll(::google::protobuf::RpcController *controller, uint64_t table_last_synced_ts = 0; std::shared_ptr status = - std::make_shared(ng_id, ng_term, false); + std::make_shared( + ng_id, ng_term, false, DataSyncStatus::Origin::FlushData); local_shards.EnqueueDataSyncTaskForTable(table_name, ng_id, diff --git a/tx_service/src/store/snapshot_manager.cpp b/tx_service/src/store/snapshot_manager.cpp index fdc47dfcc..45b2a2057 100644 --- a/tx_service/src/store/snapshot_manager.cpp +++ b/tx_service/src/store/snapshot_manager.cpp @@ -1240,8 +1240,10 @@ bool SnapshotManager::RunOneRoundCheckpoint(uint32_t node_group, local_shards.GetCatalogTableNameSnapshot(node_group, UINT64_MAX); std::shared_ptr data_sync_status = - std::make_shared(node_group, ng_leader_term, true); - data_sync_status->SetNoTruncateLog(); + std::make_shared( + node_group, ng_leader_term, true, DataSyncStatus::Origin::Snapshot); + data_sync_status->SetNoTruncateLog( + DataSyncStatus::NoTruncateReason::NotCheckpoint); bool can_be_skipped = false; uint64_t last_ckpt_ts = Sharder::Instance().GetNodeGroupCkptTs(node_group); diff --git a/tx_service/src/tx_index_operation.cpp b/tx_service/src/tx_index_operation.cpp index 1c8520207..f95f8be75 100644 --- a/tx_service/src/tx_index_operation.cpp +++ b/tx_service/src/tx_index_operation.cpp @@ -1001,8 +1001,8 @@ void UpsertTableIndexOp::Forward(TransactionExecution *txm) is_last_finished_key_str_ = false; finished_pk_range_count_ = scanned_pk_range_count_; LOG(INFO) << "Alter Table Index transaction write prepare index" - << " log with last finished end key: " - << ". Base table: " << table_key_.Name().StringView() + << " log with last finished end key: " << ". Base table: " + << table_key_.Name().StringView() << ". Txn: " << txm->TxNumber(); op_ = &prepare_data_log_op_; FillPrepareDataLogRequest(txm); @@ -1930,7 +1930,8 @@ void UpsertTableIndexOp::FlushDataIntoDataStore(const TableName &table_name, } assert(ng_term > 0); uint64_t table_last_synced_ts = 0; - auto status = std::make_shared(ng_id, ng_term, false); + auto status = std::make_shared( + ng_id, ng_term, false, DataSyncStatus::Origin::CreateIndex); local_cc_shards->EnqueueDataSyncTaskForTable(table_name, ng_id, ng_term, @@ -2590,9 +2591,9 @@ void UpsertTableIndexOp::HandleRangeTask( // Asynchronous mode stub.GenerateSkFromPk(cntl_ptr, req_ptr, resp_ptr, closure); DLOG(INFO) << "Acquire GenerateSkFromPk service for partition id: " - << partition_id << " with start key: " - << " and end key: " - << " of ng#" << range_owner; + << partition_id + << " with start key: " << " and end key: " << " of ng#" + << range_owner; } { diff --git a/tx_service/tests/CMakeLists.txt b/tx_service/tests/CMakeLists.txt index 0625df677..cec3ae450 100644 --- a/tx_service/tests/CMakeLists.txt +++ b/tx_service/tests/CMakeLists.txt @@ -62,6 +62,7 @@ set(CATCH_MAIN_TESTS CcPage-Test LargeObjLRU-Test CcRequestWait-Test + CheckpointMetricsState-Test NonBlockingLock-Test AcquireAllError-Test StandbyForward-Test diff --git a/tx_service/tests/CheckpointMetricsState-Test.cpp b/tx_service/tests/CheckpointMetricsState-Test.cpp new file mode 100644 index 000000000..8091189d7 --- /dev/null +++ b/tx_service/tests/CheckpointMetricsState-Test.cpp @@ -0,0 +1,66 @@ +#include +#include + +#include "checkpoint_metrics_state.h" + +using namespace std::chrono_literals; +using txservice::CheckpointMetricsState; + +TEST_CASE("checkpoint intervals are independent per NG and term", + "[checkpoint-metrics]") +{ + CheckpointMetricsState state(3); + const auto start = CheckpointMetricsState::TimePoint{}; + + auto first_attempt = state.RecordAttempt(1, 10, start); + REQUIRE_FALSE(first_attempt.interval_seconds_.has_value()); + auto second_attempt = state.RecordAttempt(1, 10, start + 90s); + REQUIRE(second_attempt.interval_seconds_ == 90.0); + + auto first_advance = state.RecordAdvance(1, 10, start + 10s); + REQUIRE_FALSE(first_advance.interval_seconds_.has_value()); + auto second_advance = state.RecordAdvance(1, 10, start + 130s); + REQUIRE(second_advance.interval_seconds_ == 120.0); + + // A second NG has its own anchors, and a new term resets both anchors. + REQUIRE_FALSE( + state.RecordAttempt(2, 20, start + 200s).interval_seconds_.has_value()); + REQUIRE_FALSE( + state.RecordAttempt(1, 11, start + 200s).interval_seconds_.has_value()); + REQUIRE_FALSE( + state.RecordAdvance(1, 11, start + 200s).interval_seconds_.has_value()); +} + +TEST_CASE("continuous checkpoint failure state aggregates and erases by NG", + "[checkpoint-metrics]") +{ + CheckpointMetricsState state(3); + + REQUIRE_FALSE( + state.RecordFailure(1, 10).continuous_failure_gauge_.has_value()); + REQUIRE_FALSE( + state.RecordFailure(1, 10).continuous_failure_gauge_.has_value()); + REQUIRE(state.RecordFailure(1, 10).continuous_failure_gauge_ == true); + REQUIRE(state.ConsecutiveFailures(1) == 3); + + state.RecordFailure(2, 20); + state.RecordFailure(2, 20); + REQUIRE(state.RecordFailure(2, 20).continuous_failure_gauge_ == true); + + // Clearing one breached NG leaves the node alert set while another is + // still breached. A success only mutates its own NG. + REQUIRE(state.RecordSuccess(1, 10).continuous_failure_gauge_ == true); + REQUIRE(state.ConsecutiveFailures(1) == 0); + REQUIRE(state.Erase(2).continuous_failure_gauge_ == false); + REQUIRE_FALSE(state.Contains(2)); + + // A thresholded old term cannot leak into a new leadership tenure. + state.RecordFailure(1, 10); + state.RecordFailure(1, 10); + REQUIRE(state.RecordFailure(1, 10).continuous_failure_gauge_ == true); + auto new_term = + state.RecordAttempt(1, 11, CheckpointMetricsState::TimePoint{} + 1s); + REQUIRE(new_term.continuous_failure_gauge_ == false); + REQUIRE(state.ConsecutiveFailures(1) == 0); + REQUIRE_FALSE(new_term.interval_seconds_.has_value()); +} From d258420b796a845737d4e0f053c4f04aa9f95fc4 Mon Sep 17 00:00:00 2001 From: liunyl Date: Sat, 29 Aug 2026 13:01:56 +0000 Subject: [PATCH 2/3] fix(metrics): address observability review feedback --- docs/07-durability-and-recovery.md | 10 +++-- eloq_metrics/include/metrics.h | 6 ++- eloq_metrics/src/prometheus_collector.cc | 15 ++++++- eloq_metrics/tests/metrics_collector_test.cc | 41 ++++++++++++++++++- tx_service/include/checkpoint_metrics_state.h | 9 ++++ tx_service/include/tx_service.h | 4 +- tx_service/src/fault/cc_node.cpp | 5 +++ .../tests/CheckpointMetricsState-Test.cpp | 31 ++++++++------ 8 files changed, 98 insertions(+), 23 deletions(-) diff --git a/docs/07-durability-and-recovery.md b/docs/07-durability-and-recovery.md index dc702fcc0..f03e1174b 100644 --- a/docs/07-durability-and-recovery.md +++ b/docs/07-durability-and-recovery.md @@ -154,10 +154,12 @@ exported through a node meter without `ng_id` or `core_id`: Successful and genuine no-work checkpoints clear only their NG's consecutive failure streak. Coalesced/skipped outcomes, stalls, and term cancellations are -neutral. `CcNode::OnLeaderStop` erases that NG's streak and timing anchors after -invalidating its term, while the cumulative failure counter remains. Metric -callbacks revalidate the term while holding the same state mutex used for -cleanup, so a callback from the old term cannot recreate erased state. +neutral. Leader stop and active-standby subscription teardown erase that NG's +streak and timing anchors after invalidating its term, while the cumulative +failure counter remains. Candidate standbys do not checkpoint and therefore do +not own metric state to erase. Metric callbacks revalidate the term while +holding the same state mutex used for cleanup, so a callback from the old term +cannot recreate erased state. ## 4. Data sync beyond checkpointing (overview) diff --git a/eloq_metrics/include/metrics.h b/eloq_metrics/include/metrics.h index 0c29fb670..70be1253f 100644 --- a/eloq_metrics/include/metrics.h +++ b/eloq_metrics/include/metrics.h @@ -42,7 +42,7 @@ inline bool enable_metrics = false; class Name { public: - Name(std::string name) : name_(std::move(name)) {}; + Name(std::string name) : name_(std::move(name)){}; const std::string &GetName() const { @@ -55,7 +55,9 @@ class Name using Clock = std::chrono::steady_clock; using Labels = std::vector>; -/** Explicit upper bounds for one histogram; empty selects collector defaults. +/** + * Explicit upper bounds for one histogram. Non-empty bounds must be strictly + * increasing; an empty sequence selects collector defaults. */ using HistogramBuckets = std::vector; using TimePoint = decltype(Clock::now()); diff --git a/eloq_metrics/src/prometheus_collector.cc b/eloq_metrics/src/prometheus_collector.cc index f359bf657..46ae42aa3 100644 --- a/eloq_metrics/src/prometheus_collector.cc +++ b/eloq_metrics/src/prometheus_collector.cc @@ -28,6 +28,7 @@ #include #include #include +#include #include "metrics.h" @@ -142,12 +143,22 @@ MetricHandle PrometheusCollector::SetMetric(std::unique_ptr &metric_ptr) { case Type::Histogram: { + const auto &configured_buckets = metric_ptr->histogram_buckets_; + if (std::adjacent_find(configured_buckets.begin(), + configured_buckets.end(), + [](double lower, double upper) { + return !(lower < upper); + }) != configured_buckets.end()) + { + throw std::invalid_argument( + "Histogram bucket bounds must be strictly increasing"); + } auto &histogram_family = prometheus::BuildHistogram() .Name(metric_ptr->name_) .Register(*registry_); - const auto &buckets = metric_ptr->histogram_buckets_.empty() + const auto &buckets = configured_buckets.empty() ? PROMETHEUS_HISTOGRAM_DEF_BUCKETS - : metric_ptr->histogram_buckets_; + : configured_buckets; auto &histogram = histogram_family.Add(prometheus_labels, buckets); data = std::make_shared(histogram); break; diff --git a/eloq_metrics/tests/metrics_collector_test.cc b/eloq_metrics/tests/metrics_collector_test.cc index 034df6e26..1abc07686 100644 --- a/eloq_metrics/tests/metrics_collector_test.cc +++ b/eloq_metrics/tests/metrics_collector_test.cc @@ -19,10 +19,16 @@ * . * */ -#include +// clang-format off +#include +#include +#include #include #include +#include +// clang-format on + #include "meter.h" #include "prometheus_collector.h" @@ -208,3 +214,36 @@ SCENARIO("Histograms can override default buckets", "[HistogramBuckets]") REQUIRE(default_sample.histogram.bucket.size() == metrics::PROMETHEUS_HISTOGRAM_DEF_BUCKETS.size() + 1); } + +SCENARIO("Invalid histogram buckets are rejected before registration", + "[HistogramBuckets]") +{ + metrics::PrometheusCollector collector{"0.0.0.0", 18084}; + REQUIRE(collector.Open()); + + metrics::Metric descending_metric{"recoverable_bucket_histogram", + metrics::Type::Histogram, + {}, + {5.0, 1.0}}; + auto descending_metric_ptr = + std::make_unique(descending_metric); + REQUIRE_THROWS_AS(collector.SetMetric(descending_metric_ptr), + std::invalid_argument); + + // Reusing the name proves validation happened before family registration. + metrics::Metric recovered_metric{"recoverable_bucket_histogram", + metrics::Type::Histogram, + {}, + {1.0, 5.0}}; + auto recovered_metric_ptr = + std::make_unique(recovered_metric); + auto recovered_handle = collector.SetMetric(recovered_metric_ptr); + REQUIRE(recovered_handle.collector_data != nullptr); + + metrics::Metric duplicate_metric{ + "duplicate_bucket_histogram", metrics::Type::Histogram, {}, {1.0, 1.0}}; + auto duplicate_metric_ptr = + std::make_unique(duplicate_metric); + REQUIRE_THROWS_AS(collector.SetMetric(duplicate_metric_ptr), + std::invalid_argument); +} diff --git a/tx_service/include/checkpoint_metrics_state.h b/tx_service/include/checkpoint_metrics_state.h index 1e9bdc901..936fee64a 100644 --- a/tx_service/include/checkpoint_metrics_state.h +++ b/tx_service/include/checkpoint_metrics_state.h @@ -131,12 +131,21 @@ class CheckpointMetricsState return update; } + /** + * Returns the retained failure streak, or zero when the NG is absent. + * Erase removes the NG; the first event for a new term replaces the old + * tenure before applying that event. + */ size_t ConsecutiveFailures(NodeGroupId node_group_id) const { auto it = states_.find(node_group_id); return it == states_.end() ? 0 : it->second.consecutive_failures_; } + /** + * Returns false when the NG is absent or has been erased. A term-changing + * event replaces the old tenure with new state, so the NG remains present. + */ bool Contains(NodeGroupId node_group_id) const { return states_.find(node_group_id) != states_.end(); diff --git a/tx_service/include/tx_service.h b/tx_service/include/tx_service.h index 0bdea682b..bfcdb684c 100644 --- a/tx_service/include/tx_service.h +++ b/tx_service/include/tx_service.h @@ -1049,7 +1049,9 @@ class TxProcessor size_t busy_round_active_tx_count_{0}; size_t empty_round_count_{0}; size_t total_round_count_{0}; - size_t empty_round_threshold_{1000}; + // Accumulate empty-round samples on every loop, but publish their ratio + // once per 10,000 rounds to smooth loop-level noise and reduce metric I/O. + size_t empty_round_threshold_{10000}; // tx_current_round_ is only utilized for sampling the tx_duration and // remote request metric. diff --git a/tx_service/src/fault/cc_node.cpp b/tx_service/src/fault/cc_node.cpp index 675ae02a9..845d300d8 100644 --- a/tx_service/src/fault/cc_node.cpp +++ b/tx_service/src/fault/cc_node.cpp @@ -874,6 +874,9 @@ void CcNode::SubscribePrimaryNode(uint32_t leader_node_id, // clean old term ccm cache since this node was following on an // older term + // Candidate standbys return from Ckpt() before metrics are + // recorded. Only an active standby can own tenure state that this + // subscription transition must erase. bool had_active_standby_term = Sharder::Instance().StandbyNodeTerm() > 0; Sharder::Instance().SetStandbyNodeTerm(-1); @@ -1200,6 +1203,8 @@ void CcNode::SubscribePrimaryNode(uint32_t leader_node_id, return false; } + // A candidate standby has not checkpointed yet. Rollback therefore + // erases metrics only when it clears a promoted, active standby term. bool cleared_active_standby_term = false; if (Sharder::Instance().StandbyNodeTerm() == standby_term) { diff --git a/tx_service/tests/CheckpointMetricsState-Test.cpp b/tx_service/tests/CheckpointMetricsState-Test.cpp index 8091189d7..433e19cbc 100644 --- a/tx_service/tests/CheckpointMetricsState-Test.cpp +++ b/tx_service/tests/CheckpointMetricsState-Test.cpp @@ -1,9 +1,11 @@ -#include +// clang-format off #include +#include +// clang-format on + #include "checkpoint_metrics_state.h" -using namespace std::chrono_literals; using txservice::CheckpointMetricsState; TEST_CASE("checkpoint intervals are independent per NG and term", @@ -14,21 +16,24 @@ TEST_CASE("checkpoint intervals are independent per NG and term", auto first_attempt = state.RecordAttempt(1, 10, start); REQUIRE_FALSE(first_attempt.interval_seconds_.has_value()); - auto second_attempt = state.RecordAttempt(1, 10, start + 90s); + auto second_attempt = + state.RecordAttempt(1, 10, start + std::chrono::seconds{90}); REQUIRE(second_attempt.interval_seconds_ == 90.0); - auto first_advance = state.RecordAdvance(1, 10, start + 10s); + auto first_advance = + state.RecordAdvance(1, 10, start + std::chrono::seconds{10}); REQUIRE_FALSE(first_advance.interval_seconds_.has_value()); - auto second_advance = state.RecordAdvance(1, 10, start + 130s); + auto second_advance = + state.RecordAdvance(1, 10, start + std::chrono::seconds{130}); REQUIRE(second_advance.interval_seconds_ == 120.0); // A second NG has its own anchors, and a new term resets both anchors. - REQUIRE_FALSE( - state.RecordAttempt(2, 20, start + 200s).interval_seconds_.has_value()); - REQUIRE_FALSE( - state.RecordAttempt(1, 11, start + 200s).interval_seconds_.has_value()); - REQUIRE_FALSE( - state.RecordAdvance(1, 11, start + 200s).interval_seconds_.has_value()); + REQUIRE_FALSE(state.RecordAttempt(2, 20, start + std::chrono::seconds{200}) + .interval_seconds_.has_value()); + REQUIRE_FALSE(state.RecordAttempt(1, 11, start + std::chrono::seconds{200}) + .interval_seconds_.has_value()); + REQUIRE_FALSE(state.RecordAdvance(1, 11, start + std::chrono::seconds{200}) + .interval_seconds_.has_value()); } TEST_CASE("continuous checkpoint failure state aggregates and erases by NG", @@ -58,8 +63,8 @@ TEST_CASE("continuous checkpoint failure state aggregates and erases by NG", state.RecordFailure(1, 10); state.RecordFailure(1, 10); REQUIRE(state.RecordFailure(1, 10).continuous_failure_gauge_ == true); - auto new_term = - state.RecordAttempt(1, 11, CheckpointMetricsState::TimePoint{} + 1s); + auto new_term = state.RecordAttempt( + 1, 11, CheckpointMetricsState::TimePoint{} + std::chrono::seconds{1}); REQUIRE(new_term.continuous_failure_gauge_ == false); REQUIRE(state.ConsecutiveFailures(1) == 0); REQUIRE_FALSE(new_term.interval_seconds_.has_value()); From 97ddec47a0ff072abec1cd65d37c3afde729e783 Mon Sep 17 00:00:00 2001 From: liunyl Date: Sat, 29 Aug 2026 13:03:29 +0000 Subject: [PATCH 3/3] style(metrics): match CI clang-format output --- eloq_metrics/include/metrics.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/eloq_metrics/include/metrics.h b/eloq_metrics/include/metrics.h index 70be1253f..6ed90b573 100644 --- a/eloq_metrics/include/metrics.h +++ b/eloq_metrics/include/metrics.h @@ -42,7 +42,7 @@ inline bool enable_metrics = false; class Name { public: - Name(std::string name) : name_(std::move(name)){}; + Name(std::string name) : name_(std::move(name)) {}; const std::string &GetName() const {