/Users/deen/code/yugabyte-db/src/yb/master/master_snapshot_coordinator.cc
Line | Count | Source (jump to first uncovered line) |
1 | | // Copyright (c) YugaByte, Inc. |
2 | | // |
3 | | // Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except |
4 | | // in compliance with the License. You may obtain a copy of the License at |
5 | | // |
6 | | // http://www.apache.org/licenses/LICENSE-2.0 |
7 | | // |
8 | | // Unless required by applicable law or agreed to in writing, software distributed under the License |
9 | | // is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
10 | | // or implied. See the License for the specific language governing permissions and limitations |
11 | | // under the License. |
12 | | // |
13 | | |
14 | | #include "yb/master/master_snapshot_coordinator.h" |
15 | | |
16 | | #include <unordered_map> |
17 | | |
18 | | #include <boost/multi_index/composite_key.hpp> |
19 | | #include <boost/multi_index/mem_fun.hpp> |
20 | | #include <boost/asio/io_context.hpp> |
21 | | |
22 | | #include "yb/common/snapshot.h" |
23 | | |
24 | | #include "yb/docdb/doc_key.h" |
25 | | #include "yb/docdb/value.h" |
26 | | #include "yb/docdb/value_type.h" |
27 | | |
28 | | #include "yb/master/async_snapshot_tasks.h" |
29 | | #include "yb/master/catalog_entity_info.h" |
30 | | #include "yb/master/master_error.h" |
31 | | #include "yb/master/master_heartbeat.pb.h" |
32 | | #include "yb/master/master_util.h" |
33 | | #include "yb/master/restoration_state.h" |
34 | | #include "yb/master/snapshot_coordinator_context.h" |
35 | | #include "yb/master/snapshot_schedule_state.h" |
36 | | #include "yb/master/snapshot_state.h" |
37 | | #include "yb/master/sys_catalog_writer.h" |
38 | | |
39 | | #include "yb/rpc/poller.h" |
40 | | #include "yb/rpc/scheduler.h" |
41 | | |
42 | | #include "yb/tablet/operations/snapshot_operation.h" |
43 | | #include "yb/tablet/operations/write_operation.h" |
44 | | #include "yb/tablet/tablet.h" |
45 | | #include "yb/tablet/tablet_snapshots.h" |
46 | | #include "yb/tablet/write_query.h" |
47 | | |
48 | | #include "yb/util/async_util.h" |
49 | | #include "yb/util/flag_tags.h" |
50 | | #include "yb/util/pb_util.h" |
51 | | #include "yb/util/status_format.h" |
52 | | #include "yb/util/status_log.h" |
53 | | #include "yb/util/stopwatch.h" |
54 | | |
55 | | using namespace std::literals; |
56 | | using namespace std::placeholders; |
57 | | |
58 | | DECLARE_int32(sys_catalog_write_timeout_ms); |
59 | | |
60 | | DEFINE_uint64(snapshot_coordinator_poll_interval_ms, 5000, |
61 | | "Poll interval for snapshot coordinator in milliseconds."); |
62 | | |
63 | | DEFINE_test_flag(bool, skip_sending_restore_finished, false, |
64 | | "Whether we should skip sending RESTORE_FINISHED to tablets."); |
65 | | |
66 | | DEFINE_bool(schedule_snapshot_rpcs_out_of_band, true, |
67 | | "Should tablet snapshot RPCs be scheduled out of band from the periodic" |
68 | | " background scheduling."); |
69 | | TAG_FLAG(schedule_snapshot_rpcs_out_of_band, runtime); |
70 | | |
71 | | DECLARE_bool(allow_consecutive_restore); |
72 | | |
73 | | namespace yb { |
74 | | namespace master { |
75 | | |
76 | | namespace { |
77 | | |
78 | | YB_DEFINE_ENUM(Bound, (kFirst)(kLast)); |
79 | | YB_DEFINE_ENUM(RestorePhase, (kInitial)(kPostSysCatalogLoad)); |
80 | | |
81 | | void SubmitWrite( |
82 | | docdb::KeyValueWriteBatchPB&& write_batch, int64_t leader_term, |
83 | | SnapshotCoordinatorContext* context, |
84 | 28 | const std::shared_ptr<Synchronizer>& synchronizer = nullptr) { |
85 | 28 | auto query = std::make_unique<tablet::WriteQuery>( |
86 | 28 | leader_term, CoarseMonoClock::now() + FLAGS_sys_catalog_write_timeout_ms * 1ms, |
87 | 28 | /* context */ nullptr, /* tablet= */ nullptr); |
88 | 28 | if (synchronizer) { |
89 | 8 | query->set_callback( |
90 | 8 | tablet::MakeWeakSynchronizerOperationCompletionCallback(synchronizer)); |
91 | 8 | } |
92 | 28 | *query->operation().AllocateRequest()->mutable_write_batch() = std::move(write_batch); |
93 | 28 | context->Submit(query.release()->PrepareSubmit(), leader_term); |
94 | 28 | } |
95 | | |
96 | | CHECKED_STATUS SynchronizedWrite( |
97 | | docdb::KeyValueWriteBatchPB&& write_batch, int64_t leader_term, CoarseTimePoint deadline, |
98 | 8 | SnapshotCoordinatorContext* context) { |
99 | 8 | auto synchronizer = std::make_shared<Synchronizer>(); |
100 | 8 | SubmitWrite(std::move(write_batch), leader_term, context, synchronizer); |
101 | 8 | return synchronizer->WaitUntil(ToSteady(deadline)); |
102 | 8 | } |
103 | | |
104 | | struct NoOp { |
105 | | template <class... Args> |
106 | | void operator()(Args&&... args) const {} |
107 | | }; |
108 | | |
109 | | // Utility to create callback that is invoked when operation done. |
110 | | // Finds appropriate entry in passed collection and invokes Done on it. |
111 | | template <class Collection, class PostProcess = NoOp> |
112 | | auto MakeDoneCallback( |
113 | | std::mutex* mutex, const Collection& collection, const typename Collection::key_type& key, |
114 | 35 | const TabletId& tablet_id, const PostProcess& post_process = PostProcess()) { |
115 | 35 | struct DoneFunctor { |
116 | 35 | std::mutex& mutex; |
117 | 35 | const Collection& collection; |
118 | 35 | typename Collection::key_type key; |
119 | 35 | TabletId tablet_id; |
120 | 35 | PostProcess post_process; |
121 | | |
122 | 35 | void operator()(Result<const tserver::TabletSnapshotOpResponsePB&> resp) const { |
123 | 35 | std::unique_lock<std::mutex> lock(mutex); |
124 | 35 | auto it = collection.find(key); |
125 | 35 | if (it == collection.end()) { |
126 | 0 | LOG(DFATAL) << "Received reply for unknown " << key; |
127 | 0 | return; |
128 | 0 | } |
129 | | |
130 | 35 | (**it).Done(tablet_id, ResultToStatus(resp)); |
131 | 35 | post_process(it->get(), &lock); |
132 | 35 | } master_snapshot_coordinator.cc:auto yb::master::(anonymous namespace)::MakeDoneCallback<boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > >, std::__1::__bind<void (yb::master::MasterSnapshotCoordinator::Impl::*)(yb::master::SnapshotState*, long long, std::__1::unique_lock<std::__1::mutex>*), yb::master::MasterSnapshotCoordinator::Impl*, std::__1::placeholders::__ph<1> const&, long long&, std::__1::placeholders::__ph<2> const&> >(std::__1::mutex*, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > const::key_type const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, std::__1::__bind<void (yb::master::MasterSnapshotCoordinator::Impl::*)(yb::master::SnapshotState*, long long, std::__1::unique_lock<std::__1::mutex>*), yb::master::MasterSnapshotCoordinator::Impl*, std::__1::placeholders::__ph<1> const&, long long&, std::__1::placeholders::__ph<2> const&> const&)::DoneFunctor::operator()(yb::Result<yb::tserver::TabletSnapshotOpResponsePB const&>) const Line | Count | Source | 122 | 25 | void operator()(Result<const tserver::TabletSnapshotOpResponsePB&> resp) const { | 123 | 25 | std::unique_lock<std::mutex> lock(mutex); | 124 | 25 | auto it = collection.find(key); | 125 | 25 | if (it == collection.end()) { | 126 | 0 | LOG(DFATAL) << "Received reply for unknown " << key; | 127 | 0 | return; | 128 | 0 | } | 129 | | | 130 | 25 | (**it).Done(tablet_id, ResultToStatus(resp)); | 131 | 25 | post_process(it->get(), &lock); | 132 | 25 | } |
master_snapshot_coordinator.cc:auto yb::master::(anonymous namespace)::MakeDoneCallback<boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::RestorationState, yb::StronglyTypedUuid<yb::TxnSnapshotRestorationId_Tag> const&, &(yb::master::RestorationState::restoration_id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> > > >, std::__1::__bind<void (yb::master::MasterSnapshotCoordinator::Impl::*)(yb::master::RestorationState*, long long), yb::master::MasterSnapshotCoordinator::Impl*, std::__1::placeholders::__ph<1> const&, long long&> >(std::__1::mutex*, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::RestorationState, yb::StronglyTypedUuid<yb::TxnSnapshotRestorationId_Tag> const&, &(yb::master::RestorationState::restoration_id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> > > > const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::RestorationState, yb::StronglyTypedUuid<yb::TxnSnapshotRestorationId_Tag> const&, &(yb::master::RestorationState::restoration_id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> > > > const::key_type const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, std::__1::__bind<void (yb::master::MasterSnapshotCoordinator::Impl::*)(yb::master::RestorationState*, long long), yb::master::MasterSnapshotCoordinator::Impl*, std::__1::placeholders::__ph<1> const&, long long&> const&)::DoneFunctor::operator()(yb::Result<yb::tserver::TabletSnapshotOpResponsePB const&>) const Line | Count | Source | 122 | 10 | void operator()(Result<const tserver::TabletSnapshotOpResponsePB&> resp) const { | 123 | 10 | std::unique_lock<std::mutex> lock(mutex); | 124 | 10 | auto it = collection.find(key); | 125 | 10 | if (it == collection.end()) { | 126 | 0 | LOG(DFATAL) << "Received reply for unknown " << key; | 127 | 0 | return; | 128 | 0 | } | 129 | | | 130 | 10 | (**it).Done(tablet_id, ResultToStatus(resp)); | 131 | 10 | post_process(it->get(), &lock); | 132 | 10 | } |
|
133 | 35 | }; |
134 | | |
135 | 35 | return DoneFunctor { |
136 | 35 | .mutex = *mutex, |
137 | 35 | .collection = collection, |
138 | 35 | .key = key, |
139 | 35 | .tablet_id = tablet_id, |
140 | 35 | .post_process = post_process, |
141 | 35 | }; |
142 | 35 | } master_snapshot_coordinator.cc:auto yb::master::(anonymous namespace)::MakeDoneCallback<boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > >, std::__1::__bind<void (yb::master::MasterSnapshotCoordinator::Impl::*)(yb::master::SnapshotState*, long long, std::__1::unique_lock<std::__1::mutex>*), yb::master::MasterSnapshotCoordinator::Impl*, std::__1::placeholders::__ph<1> const&, long long&, std::__1::placeholders::__ph<2> const&> >(std::__1::mutex*, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > const::key_type const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, std::__1::__bind<void (yb::master::MasterSnapshotCoordinator::Impl::*)(yb::master::SnapshotState*, long long, std::__1::unique_lock<std::__1::mutex>*), yb::master::MasterSnapshotCoordinator::Impl*, std::__1::placeholders::__ph<1> const&, long long&, std::__1::placeholders::__ph<2> const&> const&) Line | Count | Source | 114 | 25 | const TabletId& tablet_id, const PostProcess& post_process = PostProcess()) { | 115 | 25 | struct DoneFunctor { | 116 | 25 | std::mutex& mutex; | 117 | 25 | const Collection& collection; | 118 | 25 | typename Collection::key_type key; | 119 | 25 | TabletId tablet_id; | 120 | 25 | PostProcess post_process; | 121 | | | 122 | 25 | void operator()(Result<const tserver::TabletSnapshotOpResponsePB&> resp) const { | 123 | 25 | std::unique_lock<std::mutex> lock(mutex); | 124 | 25 | auto it = collection.find(key); | 125 | 25 | if (it == collection.end()) { | 126 | 25 | LOG(DFATAL) << "Received reply for unknown " << key; | 127 | 25 | return; | 128 | 25 | } | 129 | | | 130 | 25 | (**it).Done(tablet_id, ResultToStatus(resp)); | 131 | 25 | post_process(it->get(), &lock); | 132 | 25 | } | 133 | 25 | }; | 134 | | | 135 | 25 | return DoneFunctor { | 136 | 25 | .mutex = *mutex, | 137 | 25 | .collection = collection, | 138 | 25 | .key = key, | 139 | 25 | .tablet_id = tablet_id, | 140 | 25 | .post_process = post_process, | 141 | 25 | }; | 142 | 25 | } |
master_snapshot_coordinator.cc:auto yb::master::(anonymous namespace)::MakeDoneCallback<boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::RestorationState, yb::StronglyTypedUuid<yb::TxnSnapshotRestorationId_Tag> const&, &(yb::master::RestorationState::restoration_id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> > > >, std::__1::__bind<void (yb::master::MasterSnapshotCoordinator::Impl::*)(yb::master::RestorationState*, long long), yb::master::MasterSnapshotCoordinator::Impl*, std::__1::placeholders::__ph<1> const&, long long&> >(std::__1::mutex*, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::RestorationState, yb::StronglyTypedUuid<yb::TxnSnapshotRestorationId_Tag> const&, &(yb::master::RestorationState::restoration_id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> > > > const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::RestorationState, yb::StronglyTypedUuid<yb::TxnSnapshotRestorationId_Tag> const&, &(yb::master::RestorationState::restoration_id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::RestorationState, std::__1::default_delete<yb::master::RestorationState> > > > const::key_type const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, std::__1::__bind<void (yb::master::MasterSnapshotCoordinator::Impl::*)(yb::master::RestorationState*, long long), yb::master::MasterSnapshotCoordinator::Impl*, std::__1::placeholders::__ph<1> const&, long long&> const&) Line | Count | Source | 114 | 10 | const TabletId& tablet_id, const PostProcess& post_process = PostProcess()) { | 115 | 10 | struct DoneFunctor { | 116 | 10 | std::mutex& mutex; | 117 | 10 | const Collection& collection; | 118 | 10 | typename Collection::key_type key; | 119 | 10 | TabletId tablet_id; | 120 | 10 | PostProcess post_process; | 121 | | | 122 | 10 | void operator()(Result<const tserver::TabletSnapshotOpResponsePB&> resp) const { | 123 | 10 | std::unique_lock<std::mutex> lock(mutex); | 124 | 10 | auto it = collection.find(key); | 125 | 10 | if (it == collection.end()) { | 126 | 10 | LOG(DFATAL) << "Received reply for unknown " << key; | 127 | 10 | return; | 128 | 10 | } | 129 | | | 130 | 10 | (**it).Done(tablet_id, ResultToStatus(resp)); | 131 | 10 | post_process(it->get(), &lock); | 132 | 10 | } | 133 | 10 | }; | 134 | | | 135 | 10 | return DoneFunctor { | 136 | 10 | .mutex = *mutex, | 137 | 10 | .collection = collection, | 138 | 10 | .key = key, | 139 | 10 | .tablet_id = tablet_id, | 140 | 10 | .post_process = post_process, | 141 | 10 | }; | 142 | 10 | } |
|
143 | | |
144 | | } // namespace |
145 | | |
146 | | class MasterSnapshotCoordinator::Impl { |
147 | | public: |
148 | | explicit Impl(SnapshotCoordinatorContext* context) |
149 | 8.07k | : context_(*context), poller_(std::bind(&Impl::Poll, this)) {} |
150 | | |
151 | | Result<TxnSnapshotId> Create( |
152 | 2 | const SysRowEntries& entries, bool imported, int64_t leader_term, CoarseTimePoint deadline) { |
153 | 2 | auto synchronizer = std::make_shared<Synchronizer>(); |
154 | 2 | auto snapshot_id = TxnSnapshotId::GenerateRandom(); |
155 | 2 | SubmitCreate( |
156 | 2 | entries, imported, SnapshotScheduleId::Nil(), HybridTime::kInvalid, snapshot_id, |
157 | 2 | leader_term, |
158 | 2 | tablet::MakeWeakSynchronizerOperationCompletionCallback(synchronizer)); |
159 | 2 | RETURN_NOT_OK(synchronizer->WaitUntil(ToSteady(deadline))); |
160 | | |
161 | 2 | return snapshot_id; |
162 | 2 | } |
163 | | |
164 | | Result<TxnSnapshotId> CreateForSchedule( |
165 | 1 | const SnapshotScheduleId& schedule_id, int64_t leader_term, CoarseTimePoint deadline) { |
166 | 1 | boost::optional<SnapshotScheduleOperation> operation; |
167 | 1 | { |
168 | 1 | std::lock_guard<std::mutex> lock(mutex_); |
169 | 1 | auto it = schedules_.find(schedule_id); |
170 | 1 | if (it == schedules_.end()) { |
171 | 0 | return STATUS_FORMAT(NotFound, "Unknown snapshot schedule: $0", schedule_id); |
172 | 0 | } |
173 | 1 | auto* last_snapshot = BoundingSnapshot((**it).id(), Bound::kLast); |
174 | 1 | auto last_snapshot_time = last_snapshot ? last_snapshot->snapshot_hybrid_time() |
175 | 1 | : HybridTime::kInvalid0 ; |
176 | 1 | auto creating_snapshot_data = (**it).creating_snapshot_data(); |
177 | 1 | if (creating_snapshot_data.snapshot_id) { |
178 | 0 | auto snapshot_it = snapshots_.find(creating_snapshot_data.snapshot_id); |
179 | 0 | if (snapshot_it != snapshots_.end()) { |
180 | 0 | VLOG(2) << __func__ << " for " << schedule_id << " while creating snapshot: " |
181 | 0 | << (**snapshot_it).ToString(); |
182 | 0 | } else { |
183 | 0 | auto passed = CoarseMonoClock::now() - creating_snapshot_data.start_time; |
184 | 0 | auto message = Format( |
185 | 0 | "$0 for $1 while creating unknown snapshot: $2 (passed $3)", |
186 | 0 | __func__, schedule_id, creating_snapshot_data.snapshot_id, passed); |
187 | 0 | if (passed > 30s) { |
188 | 0 | LOG(DFATAL) << message; |
189 | 0 | } else { |
190 | 0 | VLOG(2) << message; |
191 | 0 | } |
192 | 0 | } |
193 | 0 | } |
194 | 1 | operation = VERIFY_RESULT((**it).ForceCreateSnapshot(last_snapshot_time)); |
195 | 1 | } |
196 | | |
197 | 0 | auto synchronizer = std::make_shared<Synchronizer>(); |
198 | 1 | RETURN_NOT_OK(ExecuteScheduleOperation(*operation, leader_term, synchronizer)); |
199 | 1 | RETURN_NOT_OK(synchronizer->WaitUntil(ToSteady(deadline))); |
200 | | |
201 | 1 | return operation->snapshot_id; |
202 | 1 | } |
203 | | |
204 | | CHECKED_STATUS CreateReplicated( |
205 | 51 | int64_t leader_term, const tablet::SnapshotOperation& operation) { |
206 | | // TODO(txn_backup) retain logs with this operation while doing snapshot |
207 | 51 | auto id = VERIFY_RESULT(FullyDecodeTxnSnapshotId(operation.request()->snapshot_id())); |
208 | | |
209 | 0 | VLOG(1) << __func__ << "(" << id << ", " << operation.ToString() << ")"; |
210 | | |
211 | 51 | auto snapshot = std::make_unique<SnapshotState>( |
212 | 51 | &context_, id, *operation.request(), |
213 | 51 | GetRpcLimit(FLAGS_max_concurrent_snapshot_rpcs, |
214 | 51 | FLAGS_max_concurrent_snapshot_rpcs_per_tserver, leader_term)); |
215 | | |
216 | 51 | TabletSnapshotOperations operations; |
217 | 51 | docdb::KeyValueWriteBatchPB write_batch; |
218 | 51 | RETURN_NOT_OK(snapshot->StoreToWriteBatch(&write_batch)); |
219 | 51 | boost::optional<tablet::CreateSnapshotData> sys_catalog_snapshot_data; |
220 | 51 | bool snapshot_empty = false; |
221 | 51 | { |
222 | 51 | std::lock_guard<std::mutex> lock(mutex_); |
223 | 51 | auto emplace_result = snapshots_.emplace(std::move(snapshot)); |
224 | 51 | if (!emplace_result.second) { |
225 | 0 | return STATUS_FORMAT(IllegalState, "Duplicate snapshot id: $0", id); |
226 | 0 | } |
227 | | |
228 | 51 | if (leader_term >= 0) { |
229 | 21 | (**emplace_result.first).PrepareOperations(&operations); |
230 | 21 | } |
231 | 51 | auto temp = (**emplace_result.first).SysCatalogSnapshotData(operation); |
232 | 51 | if (temp.ok()) { |
233 | 45 | sys_catalog_snapshot_data = *temp; |
234 | 45 | } else if (6 !temp.status().IsUninitialized()6 ) { |
235 | 4 | return temp.status(); |
236 | 4 | } |
237 | 47 | snapshot_empty = (**emplace_result.first).Empty(); |
238 | 47 | } |
239 | | |
240 | 47 | RETURN_NOT_OK(operation.tablet()->ApplyOperation(operation, /* batch_idx= */ -1, write_batch)); |
241 | 47 | if (sys_catalog_snapshot_data) { |
242 | 45 | RETURN_NOT_OK(operation.tablet()->snapshots().Create(*sys_catalog_snapshot_data)); |
243 | 45 | } |
244 | | |
245 | 47 | ExecuteOperations(operations, leader_term); |
246 | | |
247 | 47 | if (leader_term >= 0 && snapshot_empty17 ) { |
248 | | // There could be snapshot for 0 tables, so they should be marked as complete right after |
249 | | // creation. |
250 | 8 | UpdateSnapshotIfPresent(id, leader_term); |
251 | 8 | } |
252 | | |
253 | 47 | return Status::OK(); |
254 | 47 | } |
255 | | |
256 | | void UpdateSnapshotIfPresent(const TxnSnapshotId& id, int64_t leader_term) |
257 | 8 | NO_THREAD_SAFETY_ANALYSIS EXCLUDES(mutex_) { |
258 | 8 | std::unique_lock<std::mutex> lock(mutex_); |
259 | 8 | auto it = snapshots_.find(id); |
260 | 8 | if (it != snapshots_.end()) { |
261 | 8 | UpdateSnapshot(it->get(), leader_term, &lock); |
262 | 8 | } |
263 | 8 | } |
264 | | |
265 | 89 | CHECKED_STATUS Load(tablet::Tablet* tablet) { |
266 | 89 | std::lock_guard<std::mutex> lock(mutex_); |
267 | 89 | RETURN_NOT_OK(EnumerateSysCatalog(tablet, context_.schema(), SysRowEntryType::SNAPSHOT, |
268 | 89 | [this](const Slice& id, const Slice& data) NO_THREAD_SAFETY_ANALYSIS -> Status { |
269 | 89 | return LoadEntry<SysSnapshotEntryPB>(id, data, &snapshots_); |
270 | 89 | })); |
271 | 89 | return EnumerateSysCatalog(tablet, context_.schema(), SysRowEntryType::SNAPSHOT_SCHEDULE, |
272 | 89 | [this](const Slice& id, const Slice& data) NO_THREAD_SAFETY_ANALYSIS -> Status { |
273 | 0 | return LoadEntry<SnapshotScheduleOptionsPB>(id, data, &schedules_); |
274 | 0 | }); |
275 | 89 | } |
276 | | |
277 | 99.2M | CHECKED_STATUS ApplyWritePair(Slice key, const Slice& value) { |
278 | 99.2M | docdb::SubDocKey sub_doc_key; |
279 | 99.2M | RETURN_NOT_OK(sub_doc_key.FullyDecodeFrom(key, docdb::HybridTimeRequired::kFalse)); |
280 | | |
281 | 99.2M | if (sub_doc_key.doc_key().has_cotable_id()) { |
282 | 98.1M | return Status::OK(); |
283 | 98.1M | } |
284 | | |
285 | 1.11M | if (sub_doc_key.doc_key().range_group().size() != 2) { |
286 | 0 | LOG(DFATAL) << "Unexpected size of range group in sys catalog entry (2 expected): " |
287 | 0 | << AsString(sub_doc_key.doc_key().range_group()) << "(" << sub_doc_key.ToString() |
288 | 0 | << ")"; |
289 | 0 | return Status::OK(); |
290 | 0 | } |
291 | | |
292 | 1.11M | auto first_key = sub_doc_key.doc_key().range_group().front(); |
293 | 1.11M | if (first_key.value_type() != docdb::ValueType::kInt32) { |
294 | 0 | LOG(DFATAL) << "Unexpected value type for the first range component of sys catalog entry " |
295 | 0 | << "(kInt32 expected): " |
296 | 0 | << AsString(sub_doc_key.doc_key().range_group());; |
297 | 0 | } |
298 | | |
299 | 1.11M | if (first_key.GetInt32() == SysRowEntryType::SNAPSHOT) { |
300 | 94 | return DoApplyWrite<SysSnapshotEntryPB>( |
301 | 94 | sub_doc_key.doc_key().range_group()[1].GetString(), value, &snapshots_); |
302 | 94 | } |
303 | | |
304 | 1.11M | if (first_key.GetInt32() == SysRowEntryType::SNAPSHOT_SCHEDULE) { |
305 | 33 | return DoApplyWrite<SnapshotScheduleOptionsPB>( |
306 | 33 | sub_doc_key.doc_key().range_group()[1].GetString(), value, &schedules_); |
307 | 33 | } |
308 | | |
309 | 1.11M | return Status::OK(); |
310 | 1.11M | } |
311 | | |
312 | | template <class Pb, class Map> |
313 | 127 | CHECKED_STATUS DoApplyWrite(const std::string& id_str, const Slice& value, Map* map) { |
314 | 127 | docdb::Value decoded_value; |
315 | 127 | RETURN_NOT_OK(decoded_value.Decode(value)); |
316 | | |
317 | 127 | auto value_type = decoded_value.primitive_value().value_type(); |
318 | | |
319 | 127 | if (value_type == docdb::ValueType::kTombstone) { |
320 | 0 | std::lock_guard<std::mutex> lock(mutex_); |
321 | 0 | auto id = Uuid::TryFullyDecode(id_str); |
322 | 0 | if (id.IsNil()) { |
323 | 0 | LOG(WARNING) << "Unable to decode id: " << id_str; |
324 | 0 | return Status::OK(); |
325 | 0 | } |
326 | 0 | bool erased = map->erase(typename Map::key_type(id)) != 0; |
327 | 0 | LOG_IF(DFATAL, !erased) << "Unknown entry tombstoned: " << id.ToString(); |
328 | 0 | return Status::OK(); |
329 | 0 | } |
330 | | |
331 | 127 | if (value_type != docdb::ValueType::kString) { |
332 | 0 | return STATUS_FORMAT( |
333 | 0 | Corruption, |
334 | 0 | "Bad value type: $0, expected kString while replaying write for sys catalog", |
335 | 0 | decoded_value.primitive_value().value_type()); |
336 | 0 | } |
337 | | |
338 | 127 | std::lock_guard<std::mutex> lock(mutex_); |
339 | 127 | return LoadEntry<Pb>(id_str, decoded_value.primitive_value().GetString(), map); |
340 | 127 | } yb::Status yb::master::MasterSnapshotCoordinator::Impl::DoApplyWrite<yb::master::SysSnapshotEntryPB, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > >(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, yb::Slice const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > >*) Line | Count | Source | 313 | 94 | CHECKED_STATUS DoApplyWrite(const std::string& id_str, const Slice& value, Map* map) { | 314 | 94 | docdb::Value decoded_value; | 315 | 94 | RETURN_NOT_OK(decoded_value.Decode(value)); | 316 | | | 317 | 94 | auto value_type = decoded_value.primitive_value().value_type(); | 318 | | | 319 | 94 | if (value_type == docdb::ValueType::kTombstone) { | 320 | 0 | std::lock_guard<std::mutex> lock(mutex_); | 321 | 0 | auto id = Uuid::TryFullyDecode(id_str); | 322 | 0 | if (id.IsNil()) { | 323 | 0 | LOG(WARNING) << "Unable to decode id: " << id_str; | 324 | 0 | return Status::OK(); | 325 | 0 | } | 326 | 0 | bool erased = map->erase(typename Map::key_type(id)) != 0; | 327 | 0 | LOG_IF(DFATAL, !erased) << "Unknown entry tombstoned: " << id.ToString(); | 328 | 0 | return Status::OK(); | 329 | 0 | } | 330 | | | 331 | 94 | if (value_type != docdb::ValueType::kString) { | 332 | 0 | return STATUS_FORMAT( | 333 | 0 | Corruption, | 334 | 0 | "Bad value type: $0, expected kString while replaying write for sys catalog", | 335 | 0 | decoded_value.primitive_value().value_type()); | 336 | 0 | } | 337 | | | 338 | 94 | std::lock_guard<std::mutex> lock(mutex_); | 339 | 94 | return LoadEntry<Pb>(id_str, decoded_value.primitive_value().GetString(), map); | 340 | 94 | } |
yb::Status yb::master::MasterSnapshotCoordinator::Impl::DoApplyWrite<yb::master::SnapshotScheduleOptionsPB, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > > >(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, yb::Slice const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > >*) Line | Count | Source | 313 | 33 | CHECKED_STATUS DoApplyWrite(const std::string& id_str, const Slice& value, Map* map) { | 314 | 33 | docdb::Value decoded_value; | 315 | 33 | RETURN_NOT_OK(decoded_value.Decode(value)); | 316 | | | 317 | 33 | auto value_type = decoded_value.primitive_value().value_type(); | 318 | | | 319 | 33 | if (value_type == docdb::ValueType::kTombstone) { | 320 | 0 | std::lock_guard<std::mutex> lock(mutex_); | 321 | 0 | auto id = Uuid::TryFullyDecode(id_str); | 322 | 0 | if (id.IsNil()) { | 323 | 0 | LOG(WARNING) << "Unable to decode id: " << id_str; | 324 | 0 | return Status::OK(); | 325 | 0 | } | 326 | 0 | bool erased = map->erase(typename Map::key_type(id)) != 0; | 327 | 0 | LOG_IF(DFATAL, !erased) << "Unknown entry tombstoned: " << id.ToString(); | 328 | 0 | return Status::OK(); | 329 | 0 | } | 330 | | | 331 | 33 | if (value_type != docdb::ValueType::kString) { | 332 | 0 | return STATUS_FORMAT( | 333 | 0 | Corruption, | 334 | 0 | "Bad value type: $0, expected kString while replaying write for sys catalog", | 335 | 0 | decoded_value.primitive_value().value_type()); | 336 | 0 | } | 337 | | | 338 | 33 | std::lock_guard<std::mutex> lock(mutex_); | 339 | 33 | return LoadEntry<Pb>(id_str, decoded_value.primitive_value().GetString(), map); | 340 | 33 | } |
|
341 | | |
342 | | CHECKED_STATUS ListSnapshots( |
343 | 9 | const TxnSnapshotId& snapshot_id, bool list_deleted, ListSnapshotsResponsePB* resp) { |
344 | 9 | std::lock_guard<std::mutex> lock(mutex_); |
345 | 9 | if (snapshot_id.IsNil()) { |
346 | 1 | for (const auto& p : snapshots_.get<ScheduleTag>()) { |
347 | 1 | if (!list_deleted) { |
348 | 1 | auto aggreaged_state = p->AggregatedState(); |
349 | 1 | if (aggreaged_state.ok() && *aggreaged_state == SysSnapshotEntryPB::DELETED) { |
350 | 0 | continue; |
351 | 0 | } |
352 | 1 | } |
353 | 1 | RETURN_NOT_OK(p->ToPB(resp->add_snapshots())); |
354 | 1 | } |
355 | 1 | return Status::OK(); |
356 | 1 | } |
357 | | |
358 | 8 | SnapshotState& snapshot = VERIFY_RESULT(FindSnapshot(snapshot_id)); |
359 | 0 | return snapshot.ToPB(resp->add_snapshots()); |
360 | 8 | } |
361 | | |
362 | | CHECKED_STATUS Delete( |
363 | 0 | const TxnSnapshotId& snapshot_id, int64_t leader_term, CoarseTimePoint deadline) { |
364 | 0 | VLOG_WITH_FUNC(4) << snapshot_id << ", " << leader_term; |
365 | |
|
366 | 0 | { |
367 | 0 | std::lock_guard<std::mutex> lock(mutex_); |
368 | 0 | SnapshotState& snapshot = VERIFY_RESULT(FindSnapshot(snapshot_id)); |
369 | 0 | RETURN_NOT_OK(snapshot.TryStartDelete()); |
370 | 0 | } |
371 | | |
372 | 0 | auto synchronizer = std::make_shared<Synchronizer>(); |
373 | 0 | SubmitDelete(snapshot_id, leader_term, synchronizer); |
374 | 0 | return synchronizer->WaitUntil(ToSteady(deadline)); |
375 | 0 | } |
376 | | |
377 | | CHECKED_STATUS DeleteReplicated( |
378 | 0 | int64_t leader_term, const tablet::SnapshotOperation& operation) { |
379 | 0 | auto snapshot_id = VERIFY_RESULT(FullyDecodeTxnSnapshotId(operation.request()->snapshot_id())); |
380 | 0 | VLOG_WITH_FUNC(4) << leader_term << ", " << snapshot_id; |
381 | |
|
382 | 0 | docdb::KeyValueWriteBatchPB write_batch; |
383 | 0 | TabletSnapshotOperations operations; |
384 | 0 | bool delete_sys_catalog_snapshot; |
385 | 0 | { |
386 | 0 | std::lock_guard<std::mutex> lock(mutex_); |
387 | 0 | SnapshotState& snapshot = VERIFY_RESULT(FindSnapshot(snapshot_id)); |
388 | 0 | if (snapshot.schedule_id()) { |
389 | 0 | delete_sys_catalog_snapshot = true; |
390 | 0 | } |
391 | 0 | snapshot.SetInitialTabletsState(SysSnapshotEntryPB::DELETING); |
392 | 0 | RETURN_NOT_OK(snapshot.StoreToWriteBatch(&write_batch)); |
393 | 0 | if (leader_term >= 0) { |
394 | 0 | snapshot.PrepareOperations(&operations); |
395 | 0 | } |
396 | 0 | } |
397 | | |
398 | 0 | if (delete_sys_catalog_snapshot) { |
399 | 0 | RETURN_NOT_OK(operation.tablet()->snapshots().Delete(operation)); |
400 | 0 | } |
401 | | |
402 | 0 | RETURN_NOT_OK(operation.tablet()->ApplyOperation(operation, /* batch_idx= */ -1, write_batch)); |
403 | | |
404 | 0 | ExecuteOperations(operations, leader_term); |
405 | |
|
406 | 0 | return Status::OK(); |
407 | 0 | } |
408 | | |
409 | | CHECKED_STATUS RestoreSysCatalogReplicated( |
410 | 9 | int64_t leader_term, const tablet::SnapshotOperation& operation, Status* complete_status) { |
411 | 9 | auto restoration = std::make_shared<SnapshotScheduleRestoration>(SnapshotScheduleRestoration { |
412 | 9 | .snapshot_id = VERIFY_RESULT(FullyDecodeTxnSnapshotId(operation.request()->snapshot_id())), |
413 | 0 | .restore_at = HybridTime::FromPB(operation.request()->snapshot_hybrid_time()), |
414 | 9 | .restoration_id = VERIFY_RESULT(FullyDecodeTxnSnapshotRestorationId( |
415 | 0 | operation.request()->restoration_id())), |
416 | 0 | .op_id = operation.op_id(), |
417 | 9 | .write_time = operation.hybrid_time(), |
418 | 9 | .term = leader_term, |
419 | 9 | }); |
420 | 0 | { |
421 | 9 | std::lock_guard<std::mutex> lock(mutex_); |
422 | 9 | SnapshotState& snapshot = VERIFY_RESULT(FindSnapshot(restoration->snapshot_id)); |
423 | 9 | SnapshotScheduleState& schedule_state = VERIFY_RESULT( |
424 | 9 | FindSnapshotSchedule(snapshot.schedule_id())); |
425 | 0 | LOG(INFO) << "Restore sys catalog from snapshot: " << snapshot.ToString() << ", schedule: " |
426 | 9 | << schedule_state.ToString() << " at " << restoration->restore_at << ", op id: " |
427 | 9 | << restoration->op_id; |
428 | 9 | size_t this_idx = std::numeric_limits<size_t>::max(); |
429 | 9 | for (const auto& snapshot_schedule : schedules_) { |
430 | 9 | if (snapshot_schedule->id() == snapshot.schedule_id()) { |
431 | 9 | this_idx = restoration->schedules.size(); |
432 | 9 | } |
433 | 9 | restoration->schedules.emplace_back( |
434 | 9 | snapshot_schedule->id(), snapshot_schedule->options().filter()); |
435 | 9 | } |
436 | 9 | if (this_idx == std::numeric_limits<size_t>::max()) { |
437 | 0 | return STATUS_FORMAT(IllegalState, "Cannot find schedule for restoration: $0", |
438 | 0 | snapshot.schedule_id()); |
439 | 0 | } |
440 | 9 | std::swap(restoration->schedules[0], restoration->schedules[this_idx]); |
441 | 9 | if (leader_term >= 0) { |
442 | 3 | postponed_restores_.push_back(restoration); |
443 | 3 | } |
444 | 9 | } |
445 | 9 | LOG_SLOW_EXECUTION(INFO, 1000, "Restore sys catalog took") { |
446 | 9 | RETURN_NOT_OK_PREPEND( |
447 | 9 | context_.RestoreSysCatalog(restoration.get(), operation.tablet(), complete_status), |
448 | 9 | "Restore sys catalog failed"); |
449 | 9 | } |
450 | 9 | return Status::OK(); |
451 | 9 | } |
452 | | |
453 | | CHECKED_STATUS ListRestorations( |
454 | | const TxnSnapshotRestorationId& restoration_id, const TxnSnapshotId& snapshot_id, |
455 | 6 | ListSnapshotRestorationsResponsePB* resp) { |
456 | 6 | std::lock_guard<std::mutex> lock(mutex_); |
457 | 6 | if (!restoration_id) { |
458 | 1 | for (const auto& p : restorations_) { |
459 | 0 | if (!snapshot_id || p->snapshot_id() == snapshot_id) { |
460 | 0 | RETURN_NOT_OK(p->ToPB(resp->add_restorations())); |
461 | 0 | } |
462 | 0 | } |
463 | 1 | return Status::OK(); |
464 | 1 | } |
465 | | |
466 | 5 | RestorationState& restoration = VERIFY_RESULT(FindRestoration(restoration_id)); |
467 | 0 | return restoration.ToPB(resp->add_restorations()); |
468 | 5 | } |
469 | | |
470 | | Result<TxnSnapshotRestorationId> Restore( |
471 | 3 | const TxnSnapshotId& snapshot_id, HybridTime restore_at, int64_t leader_term) { |
472 | 3 | auto restoration_id = TxnSnapshotRestorationId::GenerateRandom(); |
473 | 3 | RETURN_NOT_OK(DoRestore( |
474 | 3 | snapshot_id, restore_at, restoration_id, {}, RestorePhase::kInitial, leader_term)); |
475 | 3 | return restoration_id; |
476 | 3 | } |
477 | | |
478 | | Result<SnapshotScheduleId> CreateSchedule( |
479 | 8 | const CreateSnapshotScheduleRequestPB& req, int64_t leader_term, CoarseTimePoint deadline) { |
480 | 8 | SnapshotScheduleState schedule(&context_, req); |
481 | | |
482 | 8 | docdb::KeyValueWriteBatchPB write_batch; |
483 | 8 | RETURN_NOT_OK(schedule.StoreToWriteBatch(&write_batch)); |
484 | | |
485 | 8 | RETURN_NOT_OK(SynchronizedWrite(std::move(write_batch), leader_term, deadline, &context_)); |
486 | | |
487 | 8 | return schedule.id(); |
488 | 8 | } |
489 | | |
490 | | CHECKED_STATUS ListSnapshotSchedules( |
491 | 11 | const SnapshotScheduleId& snapshot_schedule_id, ListSnapshotSchedulesResponsePB* resp) { |
492 | 11 | std::lock_guard<std::mutex> lock(mutex_); |
493 | 11 | if (snapshot_schedule_id.IsNil()) { |
494 | 1 | for (const auto& p : schedules_) { |
495 | 1 | RETURN_NOT_OK(FillSchedule(*p, resp->add_schedules())); |
496 | 1 | } |
497 | 1 | return Status::OK(); |
498 | 1 | } |
499 | | |
500 | 10 | SnapshotScheduleState& schedule = VERIFY_RESULT(FindSnapshotSchedule(snapshot_schedule_id)); |
501 | 0 | return FillSchedule(schedule, resp->add_schedules()); |
502 | 10 | } |
503 | | |
504 | | CHECKED_STATUS DeleteSnapshotSchedule( |
505 | | const SnapshotScheduleId& snapshot_schedule_id, int64_t leader_term, |
506 | 0 | CoarseTimePoint deadline) { |
507 | 0 | docdb::KeyValueWriteBatchPB write_batch; |
508 | 0 | { |
509 | 0 | std::lock_guard<std::mutex> lock(mutex_); |
510 | 0 | SnapshotScheduleState& schedule = VERIFY_RESULT(FindSnapshotSchedule(snapshot_schedule_id)); |
511 | 0 | auto encoded_key = VERIFY_RESULT(schedule.EncodedKey()); |
512 | 0 | auto pair = write_batch.add_write_pairs(); |
513 | 0 | pair->set_key(encoded_key.AsSlice().cdata(), encoded_key.size()); |
514 | 0 | auto options = schedule.options(); |
515 | 0 | options.set_delete_time(context_.Clock()->Now().ToUint64()); |
516 | 0 | auto* value = pair->mutable_value(); |
517 | 0 | value->push_back(docdb::ValueTypeAsChar::kString); |
518 | 0 | pb_util::AppendPartialToString(options, value); |
519 | 0 | } |
520 | | |
521 | 0 | return SynchronizedWrite(std::move(write_batch), leader_term, deadline, &context_); |
522 | 0 | } |
523 | | |
524 | 4.80M | CHECKED_STATUS FillHeartbeatResponse(TSHeartbeatResponsePB* resp) { |
525 | 4.80M | std::lock_guard<std::mutex> lock(mutex_); |
526 | 4.80M | auto* out = resp->mutable_snapshots_info(); |
527 | 4.80M | for (const auto& schedule : schedules_) { |
528 | | // Don't send deleted schedules. |
529 | 463 | if (schedule->deleted()) { |
530 | 0 | continue; |
531 | 0 | } |
532 | 463 | const auto& id = schedule->id(); |
533 | 463 | auto* out_schedule = out->add_schedules(); |
534 | 463 | out_schedule->set_id(id.data(), id.size()); |
535 | 463 | auto time = LastSnapshotTime(id); |
536 | 463 | if (time) { |
537 | 446 | out_schedule->set_last_snapshot_hybrid_time(time.ToUint64()); |
538 | 446 | } |
539 | 463 | } |
540 | 4.80M | out->set_last_restorations_update_ht(last_restorations_update_ht_.ToUint64()); |
541 | 4.80M | for (const auto& restoration : restorations_) { |
542 | 33 | auto* out_restoration = out->add_restorations(); |
543 | 33 | const auto& id = restoration->restoration_id(); |
544 | 33 | out_restoration->set_id(id.data(), id.size()); |
545 | 33 | auto complete_time = restoration->complete_time(); |
546 | 33 | if (complete_time) { |
547 | 9 | out_restoration->set_complete_time_ht(complete_time.ToUint64()); |
548 | 9 | } |
549 | 33 | } |
550 | 4.80M | return Status::OK(); |
551 | 4.80M | } |
552 | | |
553 | 3.00k | void SysCatalogLoaded(int64_t term) { |
554 | 3.00k | if (term == OpId::kUnknownTerm) { |
555 | | // Do nothing on follower. |
556 | 0 | return; |
557 | 0 | } |
558 | 3.00k | decltype(postponed_restores_) postponed_restores; |
559 | 3.00k | { |
560 | 3.00k | std::lock_guard<std::mutex> lock(mutex_); |
561 | 3.00k | auto filter = [term, &postponed_restores](const auto& restoration) { |
562 | 3 | if (restoration->term == term) { |
563 | 3 | postponed_restores.push_back(restoration); |
564 | 3 | } |
565 | | // TODO(pitr) cancel restorations |
566 | 3 | return restoration->term <= term; |
567 | 3 | }; |
568 | 3.00k | postponed_restores_.erase( |
569 | 3.00k | std::remove_if(postponed_restores_.begin(), postponed_restores_.end(), filter), |
570 | 3.00k | postponed_restores_.end()); |
571 | 3.00k | } |
572 | 3.00k | for (const auto& restoration : postponed_restores) { |
573 | | // TODO(pitr) Notify user about failures. |
574 | 3 | auto status = context_.VerifyRestoredObjects(*restoration); |
575 | 3 | LOG_IF(DFATAL, !status.ok()) << "Verify restoration failed: " << status0 ; |
576 | 3 | std::vector<TabletId> restore_tablets; |
577 | 17 | for (const auto& id_and_type : restoration->non_system_objects_to_restore) { |
578 | 17 | if (id_and_type.second == SysRowEntryType::TABLET) { |
579 | 10 | restore_tablets.push_back(id_and_type.first); |
580 | 10 | } |
581 | 17 | } |
582 | 3 | status = DoRestore(restoration->snapshot_id, restoration->restore_at, |
583 | 3 | restoration->restoration_id, restore_tablets, |
584 | 3 | RestorePhase::kPostSysCatalogLoad, term); |
585 | 3 | LOG_IF(DFATAL, !status.ok()) |
586 | 0 | << "Failed to restore tablets for restoration " |
587 | 0 | << restoration->restoration_id << ": " << status; |
588 | 3 | } |
589 | 3.00k | } |
590 | | |
591 | | Result<SnapshotSchedulesToObjectIdsMap> MakeSnapshotSchedulesToObjectIdsMap( |
592 | 15.4k | SysRowEntryType type) { |
593 | 15.4k | std::vector<std::pair<SnapshotScheduleId, SnapshotScheduleFilterPB>> schedules; |
594 | 15.4k | { |
595 | 15.4k | std::lock_guard<std::mutex> lock(mutex_); |
596 | 15.4k | for (const auto& schedule : schedules_) { |
597 | 17 | if (!schedule->deleted()) { |
598 | 17 | schedules.emplace_back(schedule->id(), schedule->options().filter()); |
599 | 17 | } |
600 | 17 | } |
601 | 15.4k | } |
602 | 15.4k | SnapshotSchedulesToObjectIdsMap result; |
603 | 15.4k | for (const auto& id_and_filter : schedules) { |
604 | 17 | auto entries = VERIFY_RESULT(CollectEntries(id_and_filter.second)); |
605 | 0 | auto& ids = result[id_and_filter.first]; |
606 | 73 | for (const auto& entry : entries.entries()) { |
607 | 73 | if (entry.type() == type) { |
608 | 32 | ids.push_back(entry.id()); |
609 | 32 | } |
610 | 73 | } |
611 | 17 | std::sort(ids.begin(), ids.end()); |
612 | 17 | } |
613 | 15.4k | return result; |
614 | 15.4k | } |
615 | | |
616 | 0 | Result<bool> IsTableCoveredBySomeSnapshotSchedule(const TableInfo& table_info) { |
617 | 0 | auto lock = table_info.LockForRead(); |
618 | 0 | { |
619 | 0 | std::lock_guard<std::mutex> l(mutex_); |
620 | 0 | for (const auto& schedule : schedules_) { |
621 | 0 | for (const auto& table_identifier : schedule->options().filter().tables().tables()) { |
622 | 0 | if (VERIFY_RESULT(TableMatchesIdentifier(table_info.id(), |
623 | 0 | lock->pb, |
624 | 0 | table_identifier))) { |
625 | 0 | return true; |
626 | 0 | } |
627 | 0 | } |
628 | 0 | } |
629 | 0 | } |
630 | 0 | return false; |
631 | 0 | } |
632 | | |
633 | 7.94k | void Start() { |
634 | 7.94k | { |
635 | 7.94k | std::lock_guard<std::mutex> lock(mutex_); |
636 | 7.94k | last_restorations_update_ht_ = context_.Clock()->Now(); |
637 | 7.94k | } |
638 | 7.94k | poller_.Start(&context_.Scheduler(), FLAGS_snapshot_coordinator_poll_interval_ms * 1ms); |
639 | 7.94k | } |
640 | | |
641 | 94 | void Shutdown() { |
642 | 94 | poller_.Shutdown(); |
643 | 94 | } |
644 | | |
645 | | private: |
646 | | template <class Pb, class Map> |
647 | 127 | CHECKED_STATUS LoadEntry(const Slice& id_slice, const Slice& data, Map* map) REQUIRES(mutex_) { |
648 | 127 | VLOG(2) << __func__ << "(" << id_slice.ToDebugString() << ", " << data.ToDebugString() << ")"0 ; |
649 | | |
650 | 127 | auto id = Uuid::TryFullyDecode(id_slice); |
651 | 127 | if (id.IsNil()) { |
652 | 0 | return Status::OK(); |
653 | 0 | } |
654 | 127 | auto metadata = VERIFY_RESULT(pb_util::ParseFromSlice<Pb>(data)); |
655 | 0 | return LoadEntry(typename Map::key_type(id), metadata, map); |
656 | 127 | } yb::Status yb::master::MasterSnapshotCoordinator::Impl::LoadEntry<yb::master::SysSnapshotEntryPB, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > >(yb::Slice const&, yb::Slice const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > >*) Line | Count | Source | 647 | 94 | CHECKED_STATUS LoadEntry(const Slice& id_slice, const Slice& data, Map* map) REQUIRES(mutex_) { | 648 | 94 | VLOG(2) << __func__ << "(" << id_slice.ToDebugString() << ", " << data.ToDebugString() << ")"0 ; | 649 | | | 650 | 94 | auto id = Uuid::TryFullyDecode(id_slice); | 651 | 94 | if (id.IsNil()) { | 652 | 0 | return Status::OK(); | 653 | 0 | } | 654 | 94 | auto metadata = VERIFY_RESULT(pb_util::ParseFromSlice<Pb>(data)); | 655 | 0 | return LoadEntry(typename Map::key_type(id), metadata, map); | 656 | 94 | } |
yb::Status yb::master::MasterSnapshotCoordinator::Impl::LoadEntry<yb::master::SnapshotScheduleOptionsPB, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > > >(yb::Slice const&, yb::Slice const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > >*) Line | Count | Source | 647 | 33 | CHECKED_STATUS LoadEntry(const Slice& id_slice, const Slice& data, Map* map) REQUIRES(mutex_) { | 648 | 33 | VLOG(2) << __func__ << "(" << id_slice.ToDebugString() << ", " << data.ToDebugString() << ")"0 ; | 649 | | | 650 | 33 | auto id = Uuid::TryFullyDecode(id_slice); | 651 | 33 | if (id.IsNil()) { | 652 | 0 | return Status::OK(); | 653 | 0 | } | 654 | 33 | auto metadata = VERIFY_RESULT(pb_util::ParseFromSlice<Pb>(data)); | 655 | 0 | return LoadEntry(typename Map::key_type(id), metadata, map); | 656 | 33 | } |
|
657 | | |
658 | | template <class Pb, class Map> |
659 | | CHECKED_STATUS LoadEntry( |
660 | | const typename Map::key_type& id, const Pb& data, Map* map) |
661 | 127 | REQUIRES(mutex_) { |
662 | 127 | VLOG(1) << __func__ << "(" << id << ", " << data.ShortDebugString() << ")"0 ; |
663 | | |
664 | 127 | auto new_entry = std::make_unique<typename Map::value_type::element_type>(&context_, id, data); |
665 | | |
666 | 127 | auto it = map->find(id); |
667 | 127 | if (it == map->end()) { |
668 | 24 | map->emplace(std::move(new_entry)); |
669 | 103 | } else if ((**it).ShouldUpdate(*new_entry)) { |
670 | 39 | map->replace(it, std::move(new_entry)); |
671 | 64 | } else { |
672 | 64 | VLOG_WITH_FUNC0 (1) << "Ignore because of version check, existing: " << (**it).ToString() |
673 | 0 | << ", loaded: " << new_entry->ToString(); |
674 | 64 | } |
675 | | |
676 | 127 | return Status::OK(); |
677 | 127 | } yb::Status yb::master::MasterSnapshotCoordinator::Impl::LoadEntry<yb::master::SysSnapshotEntryPB, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > >(boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > >::key_type const&, yb::master::SysSnapshotEntryPB const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > >*) Line | Count | Source | 661 | 94 | REQUIRES(mutex_) { | 662 | 94 | VLOG(1) << __func__ << "(" << id << ", " << data.ShortDebugString() << ")"0 ; | 663 | | | 664 | 94 | auto new_entry = std::make_unique<typename Map::value_type::element_type>(&context_, id, data); | 665 | | | 666 | 94 | auto it = map->find(id); | 667 | 94 | if (it == map->end()) { | 668 | 0 | map->emplace(std::move(new_entry)); | 669 | 94 | } else if ((**it).ShouldUpdate(*new_entry)) { | 670 | 30 | map->replace(it, std::move(new_entry)); | 671 | 64 | } else { | 672 | 64 | VLOG_WITH_FUNC0 (1) << "Ignore because of version check, existing: " << (**it).ToString() | 673 | 0 | << ", loaded: " << new_entry->ToString(); | 674 | 64 | } | 675 | | | 676 | 94 | return Status::OK(); | 677 | 94 | } |
yb::Status yb::master::MasterSnapshotCoordinator::Impl::LoadEntry<yb::master::SnapshotScheduleOptionsPB, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > > >(boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > >::key_type const&, yb::master::SnapshotScheduleOptionsPB const&, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > >*) Line | Count | Source | 661 | 33 | REQUIRES(mutex_) { | 662 | 33 | VLOG(1) << __func__ << "(" << id << ", " << data.ShortDebugString() << ")"0 ; | 663 | | | 664 | 33 | auto new_entry = std::make_unique<typename Map::value_type::element_type>(&context_, id, data); | 665 | | | 666 | 33 | auto it = map->find(id); | 667 | 33 | if (it == map->end()) { | 668 | 24 | map->emplace(std::move(new_entry)); | 669 | 24 | } else if (9 (**it).ShouldUpdate(*new_entry)9 ) { | 670 | 9 | map->replace(it, std::move(new_entry)); | 671 | 9 | } else { | 672 | 0 | VLOG_WITH_FUNC(1) << "Ignore because of version check, existing: " << (**it).ToString() | 673 | 0 | << ", loaded: " << new_entry->ToString(); | 674 | 0 | } | 675 | | | 676 | 33 | return Status::OK(); | 677 | 33 | } |
|
678 | | |
679 | 23 | Result<SnapshotState&> FindSnapshot(const TxnSnapshotId& snapshot_id) REQUIRES(mutex_) { |
680 | 23 | auto it = snapshots_.find(snapshot_id); |
681 | 23 | if (it == snapshots_.end()) { |
682 | 0 | return STATUS(NotFound, "Could not find snapshot", snapshot_id.ToString(), |
683 | 0 | MasterError(MasterErrorPB::SNAPSHOT_NOT_FOUND)); |
684 | 0 | } |
685 | 23 | return **it; |
686 | 23 | } |
687 | | |
688 | | Result<RestorationState&> FindRestoration( |
689 | 8 | const TxnSnapshotRestorationId& restoration_id) REQUIRES(mutex_) { |
690 | 8 | auto it = restorations_.find(restoration_id); |
691 | 8 | if (it == restorations_.end()) { |
692 | 0 | return STATUS(NotFound, "Could not find restoration", restoration_id.ToString(), |
693 | 0 | MasterError(MasterErrorPB::OBJECT_NOT_FOUND)); |
694 | 0 | } |
695 | 8 | return **it; |
696 | 8 | } |
697 | | |
698 | | Result<SnapshotScheduleState&> FindSnapshotSchedule( |
699 | 22 | const SnapshotScheduleId& id) REQUIRES(mutex_) { |
700 | 22 | auto it = schedules_.find(id); |
701 | 22 | if (it == schedules_.end()) { |
702 | 0 | return STATUS(NotFound, "Could not find snapshot schedule", id.ToString(), |
703 | 0 | MasterError(MasterErrorPB::SNAPSHOT_NOT_FOUND)); |
704 | 0 | } |
705 | 22 | return **it; |
706 | 22 | } |
707 | | |
708 | 310k | void ExecuteOperations(const TabletSnapshotOperations& operations, int64_t leader_term) { |
709 | 310k | if (operations.empty()) { |
710 | 310k | return; |
711 | 310k | } |
712 | 11 | VLOG(4) << __func__ << "(" << AsString(operations) << ")"0 ; |
713 | | |
714 | 11 | size_t num_operations = operations.size(); |
715 | 11 | LOG(INFO) << "Number of snapshot operations to be executed " << num_operations; |
716 | 11 | std::vector<TabletId> tablet_ids; |
717 | 11 | tablet_ids.reserve(num_operations); |
718 | 25 | for (const auto& operation : operations) { |
719 | 25 | tablet_ids.push_back(operation.tablet_id); |
720 | 25 | } |
721 | 11 | auto tablet_infos = context_.GetTabletInfos(tablet_ids); |
722 | 36 | for (size_t i = 0; i != num_operations; ++i25 ) { |
723 | 25 | ExecuteOperation(operations[i], tablet_infos[i], leader_term); |
724 | 25 | } |
725 | 11 | } |
726 | | |
727 | | void ExecuteOperation( |
728 | | const TabletSnapshotOperation& operation, const TabletInfoPtr& tablet_info, |
729 | 25 | int64_t leader_term) { |
730 | 25 | auto callback = MakeDoneCallback( |
731 | 25 | &mutex_, snapshots_, operation.snapshot_id, operation.tablet_id, |
732 | 25 | std::bind(&Impl::UpdateSnapshot, this, _1, leader_term, _2)); |
733 | 25 | if (!tablet_info) { |
734 | 0 | callback(STATUS_EC_FORMAT(NotFound, MasterError(MasterErrorPB::TABLET_NOT_RUNNING), |
735 | 0 | "Tablet info not found for $0", operation.tablet_id)); |
736 | 0 | return; |
737 | 0 | } |
738 | 25 | auto snapshot_id_str = operation.snapshot_id.AsSlice().ToBuffer(); |
739 | | |
740 | 25 | if (operation.state == SysSnapshotEntryPB::DELETING) { |
741 | 0 | auto task = context_.CreateAsyncTabletSnapshotOp( |
742 | 0 | tablet_info, snapshot_id_str, tserver::TabletSnapshotOpRequestPB::DELETE_ON_TABLET, |
743 | 0 | callback); |
744 | 0 | context_.ScheduleTabletSnapshotOp(task); |
745 | 25 | } else if (operation.state == SysSnapshotEntryPB::CREATING) { |
746 | 25 | auto task = context_.CreateAsyncTabletSnapshotOp( |
747 | 25 | tablet_info, snapshot_id_str, tserver::TabletSnapshotOpRequestPB::CREATE_ON_TABLET, |
748 | 25 | callback); |
749 | 25 | task->SetSnapshotScheduleId(operation.schedule_id); |
750 | 25 | task->SetSnapshotHybridTime(operation.snapshot_hybrid_time); |
751 | 25 | context_.ScheduleTabletSnapshotOp(task); |
752 | 25 | } else { |
753 | 0 | LOG(DFATAL) << "Unsupported snapshot operation: " << operation.ToString(); |
754 | 0 | } |
755 | 25 | } |
756 | | |
757 | | struct PollSchedulesData { |
758 | | std::vector<TxnSnapshotId> delete_snapshots; |
759 | | SnapshotScheduleOperations schedule_operations; |
760 | | ScheduleMinRestoreTime schedule_min_restore_time; |
761 | | }; |
762 | | |
763 | 695k | void Poll() { |
764 | 695k | auto leader_term = context_.LeaderTerm(); |
765 | 695k | if (leader_term < 0) { |
766 | 384k | return; |
767 | 384k | } |
768 | 310k | VLOG(4) << __func__ << "()"3 ; |
769 | 310k | std::vector<TxnSnapshotId> cleanup_snapshots; |
770 | 310k | TabletSnapshotOperations operations; |
771 | 310k | PollSchedulesData schedules_data; |
772 | 310k | { |
773 | 310k | std::lock_guard<std::mutex> lock(mutex_); |
774 | 310k | for (const auto& p : snapshots_) { |
775 | 249 | if (p->NeedCleanup()) { |
776 | 0 | LOG(INFO) << "Cleanup of snapshot " << p->id() << " started."; |
777 | 0 | if (!p->CleanupTracker().Start().ok()) { |
778 | 0 | LOG(DFATAL) << "Cleanup of snapshot " << p->id() << " was already started."; |
779 | 0 | } |
780 | 0 | cleanup_snapshots.push_back(p->id()); |
781 | 249 | } else { |
782 | | // Refresh the throttle limit. |
783 | 249 | p->Throttler().RefreshLimit( |
784 | 249 | GetRpcLimit(FLAGS_max_concurrent_snapshot_rpcs, |
785 | 249 | FLAGS_max_concurrent_snapshot_rpcs_per_tserver, leader_term)); |
786 | 249 | p->PrepareOperations(&operations); |
787 | 249 | } |
788 | 249 | } |
789 | 310k | PollSchedulesPrepare(&schedules_data); |
790 | 310k | } |
791 | 310k | for (const auto& id : cleanup_snapshots) { |
792 | 0 | CleanupObject(leader_term, id, snapshots_, EncodedSnapshotKey(id, &context_)); |
793 | 0 | } |
794 | 310k | ExecuteOperations(operations, leader_term); |
795 | 310k | PollSchedulesComplete(schedules_data, leader_term); |
796 | 310k | } |
797 | | |
798 | 0 | void TryDeleteSnapshot(SnapshotState* snapshot, PollSchedulesData* data) { |
799 | 0 | auto delete_status = snapshot->TryStartDelete(); |
800 | 0 | if (!delete_status.ok()) { |
801 | 0 | VLOG(1) << "Unable to delete snapshot " << snapshot->id() << "/" << snapshot->schedule_id() |
802 | 0 | << ": " << delete_status << ", state: " << snapshot->ToString(); |
803 | 0 | return; |
804 | 0 | } |
805 | | |
806 | 0 | VLOG(1) << "Cleanup snapshot: " << snapshot->id() << "/" << snapshot->schedule_id(); |
807 | 0 | data->delete_snapshots.push_back(snapshot->id()); |
808 | 0 | } |
809 | | |
810 | 310k | void PollSchedulesPrepare(PollSchedulesData* data) REQUIRES(mutex_) { |
811 | 310k | auto now = context_.Clock()->Now(); |
812 | 310k | for (const auto& p : schedules_) { |
813 | 173 | HybridTime last_snapshot_time; |
814 | 173 | if (p->deleted()) { |
815 | 0 | auto range = snapshots_.get<ScheduleTag>().equal_range(p->id()); |
816 | 0 | for (const auto& snapshot : boost::make_iterator_range(range.first, range.second)) { |
817 | 0 | TryDeleteSnapshot(snapshot.get(), data); |
818 | 0 | } |
819 | 173 | } else { |
820 | 173 | auto& index = snapshots_.get<ScheduleTag>(); |
821 | 173 | auto range = index.equal_range(p->id()); |
822 | 173 | if (range.first != range.second) { |
823 | 165 | --range.second; |
824 | 165 | for (; range.first != range.second; ++range.first0 ) { |
825 | 53 | if ((**range.first).initial_state() != SysSnapshotEntryPB::DELETING) { |
826 | 53 | break; |
827 | 53 | } |
828 | 53 | } |
829 | 165 | auto& first_snapshot = **range.first; |
830 | 165 | data->schedule_min_restore_time[p->id()] = |
831 | 165 | first_snapshot.previous_snapshot_hybrid_time() |
832 | 165 | ? first_snapshot.previous_snapshot_hybrid_time()0 |
833 | 165 | : first_snapshot.snapshot_hybrid_time(); |
834 | 165 | auto gc_limit = now.AddSeconds(-p->options().retention_duration_sec()); |
835 | 165 | VLOG_WITH_FUNC0 (4) << "Gc limit: " << gc_limit0 ; |
836 | 165 | for (; range.first != range.second; ++range.first0 ) { |
837 | 53 | if ((**range.first).snapshot_hybrid_time() >= gc_limit) { |
838 | 53 | break; |
839 | 53 | } |
840 | 0 | TryDeleteSnapshot(range.first->get(), data); |
841 | 0 | } |
842 | 165 | last_snapshot_time = (**range.second).snapshot_hybrid_time(); |
843 | 165 | } |
844 | 173 | } |
845 | 173 | p->PrepareOperations(last_snapshot_time, now, &data->schedule_operations); |
846 | 173 | } |
847 | 310k | } |
848 | | |
849 | 310k | void PollSchedulesComplete(const PollSchedulesData& data, int64_t leader_term) EXCLUDES(mutex_) { |
850 | 310k | for (const auto& id : data.delete_snapshots) { |
851 | 0 | SubmitDelete(id, leader_term, nullptr); |
852 | 0 | } |
853 | 310k | for (const auto& operation : data.schedule_operations) { |
854 | 18 | switch (operation.type) { |
855 | 18 | case SnapshotScheduleOperationType::kCreateSnapshot: |
856 | 18 | WARN_NOT_OK(ExecuteScheduleOperation(operation, leader_term), |
857 | 18 | Format("Failed to execute operation on $0", operation.schedule_id)); |
858 | 18 | break; |
859 | 0 | case SnapshotScheduleOperationType::kCleanup: |
860 | 0 | CleanupObject( |
861 | 0 | leader_term, operation.schedule_id, schedules_, |
862 | 0 | SnapshotScheduleState::EncodedKey(operation.schedule_id, &context_)); |
863 | 0 | break; |
864 | 0 | default: |
865 | 0 | LOG(DFATAL) << "Unexpected operation type: " << operation.type; |
866 | 0 | break; |
867 | 18 | } |
868 | 18 | } |
869 | 310k | context_.CleanupHiddenObjects(data.schedule_min_restore_time); |
870 | 310k | } |
871 | | |
872 | | SnapshotState* BoundingSnapshot(const SnapshotScheduleId& schedule_id, Bound bound) |
873 | 464 | REQUIRES(mutex_) { |
874 | 464 | auto& index = snapshots_.get<ScheduleTag>(); |
875 | 464 | decltype(index.begin()) it; |
876 | 464 | if (bound == Bound::kFirst) { |
877 | 0 | it = index.lower_bound(schedule_id); |
878 | 0 | if (it == index.end()) { |
879 | 0 | return nullptr; |
880 | 0 | } |
881 | 464 | } else { |
882 | 464 | it = index.upper_bound(schedule_id); |
883 | 464 | if (it == index.begin()) { |
884 | 17 | return nullptr; |
885 | 17 | } |
886 | 447 | --it; |
887 | 447 | } |
888 | 447 | return (**it).schedule_id() == schedule_id ? it->get() : nullptr0 ; |
889 | 464 | } |
890 | | |
891 | 463 | HybridTime LastSnapshotTime(const SnapshotScheduleId& schedule_id) REQUIRES(mutex_) { |
892 | 463 | auto snapshot = BoundingSnapshot(schedule_id, Bound::kLast); |
893 | 463 | return snapshot ? snapshot->snapshot_hybrid_time()446 : HybridTime::kInvalid17 ; |
894 | 463 | } |
895 | | |
896 | | template <typename Id, typename Map> |
897 | 0 | void CleanupObjectAborted(Id id, const Map& map) { |
898 | 0 | LOG(INFO) << "Aborting cleanup of object " << id; |
899 | 0 | std::lock_guard<std::mutex> l(mutex_); |
900 | 0 | auto it = map.find(id); |
901 | 0 | if (it == map.end()) { |
902 | 0 | return; |
903 | 0 | } |
904 | 0 | (**it).CleanupTracker().Abort(); |
905 | 0 | } Unexecuted instantiation: void yb::master::MasterSnapshotCoordinator::Impl::CleanupObjectAborted<yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag>, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > >(yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag>, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > const&) Unexecuted instantiation: void yb::master::MasterSnapshotCoordinator::Impl::CleanupObjectAborted<yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag>, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > > >(yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag>, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > > const&) |
906 | | |
907 | | template <typename Map, typename Id> |
908 | | void CleanupObject(int64_t leader_term, Id id, const Map& map, |
909 | 0 | const Result<docdb::KeyBytes>& encoded_key) { |
910 | 0 | if (!encoded_key.ok()) { |
911 | 0 | LOG(DFATAL) << "Failed to encode id for deletion: " << encoded_key.status(); |
912 | 0 | return; |
913 | 0 | } |
914 | | |
915 | 0 | auto query = std::make_unique<tablet::WriteQuery>( |
916 | 0 | leader_term, CoarseMonoClock::Now() + FLAGS_sys_catalog_write_timeout_ms * 1ms, |
917 | 0 | nullptr /* context */, nullptr /* tablet */); |
918 | |
|
919 | 0 | auto* write_batch = query->operation().AllocateRequest()->mutable_write_batch(); |
920 | 0 | auto pair = write_batch->add_write_pairs(); |
921 | 0 | pair->set_key((*encoded_key).AsSlice().cdata(), (*encoded_key).size()); |
922 | 0 | char value = { docdb::ValueTypeAsChar::kTombstone }; |
923 | 0 | pair->set_value(&value, 1); |
924 | |
|
925 | 0 | query->set_callback([this, id, &map](const Status& s) { |
926 | 0 | if (s.ok()) { |
927 | 0 | LOG(INFO) << "Finished cleanup of object " << id; |
928 | 0 | return; |
929 | 0 | } |
930 | 0 | CleanupObjectAborted(id, map); |
931 | 0 | }); Unexecuted instantiation: void yb::master::MasterSnapshotCoordinator::Impl::CleanupObject<boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > >, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> >(long long, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag>, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > const&, yb::Result<yb::docdb::KeyBytes> const&)::'lambda'(yb::Status const&)::operator()(yb::Status const&) const Unexecuted instantiation: void yb::master::MasterSnapshotCoordinator::Impl::CleanupObject<boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > >, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> >(long long, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag>, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > > const&, yb::Result<yb::docdb::KeyBytes> const&)::'lambda'(yb::Status const&)::operator()(yb::Status const&) const |
932 | |
|
933 | 0 | context_.Submit(query.release()->PrepareSubmit(), leader_term); |
934 | 0 | } Unexecuted instantiation: void yb::master::MasterSnapshotCoordinator::Impl::CleanupObject<boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > >, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> >(long long, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag>, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::TxnSnapshotId_Tag> const&, &(yb::master::SnapshotState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::ordered_non_unique<boost::multi_index::tag<yb::master::MasterSnapshotCoordinator::Impl::ScheduleTag, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, boost::multi_index::composite_key<yb::master::SnapshotState, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotState::schedule_id() const)>, boost::multi_index::const_mem_fun<yb::master::SnapshotState, yb::HybridTime, &(yb::master::SnapshotState::snapshot_hybrid_time() const)>, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type, boost::tuples::null_type>, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotState, std::__1::default_delete<yb::master::SnapshotState> > > > const&, yb::Result<yb::docdb::KeyBytes> const&) Unexecuted instantiation: void yb::master::MasterSnapshotCoordinator::Impl::CleanupObject<boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > >, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> >(long long, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag>, boost::multi_index::multi_index_container<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> >, boost::multi_index::indexed_by<boost::multi_index::hashed_unique<boost::multi_index::const_mem_fun<yb::master::SnapshotScheduleState, yb::StronglyTypedUuid<yb::SnapshotScheduleId_Tag> const&, &(yb::master::SnapshotScheduleState::id() const)>, mpl_::na, mpl_::na, mpl_::na>, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na, mpl_::na>, std::__1::allocator<std::__1::unique_ptr<yb::master::SnapshotScheduleState, std::__1::default_delete<yb::master::SnapshotScheduleState> > > > const&, yb::Result<yb::docdb::KeyBytes> const&) |
935 | | |
936 | | CHECKED_STATUS ExecuteScheduleOperation( |
937 | | const SnapshotScheduleOperation& operation, int64_t leader_term, |
938 | 19 | const std::weak_ptr<Synchronizer>& synchronizer = std::weak_ptr<Synchronizer>()) { |
939 | 19 | auto entries = CollectEntries(operation.filter); |
940 | 19 | VLOG(2) << __func__ << "(" << AsString(operation) << ", " << leader_term << "), entries: " |
941 | 0 | << AsString(entries); |
942 | 19 | if (!entries.ok()) { |
943 | 0 | CreateSnapshotAborted(entries.status(), operation.schedule_id, operation.snapshot_id); |
944 | 0 | return entries.status(); |
945 | 0 | } |
946 | 19 | SubmitCreate( |
947 | 19 | *entries, /* imported= */ false, operation.schedule_id, |
948 | 19 | operation.previous_snapshot_hybrid_time, operation.snapshot_id, leader_term, |
949 | 19 | [this, schedule_id = operation.schedule_id, snapshot_id = operation.snapshot_id, |
950 | 19 | synchronizer]( |
951 | 19 | const Status& status) { |
952 | 15 | if (!status.ok()) { |
953 | 0 | CreateSnapshotAborted(status, schedule_id, snapshot_id); |
954 | 0 | } |
955 | 15 | auto locked_synchronizer = synchronizer.lock(); |
956 | 15 | if (locked_synchronizer) { |
957 | 1 | locked_synchronizer->StatusCB(status); |
958 | 1 | } |
959 | 15 | }); |
960 | 19 | return Status::OK(); |
961 | 19 | } |
962 | | |
963 | | void CreateSnapshotAborted( |
964 | | const Status& status, const SnapshotScheduleId& schedule_id, |
965 | 0 | const TxnSnapshotId& snapshot_id) { |
966 | 0 | LOG(INFO) << __func__ << " for " << schedule_id << ", snapshot: " << snapshot_id |
967 | 0 | << ", status: " << status; |
968 | 0 | std::lock_guard<std::mutex> lock(mutex_); |
969 | 0 | auto it = schedules_.find(schedule_id); |
970 | 0 | if (it == schedules_.end()) { |
971 | 0 | return; |
972 | 0 | } |
973 | 0 | (**it).SnapshotFinished(snapshot_id, status); |
974 | 0 | } |
975 | | |
976 | | void SubmitCreate( |
977 | | const SysRowEntries& entries, bool imported, const SnapshotScheduleId& schedule_id, |
978 | | HybridTime previous_snapshot_hybrid_time, TxnSnapshotId snapshot_id, int64_t leader_term, |
979 | 21 | tablet::OperationCompletionCallback completion_clbk) { |
980 | 21 | auto operation = std::make_unique<tablet::SnapshotOperation>(/* tablet= */ nullptr); |
981 | 21 | auto request = operation->AllocateRequest(); |
982 | | |
983 | 21 | VLOG(1) << __func__ << "(" << AsString(entries) << ", " << imported << ", " << schedule_id |
984 | 0 | << ", " << snapshot_id << ")"; |
985 | 61 | for (const auto& entry : entries.entries()) { |
986 | 61 | if (entry.type() == SysRowEntryType::TABLET) { |
987 | 28 | request->add_tablet_id(entry.id()); |
988 | 28 | } |
989 | 61 | } |
990 | | |
991 | 21 | request->set_snapshot_hybrid_time(context_.Clock()->MaxGlobalNow().ToUint64()); |
992 | 21 | request->set_operation(tserver::TabletSnapshotOpRequestPB::CREATE_ON_MASTER); |
993 | 21 | request->set_snapshot_id(snapshot_id.data(), snapshot_id.size()); |
994 | 21 | request->set_imported(imported); |
995 | 21 | if (schedule_id) { |
996 | 19 | request->set_schedule_id(schedule_id.data(), schedule_id.size()); |
997 | 19 | } |
998 | 21 | if (previous_snapshot_hybrid_time) { |
999 | 11 | request->set_previous_snapshot_hybrid_time(previous_snapshot_hybrid_time.ToUint64()); |
1000 | 11 | } |
1001 | | |
1002 | 21 | request->mutable_extra_data()->PackFrom(entries); |
1003 | | |
1004 | 21 | operation->set_completion_callback(std::move(completion_clbk)); |
1005 | | |
1006 | 21 | context_.Submit(std::move(operation), leader_term); |
1007 | 21 | } |
1008 | | |
1009 | | void SubmitDelete(const TxnSnapshotId& snapshot_id, int64_t leader_term, |
1010 | 0 | const std::shared_ptr<Synchronizer>& synchronizer) { |
1011 | 0 | auto operation = std::make_unique<tablet::SnapshotOperation>(nullptr); |
1012 | 0 | auto request = operation->AllocateRequest(); |
1013 | |
|
1014 | 0 | request->set_operation(tserver::TabletSnapshotOpRequestPB::DELETE_ON_MASTER); |
1015 | 0 | request->set_snapshot_id(snapshot_id.data(), snapshot_id.size()); |
1016 | |
|
1017 | 0 | operation->set_completion_callback( |
1018 | 0 | [this, wsynchronizer = std::weak_ptr<Synchronizer>(synchronizer), snapshot_id] |
1019 | 0 | (const Status& status) { |
1020 | 0 | auto synchronizer = wsynchronizer.lock(); |
1021 | 0 | if (synchronizer) { |
1022 | 0 | synchronizer->StatusCB(status); |
1023 | 0 | } |
1024 | 0 | if (!status.ok()) { |
1025 | 0 | DeleteSnapshotAborted(status, snapshot_id); |
1026 | 0 | } |
1027 | 0 | }); |
1028 | |
|
1029 | 0 | context_.Submit(std::move(operation), leader_term); |
1030 | 0 | } |
1031 | | |
1032 | | CHECKED_STATUS SubmitRestore( |
1033 | | const TxnSnapshotId& snapshot_id, HybridTime restore_at, |
1034 | 3 | const TxnSnapshotRestorationId& restoration_id, int64_t leader_term) { |
1035 | 3 | auto synchronizer = std::make_shared<Synchronizer>(); |
1036 | | |
1037 | 3 | auto operation = std::make_unique<tablet::SnapshotOperation>(nullptr); |
1038 | 3 | auto request = operation->AllocateRequest(); |
1039 | | |
1040 | 3 | request->set_operation(tserver::TabletSnapshotOpRequestPB::RESTORE_SYS_CATALOG); |
1041 | 3 | request->set_snapshot_id(snapshot_id.data(), snapshot_id.size()); |
1042 | 3 | request->set_snapshot_hybrid_time(restore_at.ToUint64()); |
1043 | 3 | if (restoration_id) { |
1044 | 3 | request->set_restoration_id(restoration_id.data(), restoration_id.size()); |
1045 | 3 | } |
1046 | | |
1047 | 3 | operation->set_completion_callback( |
1048 | 3 | tablet::MakeWeakSynchronizerOperationCompletionCallback(synchronizer)); |
1049 | | |
1050 | 3 | context_.Submit(std::move(operation), leader_term); |
1051 | | |
1052 | 3 | return synchronizer->Wait(); |
1053 | 3 | } |
1054 | | |
1055 | | void DeleteSnapshotAborted( |
1056 | 0 | const Status& status, const TxnSnapshotId& snapshot_id) { |
1057 | 0 | LOG(INFO) << __func__ << ", snapshot: " << snapshot_id << ", status: " << status; |
1058 | 0 | std::lock_guard<std::mutex> lock(mutex_); |
1059 | 0 | auto it = snapshots_.find(snapshot_id); |
1060 | 0 | if (it == snapshots_.end()) { |
1061 | 0 | return; |
1062 | 0 | } |
1063 | 0 | (**it).DeleteAborted(status); |
1064 | 0 | } |
1065 | | |
1066 | | void UpdateSnapshot( |
1067 | | SnapshotState* snapshot, int64_t leader_term, std::unique_lock<std::mutex>* lock) |
1068 | 33 | REQUIRES(mutex_) { |
1069 | 33 | bool batch_done = false; |
1070 | 33 | bool is_empty = snapshot->Empty(); |
1071 | | |
1072 | 33 | if (!is_empty) { |
1073 | 25 | batch_done = snapshot->Throttler().RemoveOutstandingTask(); |
1074 | 25 | } |
1075 | 33 | if (!snapshot->AllTabletsDone()) { |
1076 | 16 | if (FLAGS_schedule_snapshot_rpcs_out_of_band && batch_done && !is_empty2 ) { |
1077 | | // Send another batch. This prevents having to wait for the regular cycle |
1078 | | // of master snapshot coordinator which can be too slow. |
1079 | 2 | context_.Scheduler().io_service().post([this]() { |
1080 | 2 | LOG(INFO) << "Rescheduling Snapshot RPCs out of band."; |
1081 | 2 | Poll(); |
1082 | 2 | }); |
1083 | 2 | } |
1084 | 16 | return; |
1085 | 16 | } |
1086 | | |
1087 | 17 | if (snapshot->schedule_id()) { |
1088 | 15 | UpdateSchedule(*snapshot); |
1089 | 15 | } |
1090 | | |
1091 | 17 | docdb::KeyValueWriteBatchPB write_batch; |
1092 | 17 | auto status = snapshot->StoreToWriteBatch(&write_batch); |
1093 | 17 | if (!status.ok()) { |
1094 | 0 | LOG(DFATAL) << "Failed to prepare write batch for snapshot: " << status; |
1095 | 0 | return; |
1096 | 0 | } |
1097 | 17 | lock->unlock(); |
1098 | | |
1099 | 17 | SubmitWrite(std::move(write_batch), leader_term, &context_); |
1100 | 17 | }; |
1101 | | |
1102 | 10 | void FinishRestoration(RestorationState* restoration, int64_t leader_term) REQUIRES(mutex_) { |
1103 | 10 | if (!restoration->AllTabletsDone()) { |
1104 | 7 | return; |
1105 | 7 | } |
1106 | | |
1107 | 3 | last_restorations_update_ht_ = context_.Clock()->Now(); |
1108 | 3 | restoration->set_complete_time(last_restorations_update_ht_); |
1109 | | |
1110 | 3 | LOG(INFO) << "Setting restore complete time to " << last_restorations_update_ht_; |
1111 | | |
1112 | 3 | if (restoration->schedule_id()) { |
1113 | 3 | auto schedule = FindSnapshotSchedule(restoration->schedule_id()); |
1114 | 3 | if (schedule.ok()) { |
1115 | 3 | docdb::KeyValueWriteBatchPB write_batch; |
1116 | 3 | schedule->mutable_options().add_restoration_times( |
1117 | 3 | last_restorations_update_ht_.ToUint64()); |
1118 | 3 | Status s = schedule->StoreToWriteBatch(&write_batch); |
1119 | 3 | if (s.ok()) { |
1120 | 3 | SubmitWrite(std::move(write_batch), leader_term, &context_); |
1121 | 3 | } else { |
1122 | 0 | LOG(INFO) << "Unable to prepare write batch for schedule " |
1123 | 0 | << schedule->id(); |
1124 | 0 | } |
1125 | 3 | } else { |
1126 | 0 | LOG(INFO) << "Snapshot Schedule with id " << restoration->schedule_id() |
1127 | 0 | << " not found"; |
1128 | 0 | } |
1129 | 3 | } |
1130 | | |
1131 | 3 | if (FLAGS_TEST_skip_sending_restore_finished) { |
1132 | 0 | return; |
1133 | 0 | } |
1134 | | |
1135 | 3 | auto temp_ids = restoration->tablet_ids(); |
1136 | 3 | std::vector<TabletId> tablet_ids(temp_ids.begin(), temp_ids.end()); |
1137 | 3 | auto tablets = context_.GetTabletInfos(tablet_ids); |
1138 | 10 | for (const auto& tablet : tablets) { |
1139 | 10 | auto task = context_.CreateAsyncTabletSnapshotOp( |
1140 | 10 | tablet, std::string(), tserver::TabletSnapshotOpRequestPB::RESTORE_FINISHED, |
1141 | 10 | /* callback= */ nullptr); |
1142 | 10 | task->SetRestorationId(restoration->restoration_id()); |
1143 | 10 | task->SetRestorationTime(restoration->complete_time()); |
1144 | 10 | context_.ScheduleTabletSnapshotOp(task); |
1145 | 10 | } |
1146 | 3 | } |
1147 | | |
1148 | 15 | void UpdateSchedule(const SnapshotState& snapshot) REQUIRES(mutex_) { |
1149 | 15 | auto it = schedules_.find(snapshot.schedule_id()); |
1150 | 15 | if (it == schedules_.end()) { |
1151 | 0 | return; |
1152 | 0 | } |
1153 | | |
1154 | 15 | auto state = snapshot.AggregatedState(); |
1155 | 15 | Status status; |
1156 | 15 | if (!state.ok()) { |
1157 | 0 | status = state.status(); |
1158 | 15 | } else { |
1159 | 15 | switch (*state) { |
1160 | 15 | case SysSnapshotEntryPB::COMPLETE: |
1161 | 15 | status = Status::OK(); |
1162 | 15 | break; |
1163 | 0 | case SysSnapshotEntryPB::FAILED: |
1164 | 0 | status = snapshot.AnyFailure(); |
1165 | 0 | break; |
1166 | 0 | case SysSnapshotEntryPB::DELETED: |
1167 | 0 | return; |
1168 | 0 | default: |
1169 | 0 | LOG(DFATAL) << "Unexpected snapshot state: " << *state << " for " << snapshot.id(); |
1170 | 0 | return; |
1171 | 15 | } |
1172 | 15 | } |
1173 | 15 | (**it).SnapshotFinished(snapshot.id(), status); |
1174 | 15 | } |
1175 | | |
1176 | | CHECKED_STATUS FillSchedule(const SnapshotScheduleState& schedule, SnapshotScheduleInfoPB* out) |
1177 | 11 | REQUIRES(mutex_) { |
1178 | 11 | RETURN_NOT_OK(schedule.ToPB(out)); |
1179 | 11 | const auto& index = snapshots_.get<ScheduleTag>(); |
1180 | 11 | auto p = index.equal_range(boost::make_tuple(schedule.id())); |
1181 | 27 | for (auto i = p.first; i != p.second; ++i16 ) { |
1182 | 16 | RETURN_NOT_OK((**i).ToPB(out->add_snapshots())); |
1183 | 16 | } |
1184 | 11 | return Status::OK(); |
1185 | 11 | } |
1186 | | |
1187 | 36 | Result<SysRowEntries> CollectEntries(const SnapshotScheduleFilterPB& filter) { |
1188 | 36 | return context_.CollectEntriesForSnapshot(filter.tables().tables()); |
1189 | 36 | } |
1190 | | |
1191 | | CHECKED_STATUS DoRestore( |
1192 | | const TxnSnapshotId& snapshot_id, HybridTime restore_at, |
1193 | | const TxnSnapshotRestorationId& restoration_id, const std::vector<TabletId>& restore_tablets, |
1194 | 6 | RestorePhase phase, int64_t leader_term) { |
1195 | 6 | TabletInfos tablet_infos; |
1196 | 6 | bool restore_sys_catalog; |
1197 | 6 | std::unordered_set<TabletId> snapshot_tablets; |
1198 | 6 | { |
1199 | 6 | std::lock_guard<std::mutex> lock(mutex_); |
1200 | 6 | SnapshotState& snapshot = VERIFY_RESULT(FindSnapshot(snapshot_id)); |
1201 | 6 | if (!VERIFY_RESULT(snapshot.Complete())) { |
1202 | 0 | return STATUS(IllegalState, "The snapshot state is not complete", snapshot_id.ToString(), |
1203 | 0 | MasterError(MasterErrorPB::SNAPSHOT_IS_NOT_READY)); |
1204 | 0 | } |
1205 | 6 | restore_sys_catalog = phase == RestorePhase::kInitial && !snapshot.schedule_id().IsNil()3 ; |
1206 | 6 | if (!FLAGS_allow_consecutive_restore && restore_sys_catalog0 ) { |
1207 | 0 | SnapshotScheduleState& schedule = |
1208 | 0 | VERIFY_RESULT(FindSnapshotSchedule(snapshot.schedule_id())); |
1209 | | |
1210 | | // Find the latest restore. |
1211 | 0 | HybridTime latest_restore_ht = HybridTime::kMin; |
1212 | 0 | for (const auto& restoration_ht : schedule.options().restoration_times()) { |
1213 | 0 | if (HybridTime::FromPB(restoration_ht) > latest_restore_ht) { |
1214 | 0 | latest_restore_ht = HybridTime::FromPB(restoration_ht); |
1215 | 0 | } |
1216 | 0 | } |
1217 | 0 | LOG(INFO) << "Last successful restoration completed at " |
1218 | 0 | << latest_restore_ht; |
1219 | |
|
1220 | 0 | if (restore_at <= latest_restore_ht) { |
1221 | 0 | LOG(INFO) << "Restore with id " << restoration_id << " not supported " |
1222 | 0 | << "because it is consecutive. Attempting to restore to " |
1223 | 0 | << restore_at << " while last successful restoration completed at " |
1224 | 0 | << latest_restore_ht; |
1225 | 0 | return STATUS_FORMAT(NotSupported, |
1226 | 0 | "Cannot restore before the previous restoration time. " |
1227 | 0 | "A Restoration was performed at $0 and the requested " |
1228 | 0 | "restoration is for $1 which is before the last restoration.", |
1229 | 0 | latest_restore_ht, restore_at); |
1230 | 0 | } |
1231 | 0 | } |
1232 | 6 | RestorationState* restoration_ptr; |
1233 | 6 | if (phase == RestorePhase::kInitial) { |
1234 | 3 | auto restoration = std::make_unique<RestorationState>(&context_, restoration_id, &snapshot); |
1235 | 3 | restoration_ptr = restorations_.emplace(std::move(restoration)).first->get(); |
1236 | 3 | last_restorations_update_ht_ = context_.Clock()->Now(); |
1237 | 3 | } else { |
1238 | 3 | restoration_ptr = &VERIFY_RESULT(FindRestoration(restoration_id)).get(); |
1239 | 3 | } |
1240 | 6 | if (!restore_sys_catalog) { |
1241 | 3 | if (phase == RestorePhase::kPostSysCatalogLoad) { |
1242 | 3 | LOG(INFO) |
1243 | 3 | << "PITR: " << restoration_id << " restore tablets: " << AsString(restore_tablets); |
1244 | | // New tablets could be changed between restoration point and snapshot time. |
1245 | | // So we take tablets list from actual catalog state. |
1246 | 3 | restoration_ptr->InitTabletIds(restore_tablets); |
1247 | 3 | } |
1248 | 3 | tablet_infos = restoration_ptr->PrepareOperations(); |
1249 | 3 | } |
1250 | 6 | auto tablet_ids = snapshot.tablet_ids(); |
1251 | 6 | snapshot_tablets.insert(tablet_ids.begin(), tablet_ids.end()); |
1252 | 6 | } |
1253 | | |
1254 | | // If sys catalog is restored, then tablets data will be restored after that using postponed |
1255 | | // restores. |
1256 | 6 | if (restore_sys_catalog) { |
1257 | 3 | return SubmitRestore(snapshot_id, restore_at, restoration_id, leader_term); |
1258 | 3 | } |
1259 | | |
1260 | 3 | auto snapshot_id_str = snapshot_id.AsSlice().ToBuffer(); |
1261 | 3 | SendMetadata send_metadata(phase == RestorePhase::kPostSysCatalogLoad); |
1262 | 3 | LOG(INFO) << "PITR: " << restoration_id << " restore tablets: " << AsString(tablet_infos); |
1263 | 10 | for (const auto& tablet : tablet_infos) { |
1264 | | // If this tablet did not participate in snapshot, i.e. was deleted. |
1265 | | // We just change hybrid hybrid time limit and clear hide state. |
1266 | 10 | auto task = context_.CreateAsyncTabletSnapshotOp( |
1267 | 10 | tablet, snapshot_tablets.count(tablet->id()) ? snapshot_id_str7 : std::string()3 , |
1268 | 10 | tserver::TabletSnapshotOpRequestPB::RESTORE_ON_TABLET, |
1269 | 10 | MakeDoneCallback(&mutex_, restorations_, restoration_id, tablet->tablet_id(), |
1270 | 10 | std::bind(&Impl::FinishRestoration, this, _1, leader_term))); |
1271 | 10 | task->SetSnapshotHybridTime(restore_at); |
1272 | 10 | task->SetRestorationId(restoration_id); |
1273 | 10 | if (send_metadata) { |
1274 | 10 | task->SetMetadata(tablet->table()->LockForRead()->pb); |
1275 | 10 | } |
1276 | | |
1277 | 10 | context_.ScheduleTabletSnapshotOp(task); |
1278 | 10 | } |
1279 | | |
1280 | | // For empty tablet list, finish the restore. |
1281 | 3 | if (tablet_infos.empty()) { |
1282 | 0 | std::lock_guard<std::mutex> lock(mutex_); |
1283 | 0 | RestorationState* restoration_ptr = &VERIFY_RESULT(FindRestoration(restoration_id)).get(); |
1284 | 0 | if (restoration_ptr) { |
1285 | 0 | FinishRestoration(restoration_ptr, leader_term); |
1286 | 0 | } |
1287 | 0 | } |
1288 | | |
1289 | 3 | return Status::OK(); |
1290 | 3 | } |
1291 | | |
1292 | | // Computes the maximum outstanding Snapshot Create/Delete RPC |
1293 | | // that is permitted. If total limit is specified then it is used otherwise |
1294 | | // the value is computed by multiplying tserver count with the per tserver limit. |
1295 | 300 | uint64_t GetRpcLimit(int64_t total_limit, int64_t per_tserver_limit, int64_t leader_term) { |
1296 | | // NO OP for followers. |
1297 | 300 | if (leader_term < 0) { |
1298 | 30 | return std::numeric_limits<int>::max(); |
1299 | 30 | } |
1300 | | // Should execute only for leaders. |
1301 | 270 | if (total_limit == 0) { |
1302 | 0 | return std::numeric_limits<int>::max(); |
1303 | 0 | } |
1304 | 270 | if (total_limit > 0) { |
1305 | 0 | return total_limit; |
1306 | 0 | } |
1307 | 270 | auto num_result = context_.GetNumLiveTServersForActiveCluster(); |
1308 | | // The cluster config could be empty for e.g. if a restore is in progress |
1309 | | // or e.g. if a new master leader is elected. This is a temporary intermediate |
1310 | | // state (we have bigger problems if the cluster config remains empty forever). |
1311 | | // We use the limit set per tserver as the overall limit in such cases. |
1312 | 270 | if (!num_result.ok()) { |
1313 | 2 | LOG(INFO) << "Cluster Config is not available. Using per tserver limit of " |
1314 | 2 | << per_tserver_limit << " as the throttled limit."; |
1315 | 2 | return per_tserver_limit; |
1316 | 2 | } |
1317 | 268 | uint64_t num_tservers = *num_result; |
1318 | 268 | return num_tservers * per_tserver_limit; |
1319 | 270 | } |
1320 | | |
1321 | | SnapshotCoordinatorContext& context_; |
1322 | | std::mutex mutex_; |
1323 | | class ScheduleTag; |
1324 | | using Snapshots = boost::multi_index_container< |
1325 | | std::unique_ptr<SnapshotState>, |
1326 | | boost::multi_index::indexed_by< |
1327 | | // Access snapshots by id. |
1328 | | boost::multi_index::hashed_unique< |
1329 | | boost::multi_index::const_mem_fun< |
1330 | | SnapshotState, const TxnSnapshotId&, &SnapshotState::id> |
1331 | | >, |
1332 | | // Group snapshots by schedule id. Ordered by hybrid time for the same schedule. |
1333 | | boost::multi_index::ordered_non_unique< |
1334 | | boost::multi_index::tag<ScheduleTag>, |
1335 | | boost::multi_index::composite_key< |
1336 | | SnapshotState, |
1337 | | boost::multi_index::const_mem_fun< |
1338 | | SnapshotState, const SnapshotScheduleId&, &SnapshotState::schedule_id>, |
1339 | | boost::multi_index::const_mem_fun< |
1340 | | SnapshotState, HybridTime, &SnapshotState::snapshot_hybrid_time> |
1341 | | > |
1342 | | > |
1343 | | > |
1344 | | >; |
1345 | | // For restorations and schedules we have to use multi_index since there are template |
1346 | | // functions that expect same interface for those collections. |
1347 | | using Restorations = boost::multi_index_container< |
1348 | | std::unique_ptr<RestorationState>, |
1349 | | boost::multi_index::indexed_by< |
1350 | | boost::multi_index::hashed_unique< |
1351 | | boost::multi_index::const_mem_fun< |
1352 | | RestorationState, const TxnSnapshotRestorationId&, |
1353 | | &RestorationState::restoration_id> |
1354 | | > |
1355 | | > |
1356 | | >; |
1357 | | using Schedules = boost::multi_index_container< |
1358 | | std::unique_ptr<SnapshotScheduleState>, |
1359 | | boost::multi_index::indexed_by< |
1360 | | boost::multi_index::hashed_unique< |
1361 | | boost::multi_index::const_mem_fun< |
1362 | | SnapshotScheduleState, const SnapshotScheduleId&, &SnapshotScheduleState::id> |
1363 | | > |
1364 | | > |
1365 | | >; |
1366 | | |
1367 | | Snapshots snapshots_ GUARDED_BY(mutex_); |
1368 | | Restorations restorations_ GUARDED_BY(mutex_); |
1369 | | HybridTime last_restorations_update_ht_ GUARDED_BY(mutex_); |
1370 | | Schedules schedules_ GUARDED_BY(mutex_); |
1371 | | rpc::Poller poller_; |
1372 | | |
1373 | | // Restores postponed until sys catalog is reloaed. |
1374 | | std::vector<SnapshotScheduleRestorationPtr> postponed_restores_ GUARDED_BY(mutex_); |
1375 | | }; |
1376 | | |
1377 | | MasterSnapshotCoordinator::MasterSnapshotCoordinator(SnapshotCoordinatorContext* context) |
1378 | 8.07k | : impl_(new Impl(context)) {} |
1379 | | |
1380 | 92 | MasterSnapshotCoordinator::~MasterSnapshotCoordinator() {} |
1381 | | |
1382 | | Result<TxnSnapshotId> MasterSnapshotCoordinator::Create( |
1383 | 2 | const SysRowEntries& entries, bool imported, int64_t leader_term, CoarseTimePoint deadline) { |
1384 | 2 | return impl_->Create(entries, imported, leader_term, deadline); |
1385 | 2 | } |
1386 | | |
1387 | | Status MasterSnapshotCoordinator::CreateReplicated( |
1388 | 51 | int64_t leader_term, const tablet::SnapshotOperation& operation) { |
1389 | 51 | return impl_->CreateReplicated(leader_term, operation); |
1390 | 51 | } |
1391 | | |
1392 | | Status MasterSnapshotCoordinator::DeleteReplicated( |
1393 | 0 | int64_t leader_term, const tablet::SnapshotOperation& operation) { |
1394 | 0 | return impl_->DeleteReplicated(leader_term, operation); |
1395 | 0 | } |
1396 | | |
1397 | | Status MasterSnapshotCoordinator::RestoreSysCatalogReplicated( |
1398 | 9 | int64_t leader_term, const tablet::SnapshotOperation& operation, Status* complete_status) { |
1399 | 9 | return impl_->RestoreSysCatalogReplicated(leader_term, operation, complete_status); |
1400 | 9 | } |
1401 | | |
1402 | | Status MasterSnapshotCoordinator::ListSnapshots( |
1403 | 9 | const TxnSnapshotId& snapshot_id, bool list_deleted, ListSnapshotsResponsePB* resp) { |
1404 | 9 | return impl_->ListSnapshots(snapshot_id, list_deleted, resp); |
1405 | 9 | } |
1406 | | |
1407 | | Status MasterSnapshotCoordinator::Delete( |
1408 | 0 | const TxnSnapshotId& snapshot_id, int64_t leader_term, CoarseTimePoint deadline) { |
1409 | 0 | return impl_->Delete(snapshot_id, leader_term, deadline); |
1410 | 0 | } |
1411 | | |
1412 | | Result<TxnSnapshotRestorationId> MasterSnapshotCoordinator::Restore( |
1413 | 3 | const TxnSnapshotId& snapshot_id, HybridTime restore_at, int64_t leader_term) { |
1414 | 3 | return impl_->Restore(snapshot_id, restore_at, leader_term); |
1415 | 3 | } |
1416 | | |
1417 | | Status MasterSnapshotCoordinator::ListRestorations( |
1418 | | const TxnSnapshotRestorationId& restoration_id, const TxnSnapshotId& snapshot_id, |
1419 | 6 | ListSnapshotRestorationsResponsePB* resp) { |
1420 | 6 | return impl_->ListRestorations(restoration_id, snapshot_id, resp); |
1421 | 6 | } |
1422 | | |
1423 | | Result<SnapshotScheduleId> MasterSnapshotCoordinator::CreateSchedule( |
1424 | | const CreateSnapshotScheduleRequestPB& request, int64_t leader_term, |
1425 | 8 | CoarseTimePoint deadline) { |
1426 | 8 | return impl_->CreateSchedule(request, leader_term, deadline); |
1427 | 8 | } |
1428 | | |
1429 | | Status MasterSnapshotCoordinator::ListSnapshotSchedules( |
1430 | 11 | const SnapshotScheduleId& snapshot_schedule_id, ListSnapshotSchedulesResponsePB* resp) { |
1431 | 11 | return impl_->ListSnapshotSchedules(snapshot_schedule_id, resp); |
1432 | 11 | } |
1433 | | |
1434 | | Status MasterSnapshotCoordinator::DeleteSnapshotSchedule( |
1435 | 0 | const SnapshotScheduleId& snapshot_schedule_id, int64_t leader_term, CoarseTimePoint deadline) { |
1436 | 0 | return impl_->DeleteSnapshotSchedule(snapshot_schedule_id, leader_term, deadline); |
1437 | 0 | } |
1438 | | |
1439 | 89 | Status MasterSnapshotCoordinator::Load(tablet::Tablet* tablet) { |
1440 | 89 | return impl_->Load(tablet); |
1441 | 89 | } |
1442 | | |
1443 | 7.94k | void MasterSnapshotCoordinator::Start() { |
1444 | 7.94k | impl_->Start(); |
1445 | 7.94k | } |
1446 | | |
1447 | 94 | void MasterSnapshotCoordinator::Shutdown() { |
1448 | 94 | impl_->Shutdown(); |
1449 | 94 | } |
1450 | | |
1451 | 99.2M | Status MasterSnapshotCoordinator::ApplyWritePair(const Slice& key, const Slice& value) { |
1452 | 99.2M | return impl_->ApplyWritePair(key, value); |
1453 | 99.2M | } |
1454 | | |
1455 | 4.81M | Status MasterSnapshotCoordinator::FillHeartbeatResponse(TSHeartbeatResponsePB* resp) { |
1456 | 4.81M | return impl_->FillHeartbeatResponse(resp); |
1457 | 4.81M | } |
1458 | | |
1459 | | Result<SnapshotSchedulesToObjectIdsMap> |
1460 | 15.4k | MasterSnapshotCoordinator::MakeSnapshotSchedulesToObjectIdsMap(SysRowEntryType type) { |
1461 | 15.4k | return impl_->MakeSnapshotSchedulesToObjectIdsMap(type); |
1462 | 15.4k | } |
1463 | | |
1464 | | Result<bool> MasterSnapshotCoordinator::IsTableCoveredBySomeSnapshotSchedule( |
1465 | 0 | const TableInfo& table_info) { |
1466 | 0 | return impl_->IsTableCoveredBySomeSnapshotSchedule(table_info); |
1467 | 0 | } |
1468 | | |
1469 | 3.00k | void MasterSnapshotCoordinator::SysCatalogLoaded(int64_t term) { |
1470 | 3.00k | impl_->SysCatalogLoaded(term); |
1471 | 3.00k | } |
1472 | | |
1473 | | Result<TxnSnapshotId> MasterSnapshotCoordinator::CreateForSchedule( |
1474 | 1 | const SnapshotScheduleId& schedule_id, int64_t leader_term, CoarseTimePoint deadline) { |
1475 | 1 | return impl_->CreateForSchedule(schedule_id, leader_term, deadline); |
1476 | 1 | } |
1477 | | |
1478 | | } // namespace master |
1479 | | } // namespace yb |