be/src/service/http/action/peer_cache_action.cpp
Line | Count | Source |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #include "service/http/action/peer_cache_action.h" |
19 | | |
20 | | #include <rapidjson/document.h> |
21 | | #include <rapidjson/prettywriter.h> |
22 | | #include <rapidjson/stringbuffer.h> |
23 | | |
24 | | #include <string> |
25 | | |
26 | | #include "cloud/cloud_storage_engine.h" |
27 | | #include "cloud/cloud_warm_up_manager.h" |
28 | | #include "service/http/http_channel.h" |
29 | | #include "service/http/http_headers.h" |
30 | | #include "service/http/http_request.h" |
31 | | #include "service/http/http_status.h" |
32 | | |
33 | | namespace doris { |
34 | | |
35 | | static constexpr std::string_view HEADER_JSON = "application/json"; |
36 | | |
37 | | // Serialize a single TabletPeerCandidates into a rapidjson object |
38 | | static void tablet_peer_to_json(int64_t tablet_id, const TabletPeerCandidates& tpc, |
39 | 0 | rapidjson::Writer<rapidjson::StringBuffer>& writer) { |
40 | 0 | writer.StartObject(); |
41 | 0 | writer.Key("tablet_id"); |
42 | 0 | writer.Int64(tablet_id); |
43 | |
|
44 | 0 | writer.Key("candidates"); |
45 | 0 | writer.StartArray(); |
46 | 0 | for (const auto& c : tpc.candidates) { |
47 | 0 | writer.StartObject(); |
48 | 0 | writer.Key("host"); |
49 | 0 | writer.String(c.host.c_str()); |
50 | 0 | writer.Key("brpc_port"); |
51 | 0 | writer.Int(c.brpc_port); |
52 | 0 | writer.Key("compute_group_id"); |
53 | 0 | writer.String(c.compute_group_id.c_str()); |
54 | 0 | writer.Key("last_access_time_ms"); |
55 | 0 | writer.Int64(c.last_access_time_ms); |
56 | 0 | writer.Key("consecutive_rpc_failures"); |
57 | 0 | writer.Int(c.consecutive_rpc_failures); |
58 | 0 | writer.EndObject(); |
59 | 0 | } |
60 | 0 | writer.EndArray(); |
61 | |
|
62 | 0 | writer.Key("last_successful_compute_group_id"); |
63 | 0 | writer.String(tpc.last_successful_compute_group_id.c_str()); |
64 | 0 | writer.Key("fetching_from_fe"); |
65 | 0 | writer.Bool(tpc.fetching_from_fe); |
66 | 0 | writer.Key("consecutive_all_miss"); |
67 | 0 | writer.Int(tpc.consecutive_all_miss); |
68 | 0 | writer.Key("cooldown_until_ms"); |
69 | 0 | writer.Int64(tpc.cooldown_until_ms); |
70 | |
|
71 | 0 | writer.EndObject(); |
72 | 0 | } |
73 | | |
74 | 0 | static int64_t parse_tablet_id(HttpRequest* req) { |
75 | 0 | const std::string& tablet_id_str = req->param("tablet_id"); |
76 | 0 | if (tablet_id_str.empty()) { |
77 | 0 | return -1; |
78 | 0 | } |
79 | 0 | try { |
80 | 0 | return std::stoll(tablet_id_str); |
81 | 0 | } catch (...) { |
82 | 0 | return -1; |
83 | 0 | } |
84 | 0 | } |
85 | | |
86 | 0 | Status PeerCacheAction::_handle_show(HttpRequest* req, std::string* result) { |
87 | 0 | int64_t tablet_id = parse_tablet_id(req); |
88 | 0 | if (tablet_id < 0) { |
89 | 0 | return Status::InvalidArgument("missing or invalid parameter: tablet_id"); |
90 | 0 | } |
91 | | |
92 | 0 | auto& mgr = _engine.cloud_warm_up_manager(); |
93 | 0 | auto opt = mgr.get_tablet_peer_info(tablet_id); |
94 | 0 | if (!opt.has_value()) { |
95 | 0 | return Status::NotFound("tablet_id {} has no peer candidates", tablet_id); |
96 | 0 | } |
97 | | |
98 | 0 | rapidjson::StringBuffer buf; |
99 | 0 | rapidjson::Writer<rapidjson::StringBuffer> writer(buf); |
100 | 0 | tablet_peer_to_json(tablet_id, opt.value(), writer); |
101 | 0 | *result = buf.GetString(); |
102 | 0 | return Status::OK(); |
103 | 0 | } |
104 | | |
105 | 0 | Status PeerCacheAction::_handle_show_all(HttpRequest* req, std::string* result) { |
106 | 0 | int64_t limit = 1000; // default |
107 | 0 | const std::string& limit_str = req->param("limit"); |
108 | 0 | if (!limit_str.empty()) { |
109 | 0 | try { |
110 | 0 | limit = std::stoll(limit_str); |
111 | 0 | } catch (...) { |
112 | 0 | return Status::InvalidArgument("invalid parameter: limit"); |
113 | 0 | } |
114 | 0 | } |
115 | | |
116 | 0 | auto& mgr = _engine.cloud_warm_up_manager(); |
117 | 0 | auto all = mgr.get_all_peer_info(limit); |
118 | |
|
119 | 0 | rapidjson::StringBuffer buf; |
120 | 0 | rapidjson::Writer<rapidjson::StringBuffer> writer(buf); |
121 | 0 | writer.StartObject(); |
122 | 0 | writer.Key("total"); |
123 | 0 | writer.Int64(static_cast<int64_t>(all.size())); |
124 | 0 | writer.Key("tablets"); |
125 | 0 | writer.StartArray(); |
126 | 0 | for (const auto& [tid, tpc] : all) { |
127 | 0 | tablet_peer_to_json(tid, tpc, writer); |
128 | 0 | } |
129 | 0 | writer.EndArray(); |
130 | 0 | writer.EndObject(); |
131 | 0 | *result = buf.GetString(); |
132 | 0 | return Status::OK(); |
133 | 0 | } |
134 | | |
135 | 0 | Status PeerCacheAction::_handle_set(HttpRequest* req, std::string* result) { |
136 | 0 | int64_t tablet_id = parse_tablet_id(req); |
137 | 0 | if (tablet_id < 0) { |
138 | 0 | return Status::InvalidArgument("missing or invalid parameter: tablet_id"); |
139 | 0 | } |
140 | | |
141 | 0 | const std::string& body = req->get_request_body(); |
142 | 0 | if (body.empty()) { |
143 | 0 | return Status::InvalidArgument("missing request body (JSON expected)"); |
144 | 0 | } |
145 | | |
146 | 0 | rapidjson::Document doc; |
147 | 0 | doc.Parse(body.c_str(), body.size()); |
148 | 0 | if (doc.HasParseError()) { |
149 | 0 | return Status::InvalidArgument("invalid JSON body: parse error at offset {}", |
150 | 0 | doc.GetErrorOffset()); |
151 | 0 | } |
152 | | |
153 | 0 | TabletPeerCandidates tpc; |
154 | | |
155 | | // Parse candidates array |
156 | 0 | if (doc.HasMember("candidates") && doc["candidates"].IsArray()) { |
157 | 0 | for (const auto& item : doc["candidates"].GetArray()) { |
158 | 0 | PeerCandidate c; |
159 | 0 | if (item.HasMember("host") && item["host"].IsString()) { |
160 | 0 | c.host = item["host"].GetString(); |
161 | 0 | } |
162 | 0 | if (item.HasMember("brpc_port") && item["brpc_port"].IsInt()) { |
163 | 0 | c.brpc_port = item["brpc_port"].GetInt(); |
164 | 0 | } |
165 | 0 | if (item.HasMember("compute_group_id") && item["compute_group_id"].IsString()) { |
166 | 0 | c.compute_group_id = item["compute_group_id"].GetString(); |
167 | 0 | } |
168 | 0 | if (item.HasMember("last_access_time_ms") && item["last_access_time_ms"].IsInt64()) { |
169 | 0 | c.last_access_time_ms = item["last_access_time_ms"].GetInt64(); |
170 | 0 | } |
171 | 0 | if (item.HasMember("consecutive_rpc_failures") && |
172 | 0 | item["consecutive_rpc_failures"].IsInt()) { |
173 | 0 | c.consecutive_rpc_failures = item["consecutive_rpc_failures"].GetInt(); |
174 | 0 | } |
175 | | // If last_access_time_ms not provided, set to current time |
176 | 0 | if (c.last_access_time_ms == 0) { |
177 | 0 | c.last_access_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>( |
178 | 0 | std::chrono::system_clock::now().time_since_epoch()) |
179 | 0 | .count(); |
180 | 0 | } |
181 | 0 | tpc.candidates.push_back(std::move(c)); |
182 | 0 | } |
183 | 0 | } |
184 | |
|
185 | 0 | if (doc.HasMember("last_successful_compute_group_id") && |
186 | 0 | doc["last_successful_compute_group_id"].IsString()) { |
187 | 0 | tpc.last_successful_compute_group_id = doc["last_successful_compute_group_id"].GetString(); |
188 | 0 | } |
189 | 0 | if (doc.HasMember("consecutive_all_miss") && doc["consecutive_all_miss"].IsInt()) { |
190 | 0 | tpc.consecutive_all_miss = doc["consecutive_all_miss"].GetInt(); |
191 | 0 | } |
192 | 0 | if (doc.HasMember("cooldown_until_ms") && doc["cooldown_until_ms"].IsInt64()) { |
193 | 0 | tpc.cooldown_until_ms = doc["cooldown_until_ms"].GetInt64(); |
194 | 0 | } |
195 | |
|
196 | 0 | auto& mgr = _engine.cloud_warm_up_manager(); |
197 | 0 | mgr.set_tablet_peer_candidates(tablet_id, std::move(tpc)); |
198 | |
|
199 | 0 | *result = R"({"status":"OK","msg":"peer candidates set successfully"})"; |
200 | 0 | return Status::OK(); |
201 | 0 | } |
202 | | |
203 | 0 | Status PeerCacheAction::_handle_remove(HttpRequest* req, std::string* result) { |
204 | 0 | int64_t tablet_id = parse_tablet_id(req); |
205 | 0 | if (tablet_id < 0) { |
206 | 0 | return Status::InvalidArgument("missing or invalid parameter: tablet_id"); |
207 | 0 | } |
208 | | |
209 | 0 | auto& mgr = _engine.cloud_warm_up_manager(); |
210 | 0 | mgr.remove_balanced_tablet(tablet_id); |
211 | |
|
212 | 0 | *result = R"({"status":"OK","msg":"peer candidates removed"})"; |
213 | 0 | return Status::OK(); |
214 | 0 | } |
215 | | |
216 | 0 | Status PeerCacheAction::_handle_reset_cooldown(HttpRequest* req, std::string* result) { |
217 | 0 | int64_t tablet_id = parse_tablet_id(req); |
218 | 0 | if (tablet_id < 0) { |
219 | 0 | return Status::InvalidArgument("missing or invalid parameter: tablet_id"); |
220 | 0 | } |
221 | | |
222 | 0 | auto& mgr = _engine.cloud_warm_up_manager(); |
223 | 0 | auto opt = mgr.get_tablet_peer_info(tablet_id); |
224 | 0 | if (!opt.has_value()) { |
225 | 0 | return Status::NotFound("tablet_id {} has no peer candidates", tablet_id); |
226 | 0 | } |
227 | | |
228 | | // Reset cooldown fields and write back |
229 | 0 | auto tpc = std::move(opt.value()); |
230 | 0 | tpc.consecutive_all_miss = 0; |
231 | 0 | tpc.cooldown_until_ms = 0; |
232 | 0 | mgr.set_tablet_peer_candidates(tablet_id, std::move(tpc)); |
233 | |
|
234 | 0 | *result = R"({"status":"OK","msg":"cooldown reset"})"; |
235 | 0 | return Status::OK(); |
236 | 0 | } |
237 | | |
238 | 0 | void PeerCacheAction::handle(HttpRequest* req) { |
239 | 0 | req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.data()); |
240 | |
|
241 | 0 | const std::string& op = req->param("op"); |
242 | 0 | std::string result; |
243 | 0 | Status st; |
244 | |
|
245 | 0 | if (op == "show") { |
246 | 0 | st = _handle_show(req, &result); |
247 | 0 | } else if (op == "show_all") { |
248 | 0 | st = _handle_show_all(req, &result); |
249 | 0 | } else if (op == "set") { |
250 | 0 | st = _handle_set(req, &result); |
251 | 0 | } else if (op == "remove") { |
252 | 0 | st = _handle_remove(req, &result); |
253 | 0 | } else if (op == "reset_cooldown") { |
254 | 0 | st = _handle_reset_cooldown(req, &result); |
255 | 0 | } else { |
256 | 0 | st = Status::InvalidArgument( |
257 | 0 | "unknown op '{}', supported: show, show_all, set, remove, reset_cooldown", op); |
258 | 0 | } |
259 | |
|
260 | 0 | if (st.ok()) { |
261 | 0 | HttpChannel::send_reply(req, HttpStatus::OK, result); |
262 | 0 | } else { |
263 | 0 | HttpChannel::send_reply(req, HttpStatus::INTERNAL_SERVER_ERROR, st.to_json()); |
264 | 0 | } |
265 | 0 | } |
266 | | |
267 | | } // namespace doris |