YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

// 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.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed 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.

//

#include "yb/tserver/ts_tablet_manager.h"

#include <algorithm>

#include <chrono>

#include <memory>

#include <mutex>

#include <string>

#include <thread>

#include <vector>

#include <boost/container/static_vector.hpp>

#include <boost/optional/optional.hpp>

#include <glog/logging.h>

#include "yb/client/client.h"

#include "yb/common/wire_protocol.h"

#include "yb/consensus/consensus.h"

#include "yb/consensus/multi_raft_batcher.h"

#include "yb/consensus/consensus_meta.h"

#include "yb/consensus/log.h"

#include "yb/consensus/log_anchor_registry.h"

#include "yb/consensus/metadata.pb.h"

#include "yb/consensus/opid_util.h"

#include "yb/consensus/quorum_util.h"

#include "yb/consensus/raft_consensus.h"

#include "yb/consensus/retryable_requests.h"

#include "yb/consensus/state_change_context.h"

#include "yb/docdb/docdb_rocksdb_util.h"

#include "yb/fs/fs_manager.h"

#include "yb/gutil/bind.h"

#include "yb/gutil/stl_util.h"

#include "yb/gutil/strings/substitute.h"

#include "yb/gutil/sysinfo.h"

#include "yb/master/master_heartbeat.pb.h"

#include "yb/master/sys_catalog.h"

#include "yb/rpc/messenger.h"

#include "yb/rpc/poller.h"

#include "yb/tablet/metadata.pb.h"

#include "yb/tablet/operations/split_operation.h"

#include "yb/tablet/tablet.h"

#include "yb/tablet/tablet.pb.h"

#include "yb/tablet/tablet_bootstrap_if.h"

#include "yb/tablet/tablet_metadata.h"

#include "yb/tablet/tablet_options.h"

#include "yb/tablet/tablet_peer.h"

#include "yb/tserver/heartbeater.h"

#include "yb/tserver/remote_bootstrap_client.h"

#include "yb/tserver/remote_bootstrap_session.h"

#include "yb/tserver/tablet_server.h"

#include "yb/tserver/tserver.pb.h"

#include "yb/util/debug/long_operation_tracker.h"

#include "yb/util/debug/trace_event.h"

#include "yb/util/env.h"

#include "yb/util/fault_injection.h"

#include "yb/util/flag_tags.h"

#include "yb/util/format.h"

#include "yb/util/logging.h"

#include "yb/util/mem_tracker.h"

#include "yb/util/metrics.h"

#include "yb/util/pb_util.h"

#include "yb/util/scope_exit.h"

#include "yb/util/shared_lock.h"

#include "yb/util/status_format.h"

#include "yb/util/status_log.h"

#include "yb/util/stopwatch.h"

#include "yb/util/trace.h"

using namespace std::literals;

using namespace std::placeholders;

DEFINE_int32(num_tablets_to_open_simultaneously, 0,

             "Number of threads available to open tablets during startup. If this "

             "is set to 0 (the default), then the number of bootstrap threads will "

             "be set based on the number of data directories. If the data directories "

             "are on some very fast storage device such as SSD or a RAID array, it "

             "may make sense to manually tune this.");

TAG_FLAG(num_tablets_to_open_simultaneously, advanced);

DEFINE_int32(tablet_start_warn_threshold_ms, 500,

             "If a tablet takes more than this number of millis to start, issue "

             "a warning with a trace.");

TAG_FLAG(tablet_start_warn_threshold_ms, hidden);

DEFINE_int32(cleanup_split_tablets_interval_sec, 60,

             "Interval at which tablet manager tries to cleanup split tablets which are no longer "

             "needed. Setting this to 0 disables cleanup of split tablets.");

DEFINE_test_flag(double, fault_crash_after_blocks_deleted, 0.0,

                 "Fraction of the time when the tablet will crash immediately "

                 "after deleting the data blocks during tablet deletion.");

DEFINE_test_flag(double, fault_crash_after_wal_deleted, 0.0,

                 "Fraction of the time when the tablet will crash immediately "

                 "after deleting the WAL segments during tablet deletion.");

DEFINE_test_flag(double, fault_crash_after_cmeta_deleted, 0.0,

                 "Fraction of the time when the tablet will crash immediately "

                 "after deleting the consensus metadata during tablet deletion.");

DEFINE_test_flag(double, fault_crash_after_rb_files_fetched, 0.0,

                 "Fraction of the time when the tablet will crash immediately "

                 "after fetching the files during a remote bootstrap but before "

                 "marking the superblock as TABLET_DATA_READY.");

DEFINE_test_flag(double, fault_crash_in_split_after_log_copied, 0.0,

                 "Fraction of the time when the tablet will crash immediately after initiating a "

                 "Log::CopyTo from parent to child tablet, but before marking the child tablet as "

                 "TABLET_DATA_READY.");

DEFINE_test_flag(bool, simulate_already_present_in_remote_bootstrap, false,

                 "If true, return an AlreadyPresent error in remote bootstrap after starting the "

                 "remote bootstrap client.");

DEFINE_test_flag(double, fault_crash_in_split_before_log_flushed, 0.0,

                 "Fraction of the time when the tablet will crash immediately before flushing a "

                 "parent tablet's kSplit operation.");

DEFINE_test_flag(uint64, crash_if_remote_bootstrap_sessions_greater_than, 0,

                 "If greater than zero, this process will crash if we detect more than the "

                 "specified number of remote bootstrap sessions.");

DEFINE_test_flag(uint64, crash_if_remote_bootstrap_sessions_per_table_greater_than, 0,

                 "If greater than zero, this process will crash if for any table we exceed the "

                 "specified number of remote bootstrap sessions");

DEFINE_test_flag(bool, crash_before_apply_tablet_split_op, false,

                 "Crash inside TSTabletManager::ApplyTabletSplit before doing anything");

DEFINE_test_flag(bool, force_single_tablet_failure, false,

                 "Force exactly one tablet to a failed state.");

DEFINE_test_flag(int32, apply_tablet_split_inject_delay_ms, 0,

                 "Inject delay into TSTabletManager::ApplyTabletSplit.");

DEFINE_test_flag(bool, skip_deleting_split_tablets, false,

                 "Skip deleting tablets which have been split.");

DEFINE_test_flag(bool, skip_post_split_compaction, false,

                 "Skip processing post split compaction.");

DEFINE_int32(verify_tablet_data_interval_sec, 0,

             "The tick interval time for the tablet data integrity verification background task. "

             "This defaults to 0, which means disable the background task.");

DEFINE_int32(cleanup_metrics_interval_sec, 60,

             "The tick interval time for the metrics cleanup background task. "

             "If set to 0, it disables the background task.");

DEFINE_bool(skip_tablet_data_verification, false,

            "Skip checking tablet data for corruption.");

DEFINE_int32(read_pool_max_threads, 128,

             "The maximum number of threads allowed for read_pool_. This pool is used "

             "to run multiple read operations, that are part of the same tablet rpc, "

             "in parallel.");

DEFINE_int32(read_pool_max_queue_size, 128,

             "The maximum number of tasks that can be held in the queue for read_pool_. This pool "

             "is used to run multiple read operations, that are part of the same tablet rpc, "

             "in parallel.");

DEFINE_int32(post_split_trigger_compaction_pool_max_threads, 1,

             "The maximum number of threads allowed for post_split_trigger_compaction_pool_. This "

             "pool is used to run compactions on tablets after they have been split and still "

             "contain irrelevant data from the tablet they were sourced from.");

DEFINE_int32(post_split_trigger_compaction_pool_max_queue_size, 16,

             "The maximum number of tasks that can be held in the pool for "

             "post_split_trigger_compaction_pool_. This pool is used to run compactions on tablets "

             "after they have been split and still contain irrelevant data from the tablet they "

             "were sourced from.");

DEFINE_test_flag(int32, sleep_after_tombstoning_tablet_secs, 0,

                 "Whether we sleep in LogAndTombstone after calling DeleteTabletData.");

constexpr int kTServerYbClientDefaultTimeoutMs = 60 * 1000;

DEFINE_int32(tserver_yb_client_default_timeout_ms, kTServerYbClientDefaultTimeoutMs,

             "Default timeout for the YBClient embedded into the tablet server that is used "

             "for distributed transactions.");

DEFINE_bool(enable_restart_transaction_status_tablets_first, true,

            "Set to true to prioritize bootstrapping transaction status tablets first.");

DECLARE_string(rocksdb_compact_flush_rate_limit_sharing_mode);

namespace yb {

namespace tserver {

METRIC_DEFINE_coarse_histogram(server, op_apply_queue_length, "Operation Apply Queue Length",

                        MetricUnit::kTasks,

                        "Number of operations waiting to be applied to the tablet. "

                        "High queue lengths indicate that the server is unable to process "

                        "operations as fast as they are being written to the WAL.");

METRIC_DEFINE_coarse_histogram(server, op_apply_queue_time, "Operation Apply Queue Time",

                        MetricUnit::kMicroseconds,

                        "Time that operations spent waiting in the apply queue before being "

                        "processed. High queue times indicate that the server is unable to "

                        "process operations as fast as they are being written to the WAL.");

METRIC_DEFINE_coarse_histogram(server, op_apply_run_time, "Operation Apply Run Time",

                        MetricUnit::kMicroseconds,

                        "Time that operations spent being applied to the tablet. "

                        "High values may indicate that the server is under-provisioned or "

                        "that operations consist of very large batches.");

METRIC_DEFINE_coarse_histogram(server, op_read_queue_length, "Operation Read op Queue Length",

                        MetricUnit::kTasks,

                        "Number of operations waiting to be applied to the tablet. "

                            "High queue lengths indicate that the server is unable to process "

                            "operations as fast as they are being written to the WAL.");

METRIC_DEFINE_coarse_histogram(server, op_read_queue_time, "Operation Read op Queue Time",

                        MetricUnit::kMicroseconds,

                        "Time that operations spent waiting in the read queue before being "

                            "processed. High queue times indicate that the server is unable to "

                            "process operations as fast as they are being written to the WAL.");

METRIC_DEFINE_coarse_histogram(server, op_read_run_time, "Operation Read op Run Time",

                        MetricUnit::kMicroseconds,

                        "Time that operations spent being applied to the tablet. "

                            "High values may indicate that the server is under-provisioned or "

                            "that operations consist of very large batches.");

METRIC_DEFINE_coarse_histogram(server, ts_bootstrap_time, "TServer Bootstrap Time",

                        MetricUnit::kMicroseconds,

                        "Time that the tablet server takes to bootstrap all of its tablets.");

THREAD_POOL_METRICS_DEFINE(

    server, post_split_trigger_compaction_pool, "Thread pool for tablet compaction jobs.");

THREAD_POOL_METRICS_DEFINE(

    server, admin_triggered_compaction_pool, "Thread pool for tablet compaction jobs.");

using consensus::ConsensusMetadata;

using consensus::ConsensusStatePB;

using consensus::RaftConfigPB;

using consensus::RaftPeerPB;

using consensus::StartRemoteBootstrapRequestPB;

using log::Log;

using master::ReportedTabletPB;

using master::TabletReportPB;

using master::TabletReportUpdatesPB;

using std::shared_ptr;

using std::string;

using std::unordered_set;

using std::vector;

using strings::Substitute;

using tablet::BOOTSTRAPPING;

using tablet::NOT_STARTED;

using tablet::RaftGroupMetadata;

using tablet::RaftGroupMetadataPtr;

using tablet::RaftGroupStatePB;

using tablet::RUNNING;

using tablet::TABLET_DATA_COPYING;

using tablet::TABLET_DATA_DELETED;

using tablet::TABLET_DATA_INIT_STARTED;

using tablet::TABLET_DATA_READY;

using tablet::TABLET_DATA_SPLIT_COMPLETED;

using tablet::TABLET_DATA_TOMBSTONED;

using tablet::TabletDataState;

using tablet::TabletPeer;

using tablet::TabletPeerPtr;

using tablet::TabletStatusListener;

using tablet::TabletStatusPB;

constexpr int32_t kDefaultTserverBlockCacheSizePercentage = 50;

void TSTabletManager::VerifyTabletData() {

  LOG_WITH_PREFIX(INFO) << "Beginning tablet data verification checks";

  for (const TabletPeerPtr& peer : GetTabletPeers()) {

    if (peer->state() == RUNNING) {

      if (PREDICT_FALSE(FLAGS_skip_tablet_data_verification)) {

        LOG_WITH_PREFIX(INFO)

            << Format("Skipped tablet data verification check on $0", peer->tablet_id());

      } else {

        Status s = peer->tablet()->VerifyDataIntegrity();

        if (!s.ok()) {

          LOG(WARNING) << "Tablet data integrity verification failed on " << peer->tablet_id()

                       << ": " << s;

        }

      }

    }

  }

}

void TSTabletManager::CleanupOldMetrics() {

  VLOG(2) << "Cleaning up old metrics";

  metric_registry_->RetireOldMetrics();

}

TSTabletManager::TSTabletManager(FsManager* fs_manager,

                                 TabletServer* server,

                                 MetricRegistry* metric_registry)

  : fs_manager_(fs_manager),

    server_(server),

    metric_registry_(metric_registry),

    state_(MANAGER_INITIALIZING) {

  ThreadPoolMetrics metrics = {

      METRIC_op_apply_queue_length.Instantiate(server_->metric_entity()),

      METRIC_op_apply_queue_time.Instantiate(server_->metric_entity()),

      METRIC_op_apply_run_time.Instantiate(server_->metric_entity())

  };

  tablet_options_.ServerMetricEntity = server_->metric_entity();

  CHECK_OK(ThreadPoolBuilder("apply")

               .set_metrics(std::move(metrics))

               .Build(&apply_pool_));

  // This pool is shared by all replicas hosted by this server.

  //

  // Some submitted tasks use blocking IO, so we configure no upper bound on

  // the maximum number of threads in each pool (otherwise the default value of

  // "number of CPUs" may cause blocking tasks to starve other "fast" tasks).

  // However, the effective upper bound is the number of replicas as each will

  // submit its own tasks via a dedicated token.

  CHECK_OK(ThreadPoolBuilder("consensus")

               .set_min_threads(1)

               .unlimited_threads()

               .Build(&raft_pool_));

  CHECK_OK(ThreadPoolBuilder("prepare")

               .set_min_threads(1)

               .unlimited_threads()

               .Build(&tablet_prepare_pool_));

  CHECK_OK(ThreadPoolBuilder("append")

               .set_min_threads(1)

               .unlimited_threads()

               .set_idle_timeout(MonoDelta::FromMilliseconds(10000))

               .Build(&append_pool_));

  CHECK_OK(ThreadPoolBuilder("log-alloc")

               .set_min_threads(1)

               .unlimited_threads()

               .Build(&allocation_pool_));

  ThreadPoolMetrics read_metrics = {

      METRIC_op_read_queue_length.Instantiate(server_->metric_entity()),

      METRIC_op_read_queue_time.Instantiate(server_->metric_entity()),

      METRIC_op_read_run_time.Instantiate(server_->metric_entity())

  };

  CHECK_OK(ThreadPoolBuilder("read-parallel")

               .set_max_threads(FLAGS_read_pool_max_threads)

               .set_max_queue_size(FLAGS_read_pool_max_queue_size)

               .set_metrics(std::move(read_metrics))

               .Build(&read_pool_));

  CHECK_OK(ThreadPoolBuilder("tablet-split-compaction")

              .set_max_threads(FLAGS_post_split_trigger_compaction_pool_max_threads)

              .set_max_queue_size(FLAGS_post_split_trigger_compaction_pool_max_queue_size)

              .set_metrics(THREAD_POOL_METRICS_INSTANCE(

                  server_->metric_entity(), post_split_trigger_compaction_pool))

              .Build(&post_split_trigger_compaction_pool_));

  CHECK_OK(ThreadPoolBuilder("admin-compaction")

              .set_max_threads(std::max(docdb::GetGlobalRocksDBPriorityThreadPoolSize(), 0))

              .set_metrics(THREAD_POOL_METRICS_INSTANCE(

                  server_->metric_entity(), admin_triggered_compaction_pool))

              .Build(&admin_triggered_compaction_pool_));

  mem_manager_ = std::make_shared<TabletMemoryManager>(

      &tablet_options_,

      server_->mem_tracker(),

      kDefaultTserverBlockCacheSizePercentage,

      server_->metric_entity(),

      [this](){ return GetTabletPeers(); });

}

TSTabletManager::~TSTabletManager() {

}

Status TSTabletManager::Init() {

  CHECK_EQ(state(), MANAGER_INITIALIZING);

  async_client_init_.emplace(

      "tserver_client", 0 /* num_reactors */,

      FLAGS_tserver_yb_client_default_timeout_ms / 1000, server_->permanent_uuid(),

      &server_->options(), server_->metric_entity(), server_->mem_tracker(),

      server_->messenger());

  async_client_init_->AddPostCreateHook([this](client::YBClient* client) {

    auto* tserver = server();

    if (tserver != nullptr && tserver->proxy() != nullptr) {

      client->SetLocalTabletServer(tserver->permanent_uuid(), tserver->proxy(), tserver);

    }

  });

  tablet_options_.env = server_->GetEnv();

  tablet_options_.rocksdb_env = server_->GetRocksDBEnv();

  tablet_options_.listeners = server_->options().listeners;

  if (docdb::GetRocksDBRateLimiterSharingMode() == docdb::RateLimiterSharingMode::TSERVER) {

    tablet_options_.rate_limiter = docdb::CreateRocksDBRateLimiter();

  }

  // Start the threadpool we'll use to open tablets.

  // This has to be done in Init() instead of the constructor, since the

  // FsManager isn't initialized until this point.

  int max_bootstrap_threads = FLAGS_num_tablets_to_open_simultaneously;

  if (max_bootstrap_threads == 0) {

    int num_cpus = base::NumCPUs();

    if (num_cpus <= 2) {

      max_bootstrap_threads = 2;

    } else {

      max_bootstrap_threads = min(

          num_cpus - 1, narrow_cast<int>(fs_manager_->GetDataRootDirs().size()) * 8);

    }

    LOG_WITH_PREFIX(INFO) <<  "max_bootstrap_threads=" << max_bootstrap_threads;

  }

  ThreadPoolMetrics bootstrap_metrics = {

          nullptr,

          nullptr,

          METRIC_ts_bootstrap_time.Instantiate(server_->metric_entity())

  };

  RETURN_NOT_OK(ThreadPoolBuilder("tablet-bootstrap")

                .set_max_threads(max_bootstrap_threads)

                .set_metrics(std::move(bootstrap_metrics))

                .Build(&open_tablet_pool_));

  CleanupCheckpoints();

  // Search for tablets in the metadata dir.

  vector<string> tablet_ids;

  RETURN_NOT_OK(fs_manager_->ListTabletIds(&tablet_ids));

  InitLocalRaftPeerPB();

  multi_raft_manager_ = std::make_unique<consensus::MultiRaftManager>(server_->messenger(),

                                                                      &server_->proxy_cache(),

                                                                      local_peer_pb_.cloud_info());

  deque<RaftGroupMetadataPtr> metas;

  // First, load all of the tablet metadata. We do this before we start

  // submitting the actual OpenTablet() tasks so that we don't have to compete

  // for disk resources, etc, with bootstrap processes and running tablets.

  MonoTime start(MonoTime::Now());

  for (const string& tablet_id : tablet_ids) {

    RaftGroupMetadataPtr meta;

    RETURN_NOT_OK_PREPEND(OpenTabletMeta(tablet_id, &meta),

                          "Failed to open tablet metadata for tablet: " + tablet_id);

    if (PREDICT_FALSE(!CanServeTabletData(meta->tablet_data_state()))) {

      RETURN_NOT_OK(HandleNonReadyTabletOnStartup(meta));

      continue;

    }

    RegisterDataAndWalDir(

        fs_manager_, meta->table_id(), meta->raft_group_id(), meta->data_root_dir(),

        meta->wal_root_dir());

    if (FLAGS_enable_restart_transaction_status_tablets_first) {

      // Prioritize bootstrapping transaction status tablets first.

      if (meta->table_type() == TRANSACTION_STATUS_TABLE_TYPE) {

        metas.push_front(meta);

      } else {

        metas.push_back(meta);

      }

    } else {

      metas.push_back(meta);

    }

  }

  MonoDelta elapsed = MonoTime::Now().GetDeltaSince(start);

  LOG(INFO) << "Loaded metadata for " << tablet_ids.size() << " tablet in "

            << elapsed.ToMilliseconds() << " ms";

  // Now submit the "Open" task for each.

  for (const RaftGroupMetadataPtr& meta : metas) {

    scoped_refptr<TransitionInProgressDeleter> deleter;

    RETURN_NOT_OK(StartTabletStateTransition(

        meta->raft_group_id(), "opening tablet", &deleter));

    TabletPeerPtr tablet_peer = VERIFY_RESULT(CreateAndRegisterTabletPeer(meta, NEW_PEER));

    RETURN_NOT_OK(open_tablet_pool_->SubmitFunc(

        std::bind(&TSTabletManager::OpenTablet, this, meta, deleter)));

  }

  {

    std::lock_guard<RWMutex> lock(mutex_);

    state_ = MANAGER_RUNNING;

  }

  RETURN_NOT_OK(mem_manager_->Init());

  tablets_cleaner_ = std::make_unique<rpc::Poller>(

      LogPrefix(), std::bind(&TSTabletManager::CleanupSplitTablets, this));

  verify_tablet_data_poller_ = std::make_unique<rpc::Poller>(

      LogPrefix(), std::bind(&TSTabletManager::VerifyTabletData, this));

  metrics_cleaner_ = std::make_unique<rpc::Poller>(

      LogPrefix(), std::bind(&TSTabletManager::CleanupOldMetrics, this));

  return Status::OK();

}

void TSTabletManager::CleanupCheckpoints() {

  for (const auto& data_root : fs_manager_->GetDataRootDirs()) {

    auto tables_dir = JoinPathSegments(data_root, FsManager::kRocksDBDirName);

    auto tables = fs_manager_->env()->GetChildren(tables_dir, ExcludeDots::kTrue);

    if (!tables.ok()) {

      LOG_WITH_PREFIX(WARNING)

          << "Failed to get tables in " << tables_dir << ": " << tables.status();

      continue;

    }

    for (const auto& table : *tables) {

      auto table_dir = JoinPathSegments(tables_dir, table);

      auto tablets = fs_manager_->env()->GetChildren(table_dir, ExcludeDots::kTrue);

      if (!tablets.ok()) {

        LOG_WITH_PREFIX(WARNING)

            << "Failed to get tablets in " << table_dir << ": " << tablets.status();

        continue;

      }

      for (const auto& tablet : *tablets) {

        auto checkpoints_dir = JoinPathSegments(

            table_dir, tablet, RemoteBootstrapSession::kCheckpointsDir);

        if (fs_manager_->env()->FileExists(checkpoints_dir)) {

          LOG_WITH_PREFIX(INFO) << "Cleaning up checkpoints dir: " << yb::ToString(checkpoints_dir);

          auto status = fs_manager_->env()->DeleteRecursively(checkpoints_dir);

          WARN_NOT_OK(status, Format("Cleanup of checkpoints dir $0 failed", checkpoints_dir));

        }

      }

    }

  }

}

Status TSTabletManager::Start() {

  async_client_init_->Start();

  if (FLAGS_cleanup_split_tablets_interval_sec > 0) {

    tablets_cleaner_->Start(

        &server_->messenger()->scheduler(), FLAGS_cleanup_split_tablets_interval_sec * 1s);

    LOG(INFO) << "Split tablets cleanup monitor started...";

  } else {

    LOG(INFO)

        << "Split tablets cleanup is disabled by cleanup_split_tablets_interval_sec flag set to 0";

  }

  if (FLAGS_verify_tablet_data_interval_sec > 0) {

    verify_tablet_data_poller_->Start(

        &server_->messenger()->scheduler(), FLAGS_verify_tablet_data_interval_sec * 1s);

    LOG(INFO) << "Tablet data verification task started...";

  } else {

    LOG(INFO)

        << "Tablet data verification is disabled by verify_tablet_data_interval_sec flag set to 0";

  }

  if (FLAGS_cleanup_metrics_interval_sec > 0) {

    metrics_cleaner_->Start(

        &server_->messenger()->scheduler(), FLAGS_cleanup_metrics_interval_sec * 1s);

    LOG(INFO) << "Old metrics cleanup task started...";

  } else {

    LOG(INFO)

        << "Old metrics cleanup is disabled by cleanup_metrics_interval_sec flag set to 0";

  }

  return Status::OK();

}

void TSTabletManager::CleanupSplitTablets() {

  VLOG_WITH_PREFIX_AND_FUNC(3) << "looking for tablets to cleanup...";

  auto tablet_peers = GetTabletPeers();

  for (const auto& tablet_peer : tablet_peers) {

    if (tablet_peer->CanBeDeleted()) {

      const auto& tablet_id = tablet_peer->tablet_id();

      if (PREDICT_FALSE(FLAGS_TEST_skip_deleting_split_tablets)) {

        LOG_WITH_PREFIX(INFO) << Format("Skipped triggering delete of tablet $0", tablet_id);

      } else {

        LOG_WITH_PREFIX(INFO) << Format("Triggering delete of tablet $0", tablet_id);

        client().DeleteNotServingTablet(

            tablet_peer->tablet_id(), [tablet_id](const Status& status) {

              LOG(INFO) << Format("Tablet $0 deletion result: $1", tablet_id, status);

            });

      }

    }

  }

}

Status TSTabletManager::WaitForAllBootstrapsToFinish() {

  CHECK_EQ(state(), MANAGER_RUNNING);

  open_tablet_pool_->Wait();

  Status s = Status::OK();

  SharedLock<RWMutex> shared_lock(mutex_);

  for (const TabletMap::value_type& entry : tablet_map_) {

    if (entry.second->state() == tablet::FAILED) {

      if (s.ok()) {

        s = entry.second->error();

      }

    }

  }

  return s;

}

Result<scoped_refptr<TransitionInProgressDeleter>>

TSTabletManager::StartTabletStateTransitionForCreation(const TabletId& tablet_id) {

  scoped_refptr<TransitionInProgressDeleter> deleter;

  SharedLock<RWMutex> lock(mutex_);

  TRACE("Acquired tablet manager lock");

  // Sanity check that the tablet isn't already registered.

  TabletPeerPtr junk;

  if (LookupTabletUnlocked(tablet_id, &junk)) {

    return STATUS(AlreadyPresent, "Tablet already registered", tablet_id);

  }

  RETURN_NOT_OK(StartTabletStateTransition(tablet_id, "creating tablet", &deleter));

  return deleter;

}

Result<TabletPeerPtr> TSTabletManager::CreateNewTablet(

    const tablet::TableInfoPtr& table_info,

    const string& tablet_id,

    const Partition& partition,

    RaftConfigPB config,

    const bool colocated,

    const std::vector<SnapshotScheduleId>& snapshot_schedules) {

  if (state() != MANAGER_RUNNING) {

    return STATUS_FORMAT(IllegalState, "Manager is not running: $0", state());

  }

  CHECK(IsRaftConfigMember(server_->instance_pb().permanent_uuid(), config));

  for (int i = 0; i < config.peers_size(); ++i) {

    const auto& config_peer = config.peers(i);

    CHECK(config_peer.has_member_type());

  }

  // Set the initial opid_index for a RaftConfigPB to -1.

  config.set_opid_index(consensus::kInvalidOpIdIndex);

  scoped_refptr<TransitionInProgressDeleter> deleter =

      VERIFY_RESULT(StartTabletStateTransitionForCreation(tablet_id));

  // Create the metadata.

  TRACE("Creating new metadata...");

  string data_root_dir;

  string wal_root_dir;

  GetAndRegisterDataAndWalDir(

      fs_manager_, table_info->table_id, tablet_id, &data_root_dir, &wal_root_dir);

  auto create_result = RaftGroupMetadata::CreateNew(tablet::RaftGroupMetadataData {

    .fs_manager = fs_manager_,

    .table_info = table_info,

    .raft_group_id = tablet_id,

    .partition = partition,

    .tablet_data_state = TABLET_DATA_READY,

    .colocated = colocated,

    .snapshot_schedules = snapshot_schedules,

  }, data_root_dir, wal_root_dir);

  if (!create_result.ok()) {

    UnregisterDataWalDir(table_info->table_id, tablet_id, data_root_dir, wal_root_dir);

  }

  RETURN_NOT_OK_PREPEND(create_result, "Couldn't create tablet metadata")

  RaftGroupMetadataPtr meta = std::move(*create_result);

  LOG(INFO) << TabletLogPrefix(tablet_id)

            << "Created tablet metadata for table: " << table_info->table_id;

  // We must persist the consensus metadata to disk before starting a new

  // tablet's TabletPeer and Consensus implementation.

  std::unique_ptr<ConsensusMetadata> cmeta;

  RETURN_NOT_OK_PREPEND(ConsensusMetadata::Create(fs_manager_, tablet_id, fs_manager_->uuid(),

                                                  config, consensus::kMinimumTerm, &cmeta),

                        "Unable to create new ConsensusMeta for tablet " + tablet_id);

  TabletPeerPtr new_peer = VERIFY_RESULT(CreateAndRegisterTabletPeer(meta, NEW_PEER));

  // We can run this synchronously since there is nothing to bootstrap.

  RETURN_NOT_OK(

      open_tablet_pool_->SubmitFunc(std::bind(&TSTabletManager::OpenTablet, this, meta, deleter)));

  return new_peer;

}

struct TabletCreationMetaData {

  TabletId tablet_id;

  scoped_refptr<TransitionInProgressDeleter> transition_deleter;

  Partition partition;

  docdb::KeyBounds key_bounds;

  RaftGroupMetadataPtr raft_group_metadata;

};

namespace {

// Creates SplitTabletsCreationMetaData for two new tablets for `tablet` splitting based on request.

SplitTabletsCreationMetaData PrepareTabletCreationMetaDataForSplit(

    const tablet::SplitTabletRequestPB& request, const tablet::Tablet& tablet) {

  SplitTabletsCreationMetaData metas;

  const auto& split_partition_key = request.split_partition_key();

  const auto& split_encoded_key = request.split_encoded_key();

  std::shared_ptr<Partition> source_partition = tablet.metadata()->partition();

  const auto source_key_bounds = *tablet.doc_db().key_bounds;

  {

    TabletCreationMetaData meta;

    meta.tablet_id = request.new_tablet1_id();

    meta.partition = *source_partition;

    meta.key_bounds = source_key_bounds;

    meta.partition.set_partition_key_end(split_partition_key);

    meta.key_bounds.upper.Reset(split_encoded_key);

    metas.push_back(meta);

  }

  {

    TabletCreationMetaData meta;

    meta.tablet_id = request.new_tablet2_id();

    meta.partition = *source_partition;

    meta.key_bounds = source_key_bounds;

    meta.partition.set_partition_key_start(split_partition_key);

    meta.key_bounds.lower.Reset(split_encoded_key);

    metas.push_back(meta);

  }

  return metas;

}

}  // namespace

Status TSTabletManager::StartSubtabletsSplit(

    const RaftGroupMetadata& source_tablet_meta, SplitTabletsCreationMetaData* tcmetas) {

  auto* const env = fs_manager_->env();

  auto iter = tcmetas->begin();

  while (iter != tcmetas->end()) {

    const auto& subtablet_id = iter->tablet_id;

    auto transition_deleter_result = StartTabletStateTransitionForCreation(subtablet_id);

    if (transition_deleter_result.ok()) {

      iter->transition_deleter = *transition_deleter_result;

    } else if (transition_deleter_result.status().IsAlreadyPresent()) {

      // State transition for sub tablet with subtablet_id could be already registered because its

      // remote bootstrap (from already split parent tablet leader) is in progress.

      iter = tcmetas->erase(iter);

      continue;

    } else {

      return transition_deleter_result.status();

    }

    // Try to load metadata from previous not completed split.

    auto load_result = RaftGroupMetadata::Load(fs_manager_, subtablet_id);

    if (load_result.ok() && CanServeTabletData((*load_result)->tablet_data_state())) {

      // Sub tablet has been already created and ready during previous split attempt at this node or

      // as a result of remote bootstrap from another node, no need to re-create.

      iter = tcmetas->erase(iter);

      continue;

    }

    // Delete on-disk data for new tablet IDs in case it is present as a leftover from previously

    // failed tablet split attempt.

    // TODO(tsplit): add test for that.

    const auto data_dir = source_tablet_meta.GetSubRaftGroupDataDir(subtablet_id);

    if (env->FileExists(data_dir)) {

      RETURN_NOT_OK_PREPEND(

          env->DeleteRecursively(data_dir),

          Format("Unable to recursively delete data dir for tablet $0", subtablet_id));

    }

    RETURN_NOT_OK(Log::DeleteOnDiskData(

        env, subtablet_id, source_tablet_meta.GetSubRaftGroupWalDir(subtablet_id),

        fs_manager_->uuid()));

    RETURN_NOT_OK(ConsensusMetadata::DeleteOnDiskData(fs_manager_, subtablet_id));

    ++iter;

  }

  return Status::OK();

}

void TSTabletManager::CreatePeerAndOpenTablet(

    const tablet::RaftGroupMetadataPtr& meta,

    const scoped_refptr<TransitionInProgressDeleter>& deleter) {

  Status s = ResultToStatus(CreateAndRegisterTabletPeer(meta, NEW_PEER));

  if (!s.ok()) {

    s = s.CloneAndPrepend("Failed to create and register tablet peer");

    if (s.IsShutdownInProgress()) {

      // If shutdown is in progress, it is not a failure to not being able to create and register

      // tablet peer.

      LOG_WITH_PREFIX(WARNING) << s;

    } else {

      LOG_WITH_PREFIX(DFATAL) << s;

    }

    return;

  }

  s = open_tablet_pool_->SubmitFunc(std::bind(&TSTabletManager::OpenTablet, this, meta, deleter));

  if (!s.ok()) {

    LOG(DFATAL) << Format("Failed to schedule opening tablet $0: $1", meta->table_id(), s);

    return;

  }

}

Status TSTabletManager::ApplyTabletSplit(tablet::SplitOperation* operation, log::Log* raft_log) {

  if (PREDICT_FALSE(FLAGS_TEST_crash_before_apply_tablet_split_op)) {

    LOG(FATAL) << "Crashing due to FLAGS_TEST_crash_before_apply_tablet_split_op";

  }

  if (state() != MANAGER_RUNNING) {

    return STATUS_FORMAT(IllegalState, "Manager is not running: $0", state());

  }

  auto* tablet = CHECK_NOTNULL(operation->tablet());

  const auto tablet_id = tablet->tablet_id();

  const auto* request = operation->request();

  SCHECK_EQ(

      request->tablet_id(), tablet_id, IllegalState,

      Format(

          "Unexpected SPLIT_OP $0 designated for tablet $1 to be applied to tablet $2",

          operation->op_id(), request->tablet_id(), tablet_id));

  SCHECK(

      tablet_id != request->new_tablet1_id() && tablet_id != request->new_tablet2_id(),

      IllegalState,

      Format(

          "One of SPLIT_OP $0 destination tablet IDs ($1, $2) is the same as source tablet ID $3",

          operation->op_id(), request->new_tablet1_id(), request->new_tablet2_id(), tablet_id));

  LOG_WITH_PREFIX(INFO) << "Tablet " << tablet_id << " split operation apply started";

  if (raft_log == nullptr) {

    auto tablet_peer = VERIFY_RESULT(LookupTablet(tablet_id));

    raft_log = tablet_peer->raft_consensus()->log().get();

  }

  MAYBE_FAULT(FLAGS_TEST_fault_crash_in_split_before_log_flushed);

  RETURN_NOT_OK(raft_log->FlushIndex());

  auto& meta = *CHECK_NOTNULL(tablet->metadata());

  // TODO(tsplit): We can later implement better per-disk distribution during compaction of split

  // tablets.

  const auto table_id = meta.table_id();

  const auto data_root_dir =

      VERIFY_RESULT(GetAssignedRootDirForTablet(TabletDirType::kData, table_id, tablet_id));

  const auto wal_root_dir =

      VERIFY_RESULT(GetAssignedRootDirForTablet(TabletDirType::kWal, table_id, tablet_id));

  if (FLAGS_TEST_apply_tablet_split_inject_delay_ms > 0) {

    LOG(INFO) << "TEST: ApplyTabletSplit: injecting delay of "

              << FLAGS_TEST_apply_tablet_split_inject_delay_ms << " ms for "

              << AsString(*operation);

    std::this_thread::sleep_for(FLAGS_TEST_apply_tablet_split_inject_delay_ms * 1ms);

    LOG(INFO) << "TEST: ApplyTabletSplit: delay finished";

  }

  auto tcmetas = PrepareTabletCreationMetaDataForSplit(*request, *tablet);

  RETURN_NOT_OK(StartSubtabletsSplit(meta, &tcmetas));

  for (const auto& tcmeta : tcmetas) {

    RegisterDataAndWalDir(fs_manager_, table_id, tcmeta.tablet_id, data_root_dir, wal_root_dir);

  }

  bool successfully_completed = false;

  auto se = ScopeExit([&] {

    if (!successfully_completed) {

      for (const auto& tcmeta : tcmetas) {

        UnregisterDataWalDir(table_id, tcmeta.tablet_id, data_root_dir, wal_root_dir);

      }

    }

  });

  std::unique_ptr<ConsensusMetadata> cmeta;

  RETURN_NOT_OK(ConsensusMetadata::Load(fs_manager_, tablet_id, fs_manager_->uuid(), &cmeta));

  for (auto& tcmeta : tcmetas) {

    const auto& new_tablet_id = tcmeta.tablet_id;

    // Copy raft group metadata.

    tcmeta.raft_group_metadata = VERIFY_RESULT(tablet->CreateSubtablet(

        new_tablet_id, tcmeta.partition, tcmeta.key_bounds, operation->op_id(),

        operation->hybrid_time()));

    LOG_WITH_PREFIX(INFO) << "Created raft group metadata for table: " << table_id

                          << " tablet: " << new_tablet_id;

    // Copy consensus metadata.

    // Here we reuse the same cmeta instance for both new tablets. This is safe, because:

    // 1) Their consensus metadata only differ by tablet id.

    // 2) Flush() will save it into a new path corresponding to tablet id we set before flushing.

    cmeta->set_tablet_id(new_tablet_id);

    cmeta->set_split_parent_tablet_id(tablet_id);

    RETURN_NOT_OK(cmeta->Flush());

    const auto& dest_wal_dir = tcmeta.raft_group_metadata->wal_dir();

    RETURN_NOT_OK(raft_log->CopyTo(dest_wal_dir));

    MAYBE_FAULT(FLAGS_TEST_fault_crash_in_split_after_log_copied);

    tcmeta.raft_group_metadata->set_tablet_data_state(TABLET_DATA_READY);

    RETURN_NOT_OK(tcmeta.raft_group_metadata->Flush());

  }

  meta.SetSplitDone(operation->op_id(), request->new_tablet1_id(), request->new_tablet2_id());

  RETURN_NOT_OK(meta.Flush());

  tablet->SplitDone();

  for (auto& tcmeta : tcmetas) {

    // Call CreatePeerAndOpenTablet asynchronously to avoid write-locking TSTabletManager::mutex_

    // here since apply of SPLIT_OP is done under ReplicaState lock and this could lead to deadlock

    // in case of reverse lock order in some other thread.

    // See https://github.com/yugabyte/yugabyte-db/issues/4312 for more details.

    RETURN_NOT_OK(apply_pool_->SubmitFunc(std::bind(

        &TSTabletManager::CreatePeerAndOpenTablet, this, tcmeta.raft_group_metadata,

        tcmeta.transition_deleter)));

  }

  successfully_completed = true;

  LOG_WITH_PREFIX(INFO) << "Tablet " << tablet_id << " split operation has been applied";

  return Status::OK();

}

string LogPrefix(const string& tablet_id, const string& uuid) {

  return "T " + tablet_id + " P " + uuid + ": ";

}

Status CheckLeaderTermNotLower(

    const string& tablet_id,

    const string& uuid,

    int64_t leader_term,

    int64_t last_logged_term) {

  if (PREDICT_FALSE(leader_term < last_logged_term)) {

    Status s = STATUS(InvalidArgument,

        Substitute("Leader has replica of tablet $0 with term $1 lower than last "

                   "logged term $2 on local replica. Rejecting remote bootstrap request",

                   tablet_id, leader_term, last_logged_term));

    LOG(WARNING) << LogPrefix(tablet_id, uuid) << "Remote bootstrap: " << s;

    return s;

  }

  return Status::OK();

}

Status HandleReplacingStaleTablet(

    RaftGroupMetadataPtr meta,

    TabletPeerPtr old_tablet_peer,

    const string& tablet_id,

    const string& uuid,

    const int64_t& leader_term) {

  TabletDataState data_state = meta->tablet_data_state();

  switch (data_state) {

    case TABLET_DATA_COPYING: {

      // This should not be possible due to the transition_in_progress_ "lock".

      LOG(FATAL) << LogPrefix(tablet_id, uuid) << " Remote bootstrap: "

                 << "Found tablet in TABLET_DATA_COPYING state during StartRemoteBootstrap()";

    }

    case TABLET_DATA_TOMBSTONED: {

      RETURN_NOT_OK(old_tablet_peer->CheckShutdownOrNotStarted());

      int64_t last_logged_term = meta->tombstone_last_logged_opid().term;

      RETURN_NOT_OK(CheckLeaderTermNotLower(tablet_id,

                                            uuid,

                                            leader_term,

                                            last_logged_term));

      break;

    }

    case TABLET_DATA_SPLIT_COMPLETED:

    case TABLET_DATA_READY: {

      if (tablet_id == master::kSysCatalogTabletId) {

        LOG(FATAL) << LogPrefix(tablet_id, uuid) << " Remote bootstrap: "

                   << "Found tablet in " << TabletDataState_Name(data_state)

                   << " state during StartRemoteBootstrap()";

      }

      // There's a valid race here that can lead us to come here:

      // 1. Leader sends a second remote bootstrap request as a result of receiving a

      // TABLET_NOT_FOUND from this tserver while it was in the middle of a remote bootstrap.

      // 2. The remote bootstrap request arrives after the first one is finished, and it is able to

      // grab the mutex.

      // 3. This tserver finds that it already has the metadata for the tablet, and determines that

      // it needs to replace the tablet setting replacing_tablet to true.

      // In this case, the master can simply ignore this error.

      return STATUS_FORMAT(

          IllegalState, "Tablet $0 in $1 state", tablet_id, TabletDataState_Name(data_state));

    }

    default: {

      return STATUS(IllegalState,

          Substitute("Found tablet $0 in unexpected state $1 for remote bootstrap.",

                     tablet_id, TabletDataState_Name(data_state)));

    }

  }

  return Status::OK();

}

Status TSTabletManager::StartRemoteBootstrap(const StartRemoteBootstrapRequestPB& req) {

  // To prevent racing against Shutdown, we increment this as soon as we start. This should be done

  // before checking for ClosingUnlocked, as on shutdown, we proceed in reverse:

  // - first mark as closing

  // - then wait for num_tablets_being_remote_bootstrapped_ == 0

  ++num_tablets_being_remote_bootstrapped_;

  auto private_addr = req.source_private_addr()[0].host();

  auto decrement_num_rbs_se = ScopeExit([this, &private_addr](){

    {

      std::lock_guard<RWMutex> lock(mutex_);

      auto iter = bootstrap_source_addresses_.find(private_addr);

      if (iter != bootstrap_source_addresses_.end()) {

        bootstrap_source_addresses_.erase(iter);

      }

    }

    --num_tablets_being_remote_bootstrapped_;

  });

  LongOperationTracker tracker("StartRemoteBootstrap", 5s);

  const string& tablet_id = req.tablet_id();

  const string& bootstrap_peer_uuid = req.bootstrap_peer_uuid();