| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| #pragma once |
| |
| #include <algorithm> |
| #include <atomic> |
| #include <functional> |
| #include <iostream> |
| #include <list> |
| #include <memory> |
| #include <string> |
| #include <thread> |
| #include <utility> |
| #include <vector> |
| |
| #include <glog/logging.h> |
| |
| #include "kudu/client/client-test-util.h" |
| #include "kudu/client/client.h" |
| #include "kudu/client/row_result.h" |
| #include "kudu/client/value.h" |
| #include "kudu/client/write_op.h" |
| #include "kudu/clock/hybrid_clock.h" |
| #include "kudu/gutil/port.h" |
| #include "kudu/gutil/stl_util.h" |
| #include "kudu/gutil/strings/join.h" |
| #include "kudu/gutil/strings/split.h" |
| #include "kudu/gutil/strings/substitute.h" |
| #include "kudu/gutil/walltime.h" |
| #include "kudu/mini-cluster/external_mini_cluster.h" |
| #include "kudu/tablet/tablet.h" |
| #include "kudu/util/blocking_queue.h" |
| #include "kudu/util/curl_util.h" |
| #include "kudu/util/hdr_histogram.h" |
| #include "kudu/util/monotime.h" |
| #include "kudu/util/random.h" |
| #include "kudu/util/stopwatch.h" |
| |
| namespace kudu { |
| |
| static const char* const kKeyColumnName = "rand_key"; |
| static const char* const kLinkColumnName = "link_to"; |
| static const char* const kInsertTsColumnName = "insert_ts"; |
| static const char* const kUpdatedColumnName = "updated"; |
| static const int64_t kNoSnapshot = -2; |
| static const int64_t kSnapshotAtNow = -1; |
| static const int64_t kNoParticularCountExpected = -1; |
| |
| // Vector of snapshot timestamp, count pairs. |
| typedef std::vector<std::pair<uint64_t, int64_t> > SnapsAndCounts; |
| |
| // Provides methods for writing data and reading it back in such a way that |
| // facilitates checking for data integrity. |
| class LinkedListTester { |
| public: |
| LinkedListTester(client::sp::shared_ptr<client::KuduClient> client, |
| std::string table_name, int num_chains, int num_tablets, |
| int num_replicas, bool enable_mutation) |
| : verify_projection_( |
| {kKeyColumnName, kLinkColumnName, kUpdatedColumnName}), |
| table_name_(std::move(table_name)), |
| num_chains_(num_chains), |
| num_tablets_(num_tablets), |
| num_replicas_(num_replicas), |
| enable_mutation_(enable_mutation), |
| latency_histogram_(1000000, 3), |
| client_(std::move(client)) { |
| client::KuduSchemaBuilder b; |
| |
| b.AddColumn(kKeyColumnName)->Type(client::KuduColumnSchema::INT64)->NotNull()->PrimaryKey(); |
| b.AddColumn(kLinkColumnName)->Type(client::KuduColumnSchema::INT64)->NotNull(); |
| b.AddColumn(kInsertTsColumnName)->Type(client::KuduColumnSchema::INT64)->NotNull(); |
| b.AddColumn(kUpdatedColumnName)->Type(client::KuduColumnSchema::BOOL)->NotNull() |
| ->Default(client::KuduValue::FromBool(false)); |
| CHECK_OK(b.Build(&schema_)); |
| } |
| |
| // The modes for WaitAndVerify |
| enum WaitAndVerifyMode { |
| // Perform snapshots scans in the past but finish with a "READ_LATEST" scan. |
| // This should be used when the a majority of the cluster is down. |
| FINISH_WITH_SCAN_LATEST, |
| // Perform the snapshot scans in the past and finish with a snapshot scan in the present. |
| FINISH_WITH_SNAPSHOT_SCAN |
| }; |
| |
| // Create the table. |
| Status CreateLinkedListTable(); |
| |
| // Load the table with the linked list test pattern. |
| // |
| // Runs for the amount of time designated by 'run_for'. |
| // Sets *written_count to the number of rows inserted. |
| Status LoadLinkedList( |
| const MonoDelta& run_for, |
| int num_samples, |
| int64_t *written_count); |
| |
| // Variant of VerifyLinkedListRemote that verifies at the specified snapshot timestamp. |
| Status VerifyLinkedListAtSnapshotRemote(uint64_t snapshot_timestamp, |
| int64_t expected, |
| bool log_errors, |
| const std::function<Status(const std::string&)>& cb, |
| int64_t* verified_count) { |
| return VerifyLinkedListRemote(snapshot_timestamp, |
| expected, |
| log_errors, |
| cb, |
| verified_count); |
| } |
| |
| Status VerifyLinkedListAtLatestRemote(int64_t expected, |
| bool log_errors, |
| const std::function<Status(const std::string&)>& cb, |
| int64_t* verified_count) { |
| return VerifyLinkedListRemote(kNoSnapshot, |
| expected, |
| log_errors, |
| cb, |
| verified_count); |
| } |
| |
| // Variant of VerifyLinkedListremote that verifies at 'now'. |
| Status VerifyLinkedListAtNowSnapshotRemote(int64_t expected, |
| bool log_errors, |
| const std::function<Status(const std::string&)>& cb, |
| int64_t* verified_count) { |
| return VerifyLinkedListRemote(kSnapshotAtNow, |
| expected, |
| log_errors, |
| cb, |
| verified_count); |
| } |
| |
| // Run the verify step on a table with RPCs. Calls the provided callback 'cb' once during |
| // verification to test scanner fault tolerance. |
| Status VerifyLinkedListRemote(uint64_t snapshot_timestamp, |
| int64_t expected, |
| bool log_errors, |
| const std::function<Status(const std::string&)>& cb, |
| int64_t* verified_count); |
| |
| // Run the verify step on a specific tablet. |
| Status VerifyLinkedListLocal(const tablet::Tablet* tablet, |
| int64_t expected, |
| int64_t* verified_count); |
| |
| // A variant of VerifyLinkedListRemote that is more robust towards ongoing |
| // bootstrapping and replication. |
| Status WaitAndVerify(int seconds_to_run, |
| int64_t expected, |
| WaitAndVerifyMode mode = FINISH_WITH_SNAPSHOT_SCAN) { |
| return WaitAndVerify(seconds_to_run, |
| expected, |
| [](const std::string& /*uuid*/) { return Status::OK(); }, |
| mode); |
| } |
| |
| // A variant of WaitAndVerify that also takes a callback to be run once during verification. |
| Status WaitAndVerify(int seconds_to_run, |
| int64_t expected, |
| const std::function<Status(const std::string&)>& cb, |
| WaitAndVerifyMode mode = FINISH_WITH_SNAPSHOT_SCAN); |
| |
| // Generates a vector of keys for the table such that each tablet is |
| // responsible for an equal fraction of the int64 key space. |
| std::vector<const KuduPartialRow*> GenerateSplitRows(const client::KuduSchema& schema); |
| |
| // Generate a vector of ints which form the split keys. |
| std::vector<int64_t> GenerateSplitInts(); |
| |
| void DumpInsertHistogram(bool print_flags); |
| |
| protected: |
| client::KuduSchema schema_; |
| const std::vector<std::string> verify_projection_; |
| const std::string table_name_; |
| const int num_chains_; |
| const int num_tablets_; |
| const int num_replicas_; |
| const bool enable_mutation_; |
| HdrHistogram latency_histogram_; |
| client::sp::shared_ptr<client::KuduClient> client_; |
| SnapsAndCounts sampled_timestamps_and_counts_; |
| }; |
| |
| // Generates the linked list pattern. |
| // Since we can insert multiple chain in parallel, this encapsulates the |
| // state for each chain. |
| class LinkedListChainGenerator { |
| public: |
| // 'chain_idx' is a unique ID for this chain. Chains with different indexes |
| // will always generate distinct sets of keys (thus avoiding the possibility of |
| // a collision even in a longer run). |
| ATTRIBUTE_NO_SANITIZE_INTEGER |
| explicit LinkedListChainGenerator(int chain_idx) |
| : chain_idx_(chain_idx), |
| rand_(chain_idx * 0xDEADBEEF), |
| prev_key_(0) { |
| CHECK_GE(chain_idx, 0); |
| CHECK_LT(chain_idx, 65536); |
| } |
| |
| ~LinkedListChainGenerator() { |
| } |
| |
| // Generate a random 64-bit unsigned int. |
| uint64_t Rand64() { |
| return (implicit_cast<uint64_t>(rand_.Next()) << 32) | rand_.Next(); |
| } |
| |
| Status GenerateNextInsert(client::KuduTable* table, client::KuduSession* session) { |
| // Encode the chain index in the lowest 16 bits so that different chains never |
| // intersect. |
| int64_t this_key = (Rand64() << 16) | chain_idx_; |
| int64_t ts = GetCurrentTimeMicros(); |
| |
| std::unique_ptr<client::KuduInsert> insert(table->NewInsert()); |
| CHECK_OK(insert->mutable_row()->SetInt64(kKeyColumnName, this_key)); |
| CHECK_OK(insert->mutable_row()->SetInt64(kInsertTsColumnName, ts)); |
| CHECK_OK(insert->mutable_row()->SetInt64(kLinkColumnName, prev_key_)); |
| RETURN_NOT_OK_PREPEND(session->Apply(insert.release()), |
| strings::Substitute("Unable to apply insert with key $0 at ts $1", |
| this_key, ts)); |
| prev_key_ = this_key; |
| return Status::OK(); |
| } |
| |
| int64_t prev_key() const { |
| return prev_key_; |
| } |
| |
| private: |
| const int chain_idx_; |
| |
| // This is a linear congruential random number generator, so it won't repeat until |
| // it has exhausted its period (which is quite large) |
| Random rand_; |
| |
| // The previously output key. |
| int64_t prev_key_; |
| |
| DISALLOW_COPY_AND_ASSIGN(LinkedListChainGenerator); |
| }; |
| |
| // A thread that updates the timestamps of rows whose keys are put in its BlockingQueue. |
| class ScopedRowUpdater { |
| public: |
| |
| // Create and start a new ScopedUpdater. 'table' must remain valid for |
| // the lifetime of this object. |
| explicit ScopedRowUpdater(client::KuduTable* table) |
| : table_(table), |
| to_update_(kint64max), // no limit |
| updater_([this]() { this->RowUpdaterThread(); }) { |
| } |
| |
| ~ScopedRowUpdater() { |
| to_update_.Shutdown(); |
| updater_.join(); |
| } |
| |
| BlockingQueue<int64_t>* to_update() { return &to_update_; } |
| |
| private: |
| void RowUpdaterThread() { |
| client::sp::shared_ptr<client::KuduSession> session(table_->client()->NewSession()); |
| session->SetTimeoutMillis(60000 /* 60 seconds */); |
| CHECK_OK(session->SetFlushMode(client::KuduSession::AUTO_FLUSH_BACKGROUND)); |
| |
| int64_t next_key = 0; |
| while (to_update_.BlockingGet(&next_key).ok()) { |
| std::unique_ptr<client::KuduUpdate> update(table_->NewUpdate()); |
| CHECK_OK(update->mutable_row()->SetInt64(kKeyColumnName, next_key)); |
| CHECK_OK(update->mutable_row()->SetBool(kUpdatedColumnName, true)); |
| CHECK_OK(session->Apply(update.release())); |
| } |
| |
| FlushSessionOrDie(session); |
| } |
| |
| client::KuduTable* table_; |
| BlockingQueue<int64_t> to_update_; |
| std::thread updater_; |
| }; |
| |
| // Helper class to hold results from a linked list scan and perform the |
| // verification step on the data. |
| class LinkedListVerifier { |
| public: |
| LinkedListVerifier(int num_chains, bool enable_mutation, int64_t expected, |
| std::vector<int64_t> split_key_ints); |
| |
| // Start the scan timer. The duration between starting the scan and verifying |
| // the data is logged in the VerifyData() step, so this should be called |
| // immediately before starting the table(t) scan. |
| void StartScanTimer(); |
| |
| // Register a new row result during the verify step. |
| void RegisterResult(int64_t key, int64_t link, bool updated); |
| |
| // Run the common verify step once the scanned data is stored. |
| Status VerifyData(int64_t* verified_count, bool log_errors); |
| |
| private: |
| // Print a summary of the broken links to the log. |
| void SummarizeBrokenLinks(const std::vector<int64_t>& broken_links); |
| |
| const int num_chains_; |
| const int64_t expected_; |
| const bool enable_mutation_; |
| const std::vector<int64_t> split_key_ints_; |
| std::vector<int64_t> seen_key_; |
| std::vector<int64_t> seen_link_to_; |
| int errors_; |
| Stopwatch scan_timer_; |
| }; |
| |
| ///////////////////////////////////////////////////////////// |
| // LinkedListTester |
| ///////////////////////////////////////////////////////////// |
| |
| inline std::vector<const KuduPartialRow*> LinkedListTester::GenerateSplitRows( |
| const client::KuduSchema& schema) { |
| std::vector<const KuduPartialRow*> split_keys; |
| for (int64_t val : GenerateSplitInts()) { |
| KuduPartialRow* row = schema.NewRow(); |
| CHECK_OK(row->SetInt64(kKeyColumnName, val)); |
| split_keys.push_back(row); |
| } |
| return split_keys; |
| } |
| |
| inline std::vector<int64_t> LinkedListTester::GenerateSplitInts() { |
| std::vector<int64_t> ret; |
| ret.reserve(num_tablets_ - 1); |
| const int64_t increment = kint64max / num_tablets_; |
| for (int64_t i = 1; i < num_tablets_; i++) { |
| ret.push_back(i * increment); |
| } |
| return ret; |
| } |
| |
| inline Status LinkedListTester::CreateLinkedListTable() { |
| std::unique_ptr<client::KuduTableCreator> table_creator( |
| client_->NewTableCreator()); |
| #pragma GCC diagnostic push |
| #pragma GCC diagnostic ignored "-Wdeprecated-declarations" |
| RETURN_NOT_OK_PREPEND(table_creator->table_name(table_name_) |
| .schema(&schema_) |
| .set_range_partition_columns({ kKeyColumnName }) |
| .split_rows(GenerateSplitRows(schema_)) |
| .num_replicas(num_replicas_) |
| .Create(), |
| "Failed to create table"); |
| #pragma GCC diagnostic pop |
| return Status::OK(); |
| } |
| |
| inline Status LinkedListTester::LoadLinkedList( |
| const MonoDelta& run_for, |
| int num_samples, |
| int64_t *written_count) { |
| |
| sampled_timestamps_and_counts_.clear(); |
| client::sp::shared_ptr<client::KuduTable> table; |
| RETURN_NOT_OK_PREPEND(client_->OpenTable(table_name_, &table), |
| "Could not open table " + table_name_); |
| |
| MonoTime start = MonoTime::Now(); |
| MonoTime deadline = start + run_for; |
| |
| client::sp::shared_ptr<client::KuduSession> session(client_->NewSession()); |
| session->SetTimeoutMillis(60000 /* 60 seconds */); |
| RETURN_NOT_OK_PREPEND(session->SetFlushMode(client::KuduSession::MANUAL_FLUSH), |
| "Couldn't set flush mode"); |
| |
| ScopedRowUpdater updater(table.get()); |
| std::vector<std::unique_ptr<LinkedListChainGenerator>> chains; |
| chains.reserve(num_chains_); |
| for (int i = 0; i < num_chains_; i++) { |
| chains.emplace_back(new LinkedListChainGenerator(i)); |
| } |
| |
| MonoDelta sample_interval = MonoDelta::FromMicroseconds(run_for.ToMicroseconds() / num_samples); |
| MonoTime next_sample = start + sample_interval; |
| LOG(INFO) << "Running for: " << run_for.ToString(); |
| LOG(INFO) << "Sampling every " << sample_interval.ToMicroseconds() << " us"; |
| |
| *written_count = 0; |
| int iter = 0; |
| while (true) { |
| if (iter++ % 10000 == 0) { |
| LOG(INFO) << "Written " << (*written_count) << " rows in chain"; |
| DumpInsertHistogram(false); |
| } |
| |
| MonoTime now = MonoTime::Now(); |
| if (next_sample < now) { |
| Timestamp now(client_->GetLatestObservedTimestamp()); |
| sampled_timestamps_and_counts_.emplace_back(now.ToUint64() + 1, *written_count); |
| next_sample += sample_interval; |
| LOG(INFO) << "Sample at HT timestamp: " << now.ToString() |
| << " Inserted count: " << *written_count; |
| } |
| if (deadline < now) { |
| LOG(INFO) << "Finished inserting list. Added " << (*written_count) << " in chain"; |
| LOG(INFO) << "Last entries inserted had keys:"; |
| for (int i = 0; i < num_chains_; i++) { |
| LOG(INFO) << i << ": " << chains[i]->prev_key(); |
| } |
| return Status::OK(); |
| } |
| for (const auto& chain : chains) { |
| RETURN_NOT_OK_PREPEND(chain->GenerateNextInsert(table.get(), session.get()), |
| "Unable to generate next insert into linked list chain"); |
| } |
| |
| MonoTime flush_start(MonoTime::Now()); |
| FlushSessionOrDie(session); |
| MonoDelta elapsed = MonoTime::Now() - flush_start; |
| latency_histogram_.Increment(elapsed.ToMicroseconds()); |
| |
| (*written_count) += chains.size(); |
| |
| if (enable_mutation_) { |
| // Rows have been inserted; they're now safe to update. |
| for (const auto& chain : chains) { |
| updater.to_update()->Put(chain->prev_key()); |
| } |
| } |
| } |
| } |
| |
| inline void LinkedListTester::DumpInsertHistogram(bool print_flags) { |
| // We dump to cout instead of using glog so the output isn't prefixed with |
| // line numbers. This makes it less ugly to copy-paste into JIRA, etc. |
| using std::cout; |
| using std::endl; |
| |
| const HdrHistogram* h = &latency_histogram_; |
| |
| cout << "------------------------------------------------------------" << endl; |
| cout << "Histogram for latency of insert operations (microseconds)" << endl; |
| if (print_flags) { |
| cout << "Flags: " << google::CommandlineFlagsIntoString() << endl; |
| } |
| cout << "Note: each insert is a batch of " << num_chains_ << " rows." << endl; |
| cout << "------------------------------------------------------------" << endl; |
| h->DumpHumanReadable(&std::cout); |
| } |
| |
| // Verify that the given sorted vector does not contain any duplicate entries. |
| // If it does, *errors will be incremented once per duplicate and the given message |
| // will be logged. |
| static void VerifyNoDuplicateEntries(const std::vector<int64_t>& ints, int* errors, |
| const std::string& message) { |
| for (int i = 1; i < ints.size(); i++) { |
| if (ints[i] == ints[i - 1]) { |
| LOG(ERROR) << message << ": " << ints[i]; |
| (*errors)++; |
| } |
| } |
| } |
| |
| inline Status LinkedListTester::VerifyLinkedListRemote( |
| uint64_t snapshot_timestamp, int64_t expected, bool log_errors, |
| const std::function<Status(const std::string&)>& cb, int64_t* verified_count) { |
| |
| client::sp::shared_ptr<client::KuduTable> table; |
| RETURN_NOT_OK(client_->OpenTable(table_name_, &table)); |
| |
| std::string snapshot_str; |
| if (snapshot_timestamp == kSnapshotAtNow) { |
| snapshot_str = "NOW"; |
| } else { |
| snapshot_str = clock::HybridClock::StringifyTimestamp(Timestamp(snapshot_timestamp)); |
| } |
| |
| client::KuduScanner scanner(table.get()); |
| RETURN_NOT_OK_PREPEND(scanner.SetProjectedColumnNames(verify_projection_), "Bad projection"); |
| RETURN_NOT_OK(scanner.SetBatchSizeBytes(0)); // Force at least one NextBatch RPC. |
| RETURN_NOT_OK(scanner.SetTimeoutMillis(60 * 1000 /* 60 seconds */)); |
| |
| if (snapshot_timestamp != kNoSnapshot) { |
| RETURN_NOT_OK(scanner.SetReadMode(client::KuduScanner::READ_AT_SNAPSHOT)); |
| RETURN_NOT_OK(scanner.SetFaultTolerant()); |
| if (snapshot_timestamp != kSnapshotAtNow) { |
| RETURN_NOT_OK(scanner.SetSnapshotRaw(snapshot_timestamp)); |
| } |
| } else { |
| RETURN_NOT_OK(scanner.SetReadMode(client::KuduScanner::READ_LATEST)); |
| } |
| |
| LOG(INFO) << "Verifying Snapshot: " << snapshot_str << " Expected Rows: " << expected; |
| |
| RETURN_NOT_OK_PREPEND(scanner.Open(), "Couldn't open scanner"); |
| |
| RETURN_NOT_OK(scanner.SetBatchSizeBytes(1024)); // More normal batch size. |
| |
| LinkedListVerifier verifier(num_chains_, enable_mutation_, expected, |
| GenerateSplitInts()); |
| verifier.StartScanTimer(); |
| |
| bool cb_called = false; |
| while (scanner.HasMoreRows()) { |
| // If we're doing a snapshot scan with a big enough cluster, call the callback on the scanner's |
| // tserver. Do this only once. |
| if (snapshot_timestamp != kSnapshotAtNow && !cb_called) { |
| client::KuduTabletServer* kts_ptr; |
| RETURN_NOT_OK(scanner.GetCurrentServer(&kts_ptr)); |
| std::unique_ptr<client::KuduTabletServer> kts(kts_ptr); |
| const auto& down_ts = kts->uuid(); |
| LOG(INFO) << "Calling callback on tserver " << down_ts; |
| RETURN_NOT_OK(cb(down_ts)); |
| cb_called = true; |
| } |
| client::KuduScanBatch batch; |
| RETURN_NOT_OK_PREPEND(scanner.NextBatch(&batch), "Couldn't fetch next row batch"); |
| for (const client::KuduScanBatch::RowPtr row : batch) { |
| int64_t key; |
| int64_t link; |
| bool updated; |
| RETURN_NOT_OK(row.GetInt64(0, &key)); |
| RETURN_NOT_OK(row.GetInt64(1, &link)); |
| |
| // For non-snapshot reads we also verify that all rows were updated. We don't |
| // for snapshot reads as updates are performed by their own thread. This means |
| // that there is no guarantee that, for any snapshot timestamp that comes before |
| // all writes are completed, all rows will be updated. |
| if (snapshot_timestamp == kSnapshotAtNow) { |
| RETURN_NOT_OK(row.GetBool(2, &updated)); |
| } else { |
| updated = enable_mutation_; |
| } |
| |
| verifier.RegisterResult(key, link, updated); |
| } |
| } |
| |
| Status s = verifier.VerifyData(verified_count, log_errors); |
| LOG(INFO) << "Snapshot: " << snapshot_str << " verified. Result: " << s.ToString(); |
| return s; |
| } |
| |
| inline Status LinkedListTester::VerifyLinkedListLocal( |
| const tablet::Tablet* tablet, int64_t expected, int64_t* verified_count) { |
| DCHECK(tablet); |
| LinkedListVerifier verifier(num_chains_, enable_mutation_, expected, |
| GenerateSplitInts()); |
| verifier.StartScanTimer(); |
| |
| const Schema* tablet_schema = tablet->schema(); |
| // Cannot use schemas with col indexes in a scan (assertions fire). |
| Schema projection(tablet_schema->columns(), tablet_schema->num_key_columns()); |
| std::unique_ptr<RowwiseIterator> iter; |
| RETURN_NOT_OK_PREPEND(tablet->NewRowIterator(projection, &iter), |
| "Cannot create new row iterator"); |
| RETURN_NOT_OK_PREPEND(iter->Init(nullptr), "Cannot initialize row iterator"); |
| |
| RowBlockMemory mem(1024); |
| RowBlock block(&projection, 100, &mem); |
| while (iter->HasNext()) { |
| mem.Reset(); |
| RETURN_NOT_OK(iter->NextBlock(&block)); |
| for (int i = 0; i < block.nrows(); i++) { |
| int64_t key; |
| int64_t link; |
| bool updated; |
| |
| const RowBlockRow& row = block.row(i); |
| key = *tablet_schema->ExtractColumnFromRow<INT64>(row, 0); |
| link = *tablet_schema->ExtractColumnFromRow<INT64>(row, 1); |
| updated = *tablet_schema->ExtractColumnFromRow<BOOL>(row, 3); |
| |
| verifier.RegisterResult(key, link, updated); |
| } |
| } |
| |
| return verifier.VerifyData(verified_count, true); |
| } |
| |
| inline Status LinkedListTester::WaitAndVerify( |
| int seconds_to_run, int64_t expected, |
| const std::function<Status(const std::string&)>& cb, WaitAndVerifyMode mode) { |
| |
| std::list<std::pair<int64_t, int64_t>> samples_as_list( |
| sampled_timestamps_and_counts_.begin(), |
| sampled_timestamps_and_counts_.end()); |
| |
| int64_t seen = 0; |
| bool called = false; |
| Stopwatch sw; |
| sw.start(); |
| |
| Status s; |
| do { |
| // We'll give the tablets 5 seconds to start up regardless of how long we |
| // inserted for. There's some fixed cost startup time, especially when |
| // replication is enabled. |
| const int kBaseTimeToWaitSecs = 5; |
| bool last_attempt = sw.elapsed().wall_seconds() > kBaseTimeToWaitSecs + seconds_to_run; |
| s = Status::OK(); |
| auto iter = samples_as_list.begin(); |
| |
| while (iter != samples_as_list.end()) { |
| // Only call the callback once, on the first verify pass, since it may be destructive. |
| if (iter == samples_as_list.begin() && !called) { |
| s = VerifyLinkedListAtSnapshotRemote((*iter).first, (*iter).second, last_attempt, cb, |
| &seen); |
| called = true; |
| } else { |
| s = VerifyLinkedListAtSnapshotRemote( |
| (*iter).first, (*iter).second, last_attempt, |
| [](const std::string& /*uuid*/) { return Status::OK(); }, |
| &seen); |
| } |
| |
| if (s.ok() && (*iter).second != seen) { |
| // If we've seen less rows than we were expecting we should fail and not retry. |
| // |
| // The reasoning is the following: |
| // |
| // - We know that when we read this snapshot's timestamp the writes had completed, thus |
| // at timestamp '(*iter).first' any replica should have precisely '(*iter).second' rows. |
| // - We also chose to perform a snapshot scan, which, when passed a timestamp, waits for |
| // that timestamp to become "clean", i.e. it makes sure that all transactions with lower |
| // timestamps have completed before it actually performs the scan. |
| // |
| // Together these conditions mean that if we don't get the expected rows back something |
| // is wrong with the read path or with the write path and we should fail immediately. |
| return Status::Corruption(strings::Substitute("Got wrong row count on snapshot. " |
| "Expected: $0, Got:$1", (*iter).second, seen)); |
| } |
| |
| if (!s.ok()) break; |
| // If the snapshot verification returned OK erase it so that we don't recheck |
| // even if a later snapshot or the final verification failed. |
| iter = samples_as_list.erase(iter); |
| } |
| |
| // Perform the last scan with the required mode. |
| switch (mode) { |
| case FINISH_WITH_SNAPSHOT_SCAN: |
| if (s.ok()) { |
| RETURN_NOT_OK(VerifyLinkedListAtNowSnapshotRemote( |
| expected, last_attempt, |
| [](const std::string& /*uuid*/) { return Status::OK(); }, &seen)); |
| } |
| break; |
| case FINISH_WITH_SCAN_LATEST: |
| // Scans in READ_LATEST mode will, by design, likely return a stale view of the tablet |
| // so, in this case, retry. |
| if (s.ok()) { |
| s = VerifyLinkedListAtLatestRemote( |
| expected, last_attempt, |
| [](const std::string& /*uuid*/) { return Status::OK(); }, &seen); |
| } |
| break; |
| } |
| |
| if (!s.ok()) { |
| KLOG_EVERY_N(INFO, 10) << "Table not yet ready: " << seen << "/" |
| << expected << " rows (status: " |
| << s.ToString() << ")"; |
| if (last_attempt) { |
| // We'll give it an equal amount of time to re-load the data as it took |
| // to write it in. Typically it completes much faster than that. |
| return Status::TimedOut("Timed out waiting for table to be accessible again", |
| s.ToString()); |
| } |
| |
| // Sleep and retry until timeout. |
| SleepFor(MonoDelta::FromMilliseconds(20)); |
| } |
| } while (!s.ok()); |
| |
| LOG(INFO) << "Successfully verified " << expected << " rows"; |
| |
| return Status::OK(); |
| } |
| |
| ///////////////////////////////////////////////////////////// |
| // LinkedListVerifier |
| ///////////////////////////////////////////////////////////// |
| |
| LinkedListVerifier::LinkedListVerifier(int num_chains, bool enable_mutation, |
| int64_t expected, |
| std::vector<int64_t> split_key_ints) |
| : num_chains_(num_chains), |
| expected_(expected), |
| enable_mutation_(enable_mutation), |
| split_key_ints_(std::move(split_key_ints)), |
| errors_(0) { |
| if (expected != kNoParticularCountExpected) { |
| DCHECK_GE(expected, 0); |
| seen_key_.reserve(expected); |
| seen_link_to_.reserve(expected); |
| } |
| } |
| |
| void LinkedListVerifier::StartScanTimer() { |
| scan_timer_.start(); |
| } |
| |
| void LinkedListVerifier::RegisterResult(int64_t key, int64_t link, bool updated) { |
| seen_key_.push_back(key); |
| if (link != 0) { |
| // Links to entry 0 don't count - the first inserts use this link |
| seen_link_to_.push_back(link); |
| } |
| |
| if (updated != enable_mutation_) { |
| LOG(ERROR) << "Entry " << key << " was incorrectly " |
| << (enable_mutation_ ? "not " : "") << "updated"; |
| errors_++; |
| } |
| } |
| |
| void LinkedListVerifier::SummarizeBrokenLinks(const std::vector<int64_t>& broken_links) { |
| std::vector<int64_t> errors_by_tablet(split_key_ints_.size() + 1); |
| |
| int n_logged = 0; |
| const int kMaxToLog = 100; |
| |
| for (int64_t broken : broken_links) { |
| int tablet = std::upper_bound(split_key_ints_.begin(), |
| split_key_ints_.end(), |
| broken) - split_key_ints_.begin(); |
| DCHECK_GE(tablet, 0); |
| DCHECK_LT(tablet, errors_by_tablet.size()); |
| errors_by_tablet[tablet]++; |
| |
| if (n_logged < kMaxToLog) { |
| LOG(ERROR) << "Entry " << broken << " was linked to but not present"; |
| n_logged++; |
| if (n_logged == kMaxToLog) { |
| LOG(ERROR) << "... no more broken links will be logged"; |
| } |
| } |
| } |
| |
| // Summarize the broken links by which tablet they fell into. |
| if (!broken_links.empty()) { |
| for (int tablet = 0; tablet < errors_by_tablet.size(); tablet++) { |
| LOG(ERROR) << "Error count for tablet #" << tablet << ": " << errors_by_tablet[tablet]; |
| } |
| } |
| } |
| |
| Status LinkedListVerifier::VerifyData(int64_t* verified_count, bool log_errors) { |
| *verified_count = seen_key_.size(); |
| LOG(INFO) << "Done collecting results (" << (*verified_count) << " rows in " |
| << scan_timer_.elapsed().wall_millis() << "ms)"; |
| |
| VLOG(1) << "Sorting results before verification of linked list structure..."; |
| std::sort(seen_key_.begin(), seen_key_.end()); |
| std::sort(seen_link_to_.begin(), seen_link_to_.end()); |
| VLOG(1) << "Done sorting"; |
| |
| // Verify that no key was seen multiple times or linked to multiple times |
| VerifyNoDuplicateEntries(seen_key_, &errors_, "Seen row key multiple times"); |
| VerifyNoDuplicateEntries(seen_link_to_, &errors_, "Seen link to row multiple times"); |
| // Verify that every key that was linked to was present |
| std::vector<int64_t> broken_links = STLSetDifference(seen_link_to_, seen_key_); |
| errors_ += broken_links.size(); |
| if (log_errors) { |
| SummarizeBrokenLinks(broken_links); |
| } |
| |
| // Verify that only the expected number of keys were seen but not linked to. |
| // Only the last "batch" should have this characteristic. |
| std::vector<int64_t> not_linked_to = STLSetDifference(seen_key_, seen_link_to_); |
| if (not_linked_to.size() != num_chains_) { |
| LOG_IF(ERROR, log_errors) |
| << "Had " << not_linked_to.size() << " entries which were seen but not" |
| << " linked to. Expected only " << num_chains_; |
| errors_++; |
| } |
| |
| if (errors_ > 0) { |
| return Status::Corruption("Had one or more errors during verification (see log)"); |
| } |
| |
| if (expected_ != *verified_count) { |
| return Status::IllegalState(strings::Substitute( |
| "Missing rows, but with no broken link in the chain. This means that " |
| "a suffix of the inserted rows went missing. Expected=$0, seen=$1.", |
| expected_, *verified_count)); |
| } |
| |
| return Status::OK(); |
| } |
| |
| } // namespace kudu |
