/Users/deen/code/yugabyte-db/src/yb/master/snapshot_schedule_state.cc

Source (jump to first uncovered line)
// Copyright (c) YugaByte, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
// in compliance with the License.  You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied.  See the License for the specific language governing permissions and limitations
// under the License.
//

#include "yb/master/snapshot_schedule_state.h"

#include "yb/docdb/docdb.pb.h"
#include "yb/docdb/key_bytes.h"
#include "yb/docdb/value_type.h"

#include "yb/master/catalog_entity_info.h"
#include "yb/master/master_error.h"
#include "yb/master/snapshot_coordinator_context.h"

#include "yb/util/pb_util.h"
#include "yb/util/result.h"
#include "yb/util/status_format.h"

DECLARE_uint64(snapshot_coordinator_cleanup_delay_ms);

namespace yb {
namespace master {

SnapshotScheduleState::SnapshotScheduleState(
    SnapshotCoordinatorContext* context, const CreateSnapshotScheduleRequestPB &req)
    : context_(*context), id_(SnapshotScheduleId::GenerateRandom()), options_(req.options()) {
}

SnapshotScheduleState::SnapshotScheduleState(
    SnapshotCoordinatorContext* context, const SnapshotScheduleId& id,
    const SnapshotScheduleOptionsPB& options)
    : context_(*context), id_(id), options_(options) {
}

Result<docdb::KeyBytes> SnapshotScheduleState::EncodedKey(
    const SnapshotScheduleId& schedule_id, SnapshotCoordinatorContext* context) {
  return master::EncodedKey(SysRowEntryType::SNAPSHOT_SCHEDULE, schedule_id.AsSlice(), context);
}

Result<docdb::KeyBytes> SnapshotScheduleState::EncodedKey() const {
  return EncodedKey(id_, &context_);
}

Status SnapshotScheduleState::StoreToWriteBatch(docdb::KeyValueWriteBatchPB* out) const {
  auto encoded_key = VERIFY_RESULT(EncodedKey());
  auto pair = out->add_write_pairs();
  pair->set_key(encoded_key.AsSlice().cdata(), encoded_key.size());
  auto* value = pair->mutable_value();
  value->push_back(docdb::ValueTypeAsChar::kString);
  pb_util::AppendPartialToString(options_, value);
  return Status::OK();
}

Status SnapshotScheduleState::ToPB(SnapshotScheduleInfoPB* pb) const {
  pb->set_id(id_.data(), id_.size());
  *pb->mutable_options() = options_;
  return Status::OK();
}

std::string SnapshotScheduleState::ToString() const {
  return YB_CLASS_TO_STRING(id, options);
}

bool SnapshotScheduleState::deleted() const {
  return HybridTime::FromPB(options_.delete_time()).is_valid();
}

void SnapshotScheduleState::PrepareOperations(
    HybridTime last_snapshot_time, HybridTime now, SnapshotScheduleOperations* operations) {
  if (creating_snapshot_data_.snapshot_id) {
    return;
  }
  auto delete_time = HybridTime::FromPB(options_.delete_time());
  if (delete_time) {
    // Check whether we are ready to cleanup deleted schedule.
    if (now > delete_time.AddMilliseconds(FLAGS_snapshot_coordinator_cleanup_delay_ms) &&
        !CleanupTracker().Started()) {
      LOG_WITH_PREFIX(INFO) << "Snapshot Schedule " << id() << " cleanup started.";
      if (!CleanupTracker().Start().ok()) {
        LOG(DFATAL) << "Snapshot Schedule " << id() << " cleanup was already started previously.";
      }
      operations->push_back(SnapshotScheduleOperation {
        .type = SnapshotScheduleOperationType::kCleanup,
        .schedule_id = id_,
        .snapshot_id = TxnSnapshotId::Nil(),
      });
    }
    return;
  }
  if (last_snapshot_time && last_snapshot_time.AddSeconds(options_.interval_sec()) > now133) {
    // Time from the last snapshot did not passed yet.
    return;
  }
  operations->push_back(MakeCreateSnapshotOperation(last_snapshot_time));
}

SnapshotScheduleOperation SnapshotScheduleState::MakeCreateSnapshotOperation(
    HybridTime last_snapshot_time) {
  creating_snapshot_data_.snapshot_id = TxnSnapshotId::GenerateRandom();
  creating_snapshot_data_.start_time = CoarseMonoClock::now();
  VLOG_WITH_PREFIX_AND_FUNC0(4) << creating_snapshot_data_.snapshot_id0;
  return SnapshotScheduleOperation {
    .type = SnapshotScheduleOperationType::kCreateSnapshot,
    .schedule_id = id_,
    .snapshot_id = creating_snapshot_data_.snapshot_id,
    .filter = options_.filter(),
    .previous_snapshot_hybrid_time = last_snapshot_time,
  };
}

Result<SnapshotScheduleOperation> SnapshotScheduleState::ForceCreateSnapshot(
    HybridTime last_snapshot_time) {
  if (creating_snapshot_data_.snapshot_id) {
    auto passed = CoarseMonoClock::now() - creating_snapshot_data_.start_time;
    return STATUS_EC_FORMAT(
        IllegalState, MasterError(MasterErrorPB::PARALLEL_SNAPSHOT_OPERATION),
        "Creating snapshot in progress: $0 (passed $1)",
        creating_snapshot_data_.snapshot_id, passed);
  }
  return MakeCreateSnapshotOperation(last_snapshot_time);
}

void SnapshotScheduleState::SnapshotFinished(
    const TxnSnapshotId& snapshot_id, const Status& status) {
  if (creating_snapshot_data_.snapshot_id != snapshot_id) {
    return;
  }
  creating_snapshot_data_.snapshot_id = TxnSnapshotId::Nil();
}

std::string SnapshotScheduleState::LogPrefix() const {
  return Format("$0: ", id_);
}

} // namespace master
} // namespace yb

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/snapshot_schedule_state.h"
15
16		#include "yb/docdb/docdb.pb.h"
17		#include "yb/docdb/key_bytes.h"
18		#include "yb/docdb/value_type.h"
19
20		#include "yb/master/catalog_entity_info.h"
21		#include "yb/master/master_error.h"
22		#include "yb/master/snapshot_coordinator_context.h"
23
24		#include "yb/util/pb_util.h"
25		#include "yb/util/result.h"
26		#include "yb/util/status_format.h"
27
28		DECLARE_uint64(snapshot_coordinator_cleanup_delay_ms);
29
30		namespace yb {
31		namespace master {
32
33		SnapshotScheduleState::SnapshotScheduleState(
34		SnapshotCoordinatorContext* context, const CreateSnapshotScheduleRequestPB &req)
35	8	: context_(*context), id_(SnapshotScheduleId::GenerateRandom()), options_(req.options()) {
36	8	}
37
38		SnapshotScheduleState::SnapshotScheduleState(
39		SnapshotCoordinatorContext* context, const SnapshotScheduleId& id,
40		const SnapshotScheduleOptionsPB& options)
41	33	: context_(*context), id_(id), options_(options) {
42	33	}
43
44		Result<docdb::KeyBytes> SnapshotScheduleState::EncodedKey(
45	11	const SnapshotScheduleId& schedule_id, SnapshotCoordinatorContext* context) {
46	11	return master::EncodedKey(SysRowEntryType::SNAPSHOT_SCHEDULE, schedule_id.AsSlice(), context);
47	11	}
48
49	11	Result<docdb::KeyBytes> SnapshotScheduleState::EncodedKey() const {
50	11	return EncodedKey(id_, &context_);
51	11	}
52
53	11	Status SnapshotScheduleState::StoreToWriteBatch(docdb::KeyValueWriteBatchPB* out) const {
54	11	auto encoded_key = VERIFY_RESULT(EncodedKey());
55	0	auto pair = out->add_write_pairs();
56	11	pair->set_key(encoded_key.AsSlice().cdata(), encoded_key.size());
57	11	auto* value = pair->mutable_value();
58	11	value->push_back(docdb::ValueTypeAsChar::kString);
59	11	pb_util::AppendPartialToString(options_, value);
60	11	return Status::OK();
61	11	}
62
63	11	Status SnapshotScheduleState::ToPB(SnapshotScheduleInfoPB* pb) const {
64	11	pb->set_id(id_.data(), id_.size());
65	11	*pb->mutable_options() = options_;
66	11	return Status::OK();
67	11	}
68
69	9	std::string SnapshotScheduleState::ToString() const {
70	9	return YB_CLASS_TO_STRING(id, options);
71	9	}
72
73	653	bool SnapshotScheduleState::deleted() const {
74	653	return HybridTime::FromPB(options_.delete_time()).is_valid();
75	653	}
76
77		void SnapshotScheduleState::PrepareOperations(
78	173	HybridTime last_snapshot_time, HybridTime now, SnapshotScheduleOperations* operations) {
79	173	if (creating_snapshot_data_.snapshot_id) {
80	32	return;
81	32	}
82	141	auto delete_time = HybridTime::FromPB(options_.delete_time());
83	141	if (delete_time) {
84		// Check whether we are ready to cleanup deleted schedule.
85	0	if (now > delete_time.AddMilliseconds(FLAGS_snapshot_coordinator_cleanup_delay_ms) &&
86	0	!CleanupTracker().Started()) {
87	0	LOG_WITH_PREFIX(INFO) << "Snapshot Schedule " << id() << " cleanup started.";
88	0	if (!CleanupTracker().Start().ok()) {
89	0	LOG(DFATAL) << "Snapshot Schedule " << id() << " cleanup was already started previously.";
90	0	}
91	0	operations->push_back(SnapshotScheduleOperation {
92	0	.type = SnapshotScheduleOperationType::kCleanup,
93	0	.schedule_id = id_,
94	0	.snapshot_id = TxnSnapshotId::Nil(),
95	0	});
96	0	}
97	0	return;
98	0	}
99	141	if (last_snapshot_time && last_snapshot_time.AddSeconds(options_.interval_sec()) > now133 ) {
100		// Time from the last snapshot did not passed yet.
101	123	return;
102	123	}
103	18	operations->push_back(MakeCreateSnapshotOperation(last_snapshot_time));
104	18	}
105
106		SnapshotScheduleOperation SnapshotScheduleState::MakeCreateSnapshotOperation(
107	19	HybridTime last_snapshot_time) {
108	19	creating_snapshot_data_.snapshot_id = TxnSnapshotId::GenerateRandom();
109	19	creating_snapshot_data_.start_time = CoarseMonoClock::now();
110	19	VLOG_WITH_PREFIX_AND_FUNC0 (4) << creating_snapshot_data_.snapshot_id0 ;
111	19	return SnapshotScheduleOperation {
112	19	.type = SnapshotScheduleOperationType::kCreateSnapshot,
113	19	.schedule_id = id_,
114	19	.snapshot_id = creating_snapshot_data_.snapshot_id,
115	19	.filter = options_.filter(),
116	19	.previous_snapshot_hybrid_time = last_snapshot_time,
117	19	};
118	19	}
119
120		Result<SnapshotScheduleOperation> SnapshotScheduleState::ForceCreateSnapshot(
121	1	HybridTime last_snapshot_time) {
122	1	if (creating_snapshot_data_.snapshot_id) {
123	0	auto passed = CoarseMonoClock::now() - creating_snapshot_data_.start_time;
124	0	return STATUS_EC_FORMAT(
125	0	IllegalState, MasterError(MasterErrorPB::PARALLEL_SNAPSHOT_OPERATION),
126	0	"Creating snapshot in progress: $0 (passed $1)",
127	0	creating_snapshot_data_.snapshot_id, passed);
128	0	}
129	1	return MakeCreateSnapshotOperation(last_snapshot_time);
130	1	}
131
132		void SnapshotScheduleState::SnapshotFinished(
133	15	const TxnSnapshotId& snapshot_id, const Status& status) {
134	15	if (creating_snapshot_data_.snapshot_id != snapshot_id) {
135	1	return;
136	1	}
137	14	creating_snapshot_data_.snapshot_id = TxnSnapshotId::Nil();
138	14	}
139
140	0	std::string SnapshotScheduleState::LogPrefix() const {
141	0	return Format("$0: ", id_);
142	0	}
143
144		} // namespace master
145		} // namespace yb

YugabyteDB (2.13.1.0-b60, 21121d69985fbf76aa6958d8f04a9bfa936293b5)

Coverage Report

Created: 2022-03-22 16:43