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