1 /*
2 * Copyright (C) 2019 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include "perfetto/base/build_config.h"
18
19 #if PERFETTO_BUILDFLAG(PERFETTO_TP_HTTPD)
20
21 #include "src/trace_processor/rpc/httpd.h"
22
23 #include "perfetto/ext/base/http/http_server.h"
24 #include "perfetto/ext/base/string_utils.h"
25 #include "perfetto/ext/base/string_view.h"
26 #include "perfetto/ext/base/unix_task_runner.h"
27 #include "perfetto/ext/base/utils.h"
28 #include "perfetto/protozero/scattered_heap_buffer.h"
29 #include "perfetto/trace_processor/trace_processor.h"
30 #include "src/trace_processor/rpc/rpc.h"
31
32 #include "protos/perfetto/trace_processor/trace_processor.pbzero.h"
33
34 namespace perfetto {
35 namespace trace_processor {
36
37 namespace {
38
39 constexpr int kBindPort = 9001;
40
41 // Sets the Access-Control-Allow-Origin: $origin on the following origins.
42 // This affects only browser clients that use CORS. Other HTTP clients (e.g. the
43 // python API) don't look at CORS headers.
44 const char* kAllowedCORSOrigins[] = {
45 "https://ui.perfetto.dev",
46 "http://localhost:10000",
47 "http://127.0.0.1:10000",
48 };
49
50 class Httpd : public base::HttpRequestHandler {
51 public:
52 explicit Httpd(std::unique_ptr<TraceProcessor>);
53 ~Httpd() override;
54 void Run(int port);
55
56 private:
57 // HttpRequestHandler implementation.
58 void OnHttpRequest(const base::HttpRequest&) override;
59 void OnWebsocketMessage(const base::WebsocketMessage&) override;
60
61 void ServeHelpPage(const base::HttpRequest&);
62
63 Rpc trace_processor_rpc_;
64 base::UnixTaskRunner task_runner_;
65 base::HttpServer http_srv_;
66 };
67
68 base::HttpServerConnection* g_cur_conn;
69
Vec2Sv(const std::vector<uint8_t> & v)70 base::StringView Vec2Sv(const std::vector<uint8_t>& v) {
71 return base::StringView(reinterpret_cast<const char*>(v.data()), v.size());
72 }
73
74 // Used both by websockets and /rpc chunked HTTP endpoints.
SendRpcChunk(const void * data,uint32_t len)75 void SendRpcChunk(const void* data, uint32_t len) {
76 if (data == nullptr) {
77 // Unrecoverable RPC error case.
78 if (!g_cur_conn->is_websocket())
79 g_cur_conn->SendResponseBody("0\r\n\r\n", 5);
80 g_cur_conn->Close();
81 return;
82 }
83 if (g_cur_conn->is_websocket()) {
84 g_cur_conn->SendWebsocketMessage(data, len);
85 } else {
86 base::StackString<32> chunk_hdr("%x\r\n", len);
87 g_cur_conn->SendResponseBody(chunk_hdr.c_str(), chunk_hdr.len());
88 g_cur_conn->SendResponseBody(data, len);
89 g_cur_conn->SendResponseBody("\r\n", 2);
90 }
91 }
92
Httpd(std::unique_ptr<TraceProcessor> preloaded_instance)93 Httpd::Httpd(std::unique_ptr<TraceProcessor> preloaded_instance)
94 : trace_processor_rpc_(std::move(preloaded_instance)),
95 http_srv_(&task_runner_, this) {}
96 Httpd::~Httpd() = default;
97
Run(int port)98 void Httpd::Run(int port) {
99 PERFETTO_ILOG("[HTTP] Starting RPC server on localhost:%d", port);
100 PERFETTO_LOG(
101 "[HTTP] This server can be used by reloading https://ui.perfetto.dev and "
102 "clicking on YES on the \"Trace Processor native acceleration\" dialog "
103 "or through the Python API (see "
104 "https://perfetto.dev/docs/analysis/trace-processor#python-api).");
105
106 for (size_t i = 0; i < base::ArraySize(kAllowedCORSOrigins); ++i)
107 http_srv_.AddAllowedOrigin(kAllowedCORSOrigins[i]);
108 http_srv_.Start(port);
109 task_runner_.Run();
110 }
111
OnHttpRequest(const base::HttpRequest & req)112 void Httpd::OnHttpRequest(const base::HttpRequest& req) {
113 base::HttpServerConnection& conn = *req.conn;
114 if (req.uri == "/") {
115 // If a user tries to open http://127.0.0.1:9001/ show a minimal help page.
116 return ServeHelpPage(req);
117 }
118
119 static int last_req_id = 0;
120 auto seq_hdr = req.GetHeader("x-seq-id").value_or(base::StringView());
121 int seq_id = base::StringToInt32(seq_hdr.ToStdString()).value_or(0);
122
123 if (seq_id) {
124 if (last_req_id && seq_id != last_req_id + 1 && seq_id != 1)
125 PERFETTO_ELOG("HTTP Request out of order");
126 last_req_id = seq_id;
127 }
128
129 // This is the default. Overridden by the /query handler for chunked replies.
130 char transfer_encoding_hdr[255] = "Transfer-Encoding: identity";
131 std::initializer_list<const char*> headers = {
132 "Cache-Control: no-cache", //
133 "Content-Type: application/x-protobuf", //
134 transfer_encoding_hdr, //
135 };
136
137 if (req.uri == "/status") {
138 auto status = trace_processor_rpc_.GetStatus();
139 return conn.SendResponse("200 OK", headers, Vec2Sv(status));
140 }
141
142 if (req.uri == "/websocket" && req.is_websocket_handshake) {
143 // Will trigger OnWebsocketMessage() when is received.
144 // It returns a 403 if the origin is not in kAllowedCORSOrigins.
145 return conn.UpgradeToWebsocket(req);
146 }
147
148 // --- Everything below this line is a legacy endpoint not used by the UI.
149 // There are two generations of pre-websocket legacy-ness:
150 // 1. The /rpc based endpoint. This is based on a chunked transfer, doing one
151 // POST request for each RPC invocation. All RPC methods are multiplexed
152 // into this one. This is still used by the python API.
153 // 2. The REST API, with one enpoint per RPC method (/parse, /query, ...).
154 // This is unused and will be removed at some point.
155
156 if (req.uri == "/rpc") {
157 // Start the chunked reply.
158 base::StringCopy(transfer_encoding_hdr, "Transfer-Encoding: chunked",
159 sizeof(transfer_encoding_hdr));
160 conn.SendResponseHeaders("200 OK", headers,
161 base::HttpServerConnection::kOmitContentLength);
162 PERFETTO_CHECK(g_cur_conn == nullptr);
163 g_cur_conn = req.conn;
164 trace_processor_rpc_.SetRpcResponseFunction(SendRpcChunk);
165 // OnRpcRequest() will call SendRpcChunk() one or more times.
166 trace_processor_rpc_.OnRpcRequest(req.body.data(), req.body.size());
167 trace_processor_rpc_.SetRpcResponseFunction(nullptr);
168 g_cur_conn = nullptr;
169
170 // Terminate chunked stream.
171 conn.SendResponseBody("0\r\n\r\n", 5);
172 return;
173 }
174
175 if (req.uri == "/parse") {
176 base::Status status = trace_processor_rpc_.Parse(
177 reinterpret_cast<const uint8_t*>(req.body.data()), req.body.size());
178 protozero::HeapBuffered<protos::pbzero::AppendTraceDataResult> result;
179 if (!status.ok()) {
180 result->set_error(status.c_message());
181 }
182 return conn.SendResponse("200 OK", headers,
183 Vec2Sv(result.SerializeAsArray()));
184 }
185
186 if (req.uri == "/notify_eof") {
187 trace_processor_rpc_.NotifyEndOfFile();
188 return conn.SendResponse("200 OK", headers);
189 }
190
191 if (req.uri == "/restore_initial_tables") {
192 trace_processor_rpc_.RestoreInitialTables();
193 return conn.SendResponse("200 OK", headers);
194 }
195
196 // New endpoint, returns data in batches using chunked transfer encoding.
197 // The batch size is determined by |cells_per_batch_| and
198 // |batch_split_threshold_| in query_result_serializer.h.
199 // This is temporary, it will be switched to WebSockets soon.
200 if (req.uri == "/query") {
201 std::vector<uint8_t> response;
202
203 // Start the chunked reply.
204 base::StringCopy(transfer_encoding_hdr, "Transfer-Encoding: chunked",
205 sizeof(transfer_encoding_hdr));
206 conn.SendResponseHeaders("200 OK", headers,
207 base::HttpServerConnection::kOmitContentLength);
208
209 // |on_result_chunk| will be called nested within the same callstack of the
210 // rpc.Query() call. No further calls will be made once Query() returns.
211 auto on_result_chunk = [&](const uint8_t* buf, size_t len, bool has_more) {
212 PERFETTO_DLOG("Sending response chunk, len=%zu eof=%d", len, !has_more);
213 char chunk_hdr[32];
214 auto hdr_len = static_cast<size_t>(sprintf(chunk_hdr, "%zx\r\n", len));
215 conn.SendResponseBody(chunk_hdr, hdr_len);
216 conn.SendResponseBody(buf, len);
217 conn.SendResponseBody("\r\n", 2);
218 if (!has_more) {
219 hdr_len = static_cast<size_t>(sprintf(chunk_hdr, "0\r\n\r\n"));
220 conn.SendResponseBody(chunk_hdr, hdr_len);
221 }
222 };
223 trace_processor_rpc_.Query(
224 reinterpret_cast<const uint8_t*>(req.body.data()), req.body.size(),
225 on_result_chunk);
226 return;
227 }
228
229 // Legacy endpoint.
230 // Returns a columnar-oriented one-shot result. Very inefficient for large
231 // result sets. Very inefficient in general too.
232 if (req.uri == "/raw_query") {
233 std::vector<uint8_t> response = trace_processor_rpc_.RawQuery(
234 reinterpret_cast<const uint8_t*>(req.body.data()), req.body.size());
235 return conn.SendResponse("200 OK", headers, Vec2Sv(response));
236 }
237
238 if (req.uri == "/compute_metric") {
239 std::vector<uint8_t> res = trace_processor_rpc_.ComputeMetric(
240 reinterpret_cast<const uint8_t*>(req.body.data()), req.body.size());
241 return conn.SendResponse("200 OK", headers, Vec2Sv(res));
242 }
243
244 if (req.uri == "/enable_metatrace") {
245 trace_processor_rpc_.EnableMetatrace();
246 return conn.SendResponse("200 OK", headers);
247 }
248
249 if (req.uri == "/disable_and_read_metatrace") {
250 std::vector<uint8_t> res = trace_processor_rpc_.DisableAndReadMetatrace();
251 return conn.SendResponse("200 OK", headers, Vec2Sv(res));
252 }
253
254 return conn.SendResponseAndClose("404 Not Found", headers);
255 }
256
OnWebsocketMessage(const base::WebsocketMessage & msg)257 void Httpd::OnWebsocketMessage(const base::WebsocketMessage& msg) {
258 PERFETTO_CHECK(g_cur_conn == nullptr);
259 g_cur_conn = msg.conn;
260 trace_processor_rpc_.SetRpcResponseFunction(SendRpcChunk);
261 // OnRpcRequest() will call SendRpcChunk() one or more times.
262 trace_processor_rpc_.OnRpcRequest(msg.data.data(), msg.data.size());
263 trace_processor_rpc_.SetRpcResponseFunction(nullptr);
264 g_cur_conn = nullptr;
265 }
266
267 } // namespace
268
RunHttpRPCServer(std::unique_ptr<TraceProcessor> preloaded_instance,std::string port_number)269 void RunHttpRPCServer(std::unique_ptr<TraceProcessor> preloaded_instance,
270 std::string port_number) {
271 Httpd srv(std::move(preloaded_instance));
272 base::Optional<int> port_opt = base::StringToInt32(port_number);
273 int port = port_opt.has_value() ? *port_opt : kBindPort;
274 srv.Run(port);
275 }
276
ServeHelpPage(const base::HttpRequest & req)277 void Httpd::ServeHelpPage(const base::HttpRequest& req) {
278 static const char kPage[] = R"(Perfetto Trace Processor RPC Server
279
280
281 This service can be used in two ways:
282
283 1. Open or reload https://ui.perfetto.dev/
284
285 It will automatically try to connect and use the server on localhost:9001 when
286 available. Click YES when prompted to use Trace Processor Native Acceleration
287 in the UI dialog.
288 See https://perfetto.dev/docs/visualization/large-traces for more.
289
290
291 2. Python API.
292
293 Example: perfetto.TraceProcessor(addr='localhost:9001')
294 See https://perfetto.dev/docs/analysis/trace-processor#python-api for more.
295
296
297 For questions:
298 https://perfetto.dev/docs/contributing/getting-started#community
299 )";
300
301 std::initializer_list<const char*> headers{"Content-Type: text/plain"};
302 req.conn->SendResponse("200 OK", headers, kPage);
303 }
304
305 } // namespace trace_processor
306 } // namespace perfetto
307
308 #endif // PERFETTO_TP_HTTPD
309