Coverage Report

Created: 2025-04-29 17:56

/root/doris/be/src/service/http_service.cpp
Line
Count
Source (jump to first uncovered line)
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_service.h"
19
20
#include <event2/bufferevent.h>
21
#include <event2/http.h>
22
#include <gen_cpp/FrontendService_types.h>
23
24
#include <string>
25
#include <vector>
26
27
#include "common/config.h"
28
#include "common/status.h"
29
#include "http/action/adjust_log_level.h"
30
#include "http/action/adjust_tracing_dump.h"
31
#include "http/action/batch_download_action.h"
32
#include "http/action/calc_file_crc_action.h"
33
#include "http/action/check_rpc_channel_action.h"
34
#include "http/action/check_tablet_segment_action.h"
35
#include "http/action/checksum_action.h"
36
#include "http/action/clear_cache_action.h"
37
#include "http/action/compaction_action.h"
38
#include "http/action/compaction_score_action.h"
39
#include "http/action/config_action.h"
40
#include "http/action/debug_point_action.h"
41
#include "http/action/download_action.h"
42
#include "http/action/download_binlog_action.h"
43
#include "http/action/file_cache_action.h"
44
#include "http/action/health_action.h"
45
#include "http/action/http_stream.h"
46
#include "http/action/jeprofile_actions.h"
47
#include "http/action/load_stream_action.h"
48
#include "http/action/meta_action.h"
49
#include "http/action/metrics_action.h"
50
#include "http/action/pad_rowset_action.h"
51
#include "http/action/pipeline_task_action.h"
52
#include "http/action/pprof_actions.h"
53
#include "http/action/reload_tablet_action.h"
54
#include "http/action/report_action.h"
55
#include "http/action/reset_rpc_channel_action.h"
56
#include "http/action/restore_tablet_action.h"
57
#include "http/action/show_nested_index_file_action.h"
58
#include "http/action/snapshot_action.h"
59
#include "http/action/stream_load.h"
60
#include "http/action/stream_load_2pc.h"
61
#include "http/action/tablet_migration_action.h"
62
#include "http/action/tablets_distribution_action.h"
63
#include "http/action/tablets_info_action.h"
64
#include "http/action/version_action.h"
65
#include "http/default_path_handlers.h"
66
#include "http/ev_http_server.h"
67
#include "http/http_method.h"
68
#include "http/web_page_handler.h"
69
#include "olap/options.h"
70
#include "runtime/exec_env.h"
71
#include "runtime/load_path_mgr.h"
72
#include "util/doris_metrics.h"
73
74
namespace doris {
75
namespace {
76
1
std::shared_ptr<bufferevent_rate_limit_group> get_rate_limit_group(event_base* event_base) {
77
1
    auto rate_limit = config::download_binlog_rate_limit_kbs;
78
1
    if (rate_limit <= 0) {
79
1
        return nullptr;
80
1
    }
81
82
0
    auto max_value = std::numeric_limits<int32_t>::max() / 1024 * 10;
83
0
    if (rate_limit > max_value) {
84
0
        LOG(WARNING) << "rate limit is too large, set to max value.";
85
0
        rate_limit = max_value;
86
0
    }
87
0
    struct timeval cfg_tick = {0, 100 * 1000}; // 100ms
88
0
    rate_limit = rate_limit / 10 * 1024;       // convert to KB/S
89
90
0
    auto token_bucket = std::unique_ptr<ev_token_bucket_cfg, decltype(&ev_token_bucket_cfg_free)>(
91
0
            ev_token_bucket_cfg_new(rate_limit, rate_limit * 2, rate_limit, rate_limit * 2,
92
0
                                    &cfg_tick),
93
0
            ev_token_bucket_cfg_free);
94
0
    return std::shared_ptr<bufferevent_rate_limit_group>(
95
0
            bufferevent_rate_limit_group_new(event_base, token_bucket.get()),
96
0
            bufferevent_rate_limit_group_free);
97
1
}
98
} // namespace
99
100
HttpService::HttpService(ExecEnv* env, int port, int num_threads)
101
        : _env(env),
102
          _ev_http_server(new EvHttpServer(port, num_threads)),
103
1
          _web_page_handler(new WebPageHandler(_ev_http_server.get())) {}
104
105
1
HttpService::~HttpService() {
106
1
    stop();
107
1
}
108
109
// NOLINTBEGIN(readability-function-size)
110
1
Status HttpService::start() {
111
1
    add_default_path_handlers(_web_page_handler.get());
112
113
1
    auto event_base = _ev_http_server->get_event_bases()[0];
114
1
    _rate_limit_group = get_rate_limit_group(event_base.get());
115
116
    // register load
117
1
    StreamLoadAction* streamload_action = _pool.add(new StreamLoadAction(_env));
118
1
    _ev_http_server->register_handler(HttpMethod::PUT, "/api/{db}/{table}/_load",
119
1
                                      streamload_action);
120
1
    _ev_http_server->register_handler(HttpMethod::PUT, "/api/{db}/{table}/_stream_load",
121
1
                                      streamload_action);
122
1
    StreamLoad2PCAction* streamload_2pc_action = _pool.add(new StreamLoad2PCAction(_env));
123
1
    _ev_http_server->register_handler(HttpMethod::PUT, "/api/{db}/_stream_load_2pc",
124
1
                                      streamload_2pc_action);
125
1
    _ev_http_server->register_handler(HttpMethod::PUT, "/api/{db}/{table}/_stream_load_2pc",
126
1
                                      streamload_2pc_action);
127
128
    // register http_stream
129
1
    HttpStreamAction* http_stream_action = _pool.add(new HttpStreamAction(_env));
130
1
    _ev_http_server->register_handler(HttpMethod::PUT, "/api/_http_stream", http_stream_action);
131
132
    // register download action
133
1
    std::vector<std::string> allow_paths;
134
1
    for (auto& path : _env->store_paths()) {
135
0
        allow_paths.emplace_back(path.path);
136
0
    }
137
1
    DownloadAction* download_action = _pool.add(new DownloadAction(_env, nullptr, allow_paths));
138
1
    _ev_http_server->register_handler(HttpMethod::HEAD, "/api/_download_load", download_action);
139
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/_download_load", download_action);
140
141
1
    DownloadAction* tablet_download_action =
142
1
            _pool.add(new DownloadAction(_env, _rate_limit_group, allow_paths));
143
1
    _ev_http_server->register_handler(HttpMethod::HEAD, "/api/_tablet/_download",
144
1
                                      tablet_download_action);
145
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/_tablet/_download",
146
1
                                      tablet_download_action);
147
1
    if (config::enable_single_replica_load) {
148
1
        DownloadAction* single_replica_download_action = _pool.add(new DownloadAction(
149
1
                _env, nullptr, allow_paths, config::single_replica_load_download_num_workers));
150
1
        _ev_http_server->register_handler(HttpMethod::HEAD, "/api/_single_replica/_download",
151
1
                                          single_replica_download_action);
152
1
        _ev_http_server->register_handler(HttpMethod::GET, "/api/_single_replica/_download",
153
1
                                          single_replica_download_action);
154
1
    }
155
156
1
    DownloadAction* error_log_download_action =
157
1
            _pool.add(new DownloadAction(_env, _env->load_path_mgr()->get_load_error_file_dir()));
158
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/_load_error_log",
159
1
                                      error_log_download_action);
160
1
    _ev_http_server->register_handler(HttpMethod::HEAD, "/api/_load_error_log",
161
1
                                      error_log_download_action);
162
163
1
    DownloadBinlogAction* download_binlog_action =
164
1
            _pool.add(new DownloadBinlogAction(_env, _rate_limit_group));
165
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/_binlog/_download",
166
1
                                      download_binlog_action);
167
1
    _ev_http_server->register_handler(HttpMethod::HEAD, "/api/_binlog/_download",
168
1
                                      download_binlog_action);
169
170
1
    BatchDownloadAction* batch_download_action =
171
1
            _pool.add(new BatchDownloadAction(_env, _rate_limit_group, allow_paths));
172
1
    _ev_http_server->register_handler(HttpMethod::HEAD, "/api/_tablet/_batch_download",
173
1
                                      batch_download_action);
174
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/_tablet/_batch_download",
175
1
                                      batch_download_action);
176
1
    _ev_http_server->register_handler(HttpMethod::POST, "/api/_tablet/_batch_download",
177
1
                                      batch_download_action);
178
179
1
    AdjustLogLevelAction* adjust_log_level_action = _pool.add(new AdjustLogLevelAction());
180
1
    _ev_http_server->register_handler(HttpMethod::POST, "api/glog/adjust", adjust_log_level_action);
181
182
    //TODO: add query GET interface
183
1
    auto* adjust_tracing_dump = _pool.add(new AdjustTracingDump());
184
1
    _ev_http_server->register_handler(HttpMethod::POST, "api/pipeline/tracing",
185
1
                                      adjust_tracing_dump);
186
187
    // Register BE version action
188
1
    VersionAction* version_action =
189
1
            _pool.add(new VersionAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::NONE));
190
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/be_version_info", version_action);
191
192
    // Register BE health action
193
1
    HealthAction* health_action = _pool.add(new HealthAction());
194
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/health", health_action);
195
196
    // Clear cache action
197
1
    ClearCacheAction* clear_cache_action = _pool.add(new ClearCacheAction());
198
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/clear_cache/{type}",
199
1
                                      clear_cache_action);
200
201
    // Dump all running pipeline tasks
202
1
    PipelineTaskAction* pipeline_task_action = _pool.add(new PipelineTaskAction());
203
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/running_pipeline_tasks",
204
1
                                      pipeline_task_action);
205
206
    // Dump all running pipeline tasks which has been running for more than {duration} seconds
207
1
    LongPipelineTaskAction* long_pipeline_task_action = _pool.add(new LongPipelineTaskAction());
208
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/running_pipeline_tasks/{duration}",
209
1
                                      long_pipeline_task_action);
210
211
    // Register BE LoadStream action
212
1
    LoadStreamAction* load_stream_action = _pool.add(new LoadStreamAction());
213
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/load_streams", load_stream_action);
214
215
1
    QueryPipelineTaskAction* query_pipeline_task_action = _pool.add(new QueryPipelineTaskAction());
216
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/query_pipeline_tasks/{query_id}",
217
1
                                      query_pipeline_task_action);
218
219
    // Register Tablets Info action
220
1
    TabletsInfoAction* tablets_info_action =
221
1
            _pool.add(new TabletsInfoAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
222
1
    _ev_http_server->register_handler(HttpMethod::GET, "/tablets_json", tablets_info_action);
223
224
    // Register Tablets Distribution action
225
1
    TabletsDistributionAction* tablets_distribution_action = _pool.add(
226
1
            new TabletsDistributionAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
227
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/tablets_distribution",
228
1
                                      tablets_distribution_action);
229
230
    // Register tablet migration action
231
1
    TabletMigrationAction* tablet_migration_action = _pool.add(
232
1
            new TabletMigrationAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
233
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/tablet_migration",
234
1
                                      tablet_migration_action);
235
236
    // register pprof actions
237
1
    static_cast<void>(PprofActions::setup(_env, _ev_http_server.get(), _pool));
238
239
    // register jeprof actions
240
1
    SetJeHeapProfileActiveActions* set_jeheap_profile_active_action =
241
1
            _pool.add(new SetJeHeapProfileActiveActions(_env));
242
1
    _ev_http_server->register_handler(HttpMethod::GET, "/jeheap/active/{prof_value}",
243
1
                                      set_jeheap_profile_active_action);
244
245
1
    DumpJeHeapProfileToDotActions* dump_jeheap_profile_to_dot_action =
246
1
            _pool.add(new DumpJeHeapProfileToDotActions(_env));
247
1
    _ev_http_server->register_handler(HttpMethod::GET, "/jeheap/dump",
248
1
                                      dump_jeheap_profile_to_dot_action);
249
250
1
    DumpJeHeapProfileActions* dump_jeheap_profile_action =
251
1
            _pool.add(new DumpJeHeapProfileActions(_env));
252
1
    _ev_http_server->register_handler(HttpMethod::GET, "/jeheap/dump_only",
253
1
                                      dump_jeheap_profile_action);
254
255
    // register metrics
256
1
    {
257
1
        auto* action =
258
1
                _pool.add(new MetricsAction(DorisMetrics::instance()->metric_registry(), _env,
259
1
                                            TPrivilegeHier::GLOBAL, TPrivilegeType::NONE));
260
1
        _ev_http_server->register_handler(HttpMethod::GET, "/metrics", action);
261
1
    }
262
263
1
    MetaAction* meta_action =
264
1
            _pool.add(new MetaAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
265
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/meta/{op}/{tablet_id}", meta_action);
266
267
1
    FileCacheAction* file_cache_action = _pool.add(new FileCacheAction());
268
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/file_cache", file_cache_action);
269
270
#ifndef BE_TEST
271
    // Register BE checksum action
272
    ChecksumAction* checksum_action =
273
            _pool.add(new ChecksumAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
274
    _ev_http_server->register_handler(HttpMethod::GET, "/api/checksum", checksum_action);
275
276
    // Register BE reload tablet action
277
    ReloadTabletAction* reload_tablet_action =
278
            _pool.add(new ReloadTabletAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
279
    _ev_http_server->register_handler(HttpMethod::GET, "/api/reload_tablet", reload_tablet_action);
280
281
    RestoreTabletAction* restore_tablet_action =
282
            _pool.add(new RestoreTabletAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
283
    _ev_http_server->register_handler(HttpMethod::POST, "/api/restore_tablet",
284
                                      restore_tablet_action);
285
286
    // Register BE snapshot action
287
    SnapshotAction* snapshot_action =
288
            _pool.add(new SnapshotAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
289
    _ev_http_server->register_handler(HttpMethod::GET, "/api/snapshot", snapshot_action);
290
291
    CompactionScoreAction* compaction_score_action =
292
            _pool.add(new CompactionScoreAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN,
293
                                                _env->get_storage_engine()->tablet_manager()));
294
    _ev_http_server->register_handler(HttpMethod::GET, "/api/compaction_score",
295
                                      compaction_score_action);
296
#endif
297
298
    // 2 compaction actions
299
1
    CompactionAction* show_compaction_action = _pool.add(new CompactionAction(
300
1
            CompactionActionType::SHOW_INFO, _env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
301
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/compaction/show",
302
1
                                      show_compaction_action);
303
1
    CompactionAction* run_compaction_action =
304
1
            _pool.add(new CompactionAction(CompactionActionType::RUN_COMPACTION, _env,
305
1
                                           TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
306
1
    _ev_http_server->register_handler(HttpMethod::POST, "/api/compaction/run",
307
1
                                      run_compaction_action);
308
1
    CompactionAction* run_status_compaction_action =
309
1
            _pool.add(new CompactionAction(CompactionActionType::RUN_COMPACTION_STATUS, _env,
310
1
                                           TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
311
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/compaction/run_status",
312
1
                                      run_status_compaction_action);
313
314
1
    ConfigAction* update_config_action =
315
1
            _pool.add(new ConfigAction(ConfigActionType::UPDATE_CONFIG));
316
1
    _ev_http_server->register_handler(HttpMethod::POST, "/api/update_config", update_config_action);
317
318
1
    ConfigAction* show_config_action = _pool.add(new ConfigAction(ConfigActionType::SHOW_CONFIG));
319
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/show_config", show_config_action);
320
321
    // 3 check action
322
1
    CheckRPCChannelAction* check_rpc_channel_action = _pool.add(
323
1
            new CheckRPCChannelAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
324
1
    _ev_http_server->register_handler(HttpMethod::GET,
325
1
                                      "/api/check_rpc_channel/{ip}/{port}/{payload_size}",
326
1
                                      check_rpc_channel_action);
327
328
1
    ResetRPCChannelAction* reset_rpc_channel_action = _pool.add(
329
1
            new ResetRPCChannelAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
330
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/reset_rpc_channel/{endpoints}",
331
1
                                      reset_rpc_channel_action);
332
333
1
    CheckTabletSegmentAction* check_tablet_segment_action = _pool.add(
334
1
            new CheckTabletSegmentAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
335
1
    _ev_http_server->register_handler(HttpMethod::POST, "/api/check_tablet_segment_lost",
336
1
                                      check_tablet_segment_action);
337
338
1
    PadRowsetAction* pad_rowset_action =
339
1
            _pool.add(new PadRowsetAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
340
1
    _ev_http_server->register_handler(HttpMethod::POST, "/api/pad_rowset", pad_rowset_action);
341
342
    // debug point
343
1
    AddDebugPointAction* add_debug_point_action =
344
1
            _pool.add(new AddDebugPointAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
345
1
    _ev_http_server->register_handler(HttpMethod::POST, "/api/debug_point/add/{debug_point}",
346
1
                                      add_debug_point_action);
347
348
1
    RemoveDebugPointAction* remove_debug_point_action = _pool.add(
349
1
            new RemoveDebugPointAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
350
1
    _ev_http_server->register_handler(HttpMethod::POST, "/api/debug_point/remove/{debug_point}",
351
1
                                      remove_debug_point_action);
352
353
1
    ClearDebugPointsAction* clear_debug_points_action = _pool.add(
354
1
            new ClearDebugPointsAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
355
1
    _ev_http_server->register_handler(HttpMethod::POST, "/api/debug_point/clear",
356
1
                                      clear_debug_points_action);
357
358
1
    ReportAction* report_tablet_action = _pool.add(new ReportAction(
359
1
            _env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN, "REPORT_OLAP_TABLE"));
360
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/report/tablet", report_tablet_action);
361
362
1
    ReportAction* report_disk_action = _pool.add(new ReportAction(
363
1
            _env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN, "REPORT_DISK_STATE"));
364
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/report/disk", report_disk_action);
365
366
1
    CalcFileCrcAction* calc_crc_action =
367
1
            _pool.add(new CalcFileCrcAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
368
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/calc_crc", calc_crc_action);
369
370
1
    ShowNestedIndexFileAction* show_nested_index_file_action = _pool.add(
371
1
            new ShowNestedIndexFileAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
372
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/show_nested_index_file",
373
1
                                      show_nested_index_file_action);
374
375
1
    ReportAction* report_task_action = _pool.add(
376
1
            new ReportAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN, "REPORT_TASK"));
377
1
    _ev_http_server->register_handler(HttpMethod::GET, "/api/report/task", report_task_action);
378
379
1
    _ev_http_server->start();
380
1
    return Status::OK();
381
1
}
382
// NOLINTEND(readability-function-size)
383
384
1
void HttpService::stop() {
385
1
    if (stopped) {
386
0
        return;
387
0
    }
388
1
    _ev_http_server->stop();
389
1
    _pool.clear();
390
1
    stopped = true;
391
1
}
392
393
1
int HttpService::get_real_port() const {
394
1
    return _ev_http_server->get_real_port();
395
1
}
396
397
} // namespace doris