1 //===--- Server.cpp - gRPC-based Remote Index Server  ---------------------===//
2 //
3 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
4 // See https://llvm.org/LICENSE.txt for license information.
5 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
6 //
7 //===----------------------------------------------------------------------===//
8 
9 #include "Feature.h"
10 #include "Index.pb.h"
11 #include "MonitoringService.grpc.pb.h"
12 #include "MonitoringService.pb.h"
13 #include "Service.grpc.pb.h"
14 #include "Service.pb.h"
15 #include "index/Index.h"
16 #include "index/Serialization.h"
17 #include "index/Symbol.h"
18 #include "index/remote/marshalling/Marshalling.h"
19 #include "support/Context.h"
20 #include "support/Logger.h"
21 #include "support/Shutdown.h"
22 #include "support/ThreadsafeFS.h"
23 #include "support/Trace.h"
24 #include "llvm/ADT/IntrusiveRefCntPtr.h"
25 #include "llvm/ADT/None.h"
26 #include "llvm/ADT/StringRef.h"
27 #include "llvm/Support/Chrono.h"
28 #include "llvm/Support/CommandLine.h"
29 #include "llvm/Support/Error.h"
30 #include "llvm/Support/FileSystem.h"
31 #include "llvm/Support/FormatVariadic.h"
32 #include "llvm/Support/Path.h"
33 #include "llvm/Support/Signals.h"
34 #include "llvm/Support/VirtualFileSystem.h"
35 
36 #include <chrono>
37 #include <grpc++/grpc++.h>
38 #include <grpc++/health_check_service_interface.h>
39 #include <memory>
40 #include <string>
41 #include <thread>
42 #include <utility>
43 
44 #if ENABLE_GRPC_REFLECTION
45 #include <grpc++/ext/proto_server_reflection_plugin.h>
46 #endif
47 
48 #ifdef __GLIBC__
49 #include <malloc.h>
50 #endif
51 
52 namespace clang {
53 namespace clangd {
54 namespace remote {
55 namespace {
56 
57 static constexpr char Overview[] = R"(
58 This is an experimental remote index implementation. The server opens Dex and
59 awaits gRPC lookup requests from the client.
60 )";
61 
62 llvm::cl::opt<std::string> IndexPath(llvm::cl::desc("<INDEX FILE>"),
63                                      llvm::cl::Positional, llvm::cl::Required);
64 
65 llvm::cl::opt<std::string> IndexRoot(llvm::cl::desc("<PROJECT ROOT>"),
66                                      llvm::cl::Positional, llvm::cl::Required);
67 
68 llvm::cl::opt<Logger::Level> LogLevel{
69     "log",
70     llvm::cl::desc("Verbosity of log messages written to stderr"),
71     values(clEnumValN(Logger::Error, "error", "Error messages only"),
72            clEnumValN(Logger::Info, "info", "High level execution tracing"),
73            clEnumValN(Logger::Debug, "verbose", "Low level details")),
74     llvm::cl::init(Logger::Info),
75 };
76 
77 llvm::cl::opt<bool> LogPublic{
78     "log-public",
79     llvm::cl::desc("Avoid logging potentially-sensitive request details"),
80     llvm::cl::init(false),
81 };
82 
83 llvm::cl::opt<std::string> LogPrefix{
84     "log-prefix",
85     llvm::cl::desc("A string that'll be prepended to all log statements. "
86                    "Useful when running multiple instances on same host."),
87 };
88 
89 llvm::cl::opt<std::string> TraceFile(
90     "trace-file",
91     llvm::cl::desc("Path to the file where tracer logs will be stored"));
92 
93 llvm::cl::opt<bool> PrettyPrint{
94     "pretty",
95     llvm::cl::desc("Pretty-print JSON output in the trace"),
96     llvm::cl::init(false),
97 };
98 
99 llvm::cl::opt<std::string> ServerAddress(
100     "server-address", llvm::cl::init("0.0.0.0:50051"),
101     llvm::cl::desc("Address of the invoked server. Defaults to 0.0.0.0:50051"));
102 
103 llvm::cl::opt<size_t> IdleTimeoutSeconds(
104     "idle-timeout", llvm::cl::init(8 * 60),
105     llvm::cl::desc("Maximum time a channel may stay idle until server closes "
106                    "the connection, in seconds. Defaults to 480."));
107 
108 llvm::cl::opt<size_t> LimitResults(
109     "limit-results", llvm::cl::init(10000),
110     llvm::cl::desc("Maximum number of results to stream as a response to "
111                    "single request. Limit is to keep the server from being "
112                    "DOS'd. Defaults to 10000."));
113 
114 static Key<grpc::ServerContext *> CurrentRequest;
115 
116 class RemoteIndexServer final : public v1::SymbolIndex::Service {
117 public:
118   RemoteIndexServer(clangd::SymbolIndex &Index, llvm::StringRef IndexRoot)
119       : Index(Index) {
120     llvm::SmallString<256> NativePath = IndexRoot;
121     llvm::sys::path::native(NativePath);
122     ProtobufMarshaller = std::unique_ptr<Marshaller>(new Marshaller(
123         /*RemoteIndexRoot=*/llvm::StringRef(NativePath),
124         /*LocalIndexRoot=*/""));
125   }
126 
127 private:
128   using stopwatch = std::chrono::steady_clock;
129 
130   grpc::Status Lookup(grpc::ServerContext *Context,
131                       const LookupRequest *Request,
132                       grpc::ServerWriter<LookupReply> *Reply) override {
133     auto StartTime = stopwatch::now();
134     WithContextValue WithRequestContext(CurrentRequest, Context);
135     logRequest(*Request);
136     trace::Span Tracer("LookupRequest");
137     auto Req = ProtobufMarshaller->fromProtobuf(Request);
138     if (!Req) {
139       elog("Can not parse LookupRequest from protobuf: {0}", Req.takeError());
140       return grpc::Status::CANCELLED;
141     }
142     unsigned Sent = 0;
143     unsigned FailedToSend = 0;
144     bool HasMore = false;
145     Index.lookup(*Req, [&](const clangd::Symbol &Item) {
146       if (Sent >= LimitResults) {
147         HasMore = true;
148         return;
149       }
150       auto SerializedItem = ProtobufMarshaller->toProtobuf(Item);
151       if (!SerializedItem) {
152         elog("Unable to convert Symbol to protobuf: {0}",
153              SerializedItem.takeError());
154         ++FailedToSend;
155         return;
156       }
157       LookupReply NextMessage;
158       *NextMessage.mutable_stream_result() = *SerializedItem;
159       logResponse(NextMessage);
160       Reply->Write(NextMessage);
161       ++Sent;
162     });
163     if (HasMore)
164       log("[public] Limiting result size for Lookup request.");
165     LookupReply LastMessage;
166     LastMessage.mutable_final_result()->set_has_more(HasMore);
167     logResponse(LastMessage);
168     Reply->Write(LastMessage);
169     SPAN_ATTACH(Tracer, "Sent", Sent);
170     SPAN_ATTACH(Tracer, "Failed to send", FailedToSend);
171     logRequestSummary("v1/Lookup", Sent, StartTime);
172     return grpc::Status::OK;
173   }
174 
175   grpc::Status FuzzyFind(grpc::ServerContext *Context,
176                          const FuzzyFindRequest *Request,
177                          grpc::ServerWriter<FuzzyFindReply> *Reply) override {
178     auto StartTime = stopwatch::now();
179     WithContextValue WithRequestContext(CurrentRequest, Context);
180     logRequest(*Request);
181     trace::Span Tracer("FuzzyFindRequest");
182     auto Req = ProtobufMarshaller->fromProtobuf(Request);
183     if (!Req) {
184       elog("Can not parse FuzzyFindRequest from protobuf: {0}",
185            Req.takeError());
186       return grpc::Status::CANCELLED;
187     }
188     if (!Req->Limit || *Req->Limit > LimitResults) {
189       log("[public] Limiting result size for FuzzyFind request from {0} to {1}",
190           Req->Limit, LimitResults);
191       Req->Limit = LimitResults;
192     }
193     unsigned Sent = 0;
194     unsigned FailedToSend = 0;
195     bool HasMore = Index.fuzzyFind(*Req, [&](const clangd::Symbol &Item) {
196       auto SerializedItem = ProtobufMarshaller->toProtobuf(Item);
197       if (!SerializedItem) {
198         elog("Unable to convert Symbol to protobuf: {0}",
199              SerializedItem.takeError());
200         ++FailedToSend;
201         return;
202       }
203       FuzzyFindReply NextMessage;
204       *NextMessage.mutable_stream_result() = *SerializedItem;
205       logResponse(NextMessage);
206       Reply->Write(NextMessage);
207       ++Sent;
208     });
209     FuzzyFindReply LastMessage;
210     LastMessage.mutable_final_result()->set_has_more(HasMore);
211     logResponse(LastMessage);
212     Reply->Write(LastMessage);
213     SPAN_ATTACH(Tracer, "Sent", Sent);
214     SPAN_ATTACH(Tracer, "Failed to send", FailedToSend);
215     logRequestSummary("v1/FuzzyFind", Sent, StartTime);
216     return grpc::Status::OK;
217   }
218 
219   grpc::Status Refs(grpc::ServerContext *Context, const RefsRequest *Request,
220                     grpc::ServerWriter<RefsReply> *Reply) override {
221     auto StartTime = stopwatch::now();
222     WithContextValue WithRequestContext(CurrentRequest, Context);
223     logRequest(*Request);
224     trace::Span Tracer("RefsRequest");
225     auto Req = ProtobufMarshaller->fromProtobuf(Request);
226     if (!Req) {
227       elog("Can not parse RefsRequest from protobuf: {0}", Req.takeError());
228       return grpc::Status::CANCELLED;
229     }
230     if (!Req->Limit || *Req->Limit > LimitResults) {
231       log("[public] Limiting result size for Refs request from {0} to {1}.",
232           Req->Limit, LimitResults);
233       Req->Limit = LimitResults;
234     }
235     unsigned Sent = 0;
236     unsigned FailedToSend = 0;
237     bool HasMore = Index.refs(*Req, [&](const clangd::Ref &Item) {
238       auto SerializedItem = ProtobufMarshaller->toProtobuf(Item);
239       if (!SerializedItem) {
240         elog("Unable to convert Ref to protobuf: {0}",
241              SerializedItem.takeError());
242         ++FailedToSend;
243         return;
244       }
245       RefsReply NextMessage;
246       *NextMessage.mutable_stream_result() = *SerializedItem;
247       logResponse(NextMessage);
248       Reply->Write(NextMessage);
249       ++Sent;
250     });
251     RefsReply LastMessage;
252     LastMessage.mutable_final_result()->set_has_more(HasMore);
253     logResponse(LastMessage);
254     Reply->Write(LastMessage);
255     SPAN_ATTACH(Tracer, "Sent", Sent);
256     SPAN_ATTACH(Tracer, "Failed to send", FailedToSend);
257     logRequestSummary("v1/Refs", Sent, StartTime);
258     return grpc::Status::OK;
259   }
260 
261   grpc::Status Relations(grpc::ServerContext *Context,
262                          const RelationsRequest *Request,
263                          grpc::ServerWriter<RelationsReply> *Reply) override {
264     auto StartTime = stopwatch::now();
265     WithContextValue WithRequestContext(CurrentRequest, Context);
266     logRequest(*Request);
267     trace::Span Tracer("RelationsRequest");
268     auto Req = ProtobufMarshaller->fromProtobuf(Request);
269     if (!Req) {
270       elog("Can not parse RelationsRequest from protobuf: {0}",
271            Req.takeError());
272       return grpc::Status::CANCELLED;
273     }
274     if (!Req->Limit || *Req->Limit > LimitResults) {
275       log("[public] Limiting result size for Relations request from {0} to "
276           "{1}.",
277           Req->Limit, LimitResults);
278       Req->Limit = LimitResults;
279     }
280     unsigned Sent = 0;
281     unsigned FailedToSend = 0;
282     Index.relations(
283         *Req, [&](const SymbolID &Subject, const clangd::Symbol &Object) {
284           auto SerializedItem = ProtobufMarshaller->toProtobuf(Subject, Object);
285           if (!SerializedItem) {
286             elog("Unable to convert Relation to protobuf: {0}",
287                  SerializedItem.takeError());
288             ++FailedToSend;
289             return;
290           }
291           RelationsReply NextMessage;
292           *NextMessage.mutable_stream_result() = *SerializedItem;
293           logResponse(NextMessage);
294           Reply->Write(NextMessage);
295           ++Sent;
296         });
297     RelationsReply LastMessage;
298     LastMessage.mutable_final_result()->set_has_more(true);
299     logResponse(LastMessage);
300     Reply->Write(LastMessage);
301     SPAN_ATTACH(Tracer, "Sent", Sent);
302     SPAN_ATTACH(Tracer, "Failed to send", FailedToSend);
303     logRequestSummary("v1/Relations", Sent, StartTime);
304     return grpc::Status::OK;
305   }
306 
307   // Proxy object to allow proto messages to be lazily serialized as text.
308   struct TextProto {
309     const google::protobuf::Message &M;
310     friend llvm::raw_ostream &operator<<(llvm::raw_ostream &OS,
311                                          const TextProto &P) {
312       return OS << P.M.DebugString();
313     }
314   };
315 
316   void logRequest(const google::protobuf::Message &M) {
317     vlog("<<< {0}\n{1}", M.GetDescriptor()->name(), TextProto{M});
318   }
319   void logResponse(const google::protobuf::Message &M) {
320     vlog(">>> {0}\n{1}", M.GetDescriptor()->name(), TextProto{M});
321   }
322   void logRequestSummary(llvm::StringLiteral RequestName, unsigned Sent,
323                          stopwatch::time_point StartTime) {
324     auto Duration = stopwatch::now() - StartTime;
325     auto Millis =
326         std::chrono::duration_cast<std::chrono::milliseconds>(Duration).count();
327     log("[public] request {0} => OK: {1} results in {2}ms", RequestName, Sent,
328         Millis);
329   }
330 
331   std::unique_ptr<Marshaller> ProtobufMarshaller;
332   clangd::SymbolIndex &Index;
333 };
334 
335 class Monitor final : public v1::Monitor::Service {
336 public:
337   Monitor(llvm::sys::TimePoint<> IndexAge)
338       : StartTime(std::chrono::system_clock::now()), IndexBuildTime(IndexAge) {}
339 
340   void updateIndex(llvm::sys::TimePoint<> UpdateTime) {
341     IndexBuildTime.exchange(UpdateTime);
342   }
343 
344 private:
345   // FIXME(kirillbobyrev): Most fields should be populated when the index
346   // reloads (probably in adjacent metadata.txt file next to loaded .idx) but
347   // they aren't right now.
348   grpc::Status MonitoringInfo(grpc::ServerContext *Context,
349                               const v1::MonitoringInfoRequest *Request,
350                               v1::MonitoringInfoReply *Reply) override {
351     Reply->set_uptime_seconds(std::chrono::duration_cast<std::chrono::seconds>(
352                                   std::chrono::system_clock::now() - StartTime)
353                                   .count());
354     // FIXME(kirillbobyrev): We are currently making use of the last
355     // modification time of the index artifact to deduce its age. This is wrong
356     // as it doesn't account for the indexing delay. Propagate some metadata
357     // with the index artifacts to indicate time of the commit we indexed.
358     Reply->set_index_age_seconds(
359         std::chrono::duration_cast<std::chrono::seconds>(
360             std::chrono::system_clock::now() - IndexBuildTime.load())
361             .count());
362     return grpc::Status::OK;
363   }
364 
365   const llvm::sys::TimePoint<> StartTime;
366   std::atomic<llvm::sys::TimePoint<>> IndexBuildTime;
367 };
368 
369 void maybeTrimMemory() {
370 #if defined(__GLIBC__) && CLANGD_MALLOC_TRIM
371   malloc_trim(0);
372 #endif
373 }
374 
375 // Detect changes in \p IndexPath file and load new versions of the index
376 // whenever they become available.
377 void hotReload(clangd::SwapIndex &Index, llvm::StringRef IndexPath,
378                llvm::vfs::Status &LastStatus,
379                llvm::IntrusiveRefCntPtr<llvm::vfs::FileSystem> &FS,
380                Monitor &Monitor) {
381   // glibc malloc doesn't shrink an arena if there are items living at the end,
382   // which might happen since we destroy the old index after building new one.
383   // Trim more aggresively to keep memory usage of the server low.
384   // Note that we do it deliberately here rather than after Index.reset(),
385   // because old index might still be kept alive after the reset call if we are
386   // serving requests.
387   maybeTrimMemory();
388   auto Status = FS->status(IndexPath);
389   // Requested file is same as loaded index: no reload is needed.
390   if (!Status || (Status->getLastModificationTime() ==
391                       LastStatus.getLastModificationTime() &&
392                   Status->getSize() == LastStatus.getSize()))
393     return;
394   vlog("Found different index version: existing index was modified at "
395        "{0}, new index was modified at {1}. Attempting to reload.",
396        LastStatus.getLastModificationTime(), Status->getLastModificationTime());
397   LastStatus = *Status;
398   std::unique_ptr<clang::clangd::SymbolIndex> NewIndex = loadIndex(IndexPath);
399   if (!NewIndex) {
400     elog("Failed to load new index. Old index will be served.");
401     return;
402   }
403   Index.reset(std::move(NewIndex));
404   Monitor.updateIndex(Status->getLastModificationTime());
405   log("New index version loaded. Last modification time: {0}, size: {1} bytes.",
406       Status->getLastModificationTime(), Status->getSize());
407 }
408 
409 void runServerAndWait(clangd::SymbolIndex &Index, llvm::StringRef ServerAddress,
410                       llvm::StringRef IndexPath, Monitor &Monitor) {
411   RemoteIndexServer Service(Index, IndexRoot);
412 
413   grpc::EnableDefaultHealthCheckService(true);
414 #if ENABLE_GRPC_REFLECTION
415   grpc::reflection::InitProtoReflectionServerBuilderPlugin();
416 #endif
417   grpc::ServerBuilder Builder;
418   Builder.AddListeningPort(ServerAddress.str(),
419                            grpc::InsecureServerCredentials());
420   Builder.AddChannelArgument(GRPC_ARG_MAX_CONNECTION_IDLE_MS,
421                              IdleTimeoutSeconds * 1000);
422   Builder.RegisterService(&Service);
423   Builder.RegisterService(&Monitor);
424   std::unique_ptr<grpc::Server> Server(Builder.BuildAndStart());
425   log("Server listening on {0}", ServerAddress);
426 
427   std::thread ServerShutdownWatcher([&]() {
428     static constexpr auto WatcherFrequency = std::chrono::seconds(5);
429     while (!clang::clangd::shutdownRequested())
430       std::this_thread::sleep_for(WatcherFrequency);
431     Server->Shutdown();
432   });
433 
434   Server->Wait();
435   ServerShutdownWatcher.join();
436 }
437 
438 std::unique_ptr<Logger> makeLogger(llvm::StringRef LogPrefix,
439                                    llvm::raw_ostream &OS) {
440   std::unique_ptr<Logger> Base;
441   if (LogPublic) {
442     // Redacted mode:
443     //  - messages outside the scope of a request: log fully
444     //  - messages tagged [public]: log fully
445     //  - errors: log the format string
446     //  - others: drop
447     class RedactedLogger : public StreamLogger {
448     public:
449       using StreamLogger::StreamLogger;
450       void log(Level L, const char *Fmt,
451                const llvm::formatv_object_base &Message) override {
452         if (Context::current().get(CurrentRequest) == nullptr ||
453             llvm::StringRef(Fmt).startswith("[public]"))
454           return StreamLogger::log(L, Fmt, Message);
455         if (L >= Error)
456           return StreamLogger::log(L, Fmt,
457                                    llvm::formatv("[redacted] {0}", Fmt));
458       }
459     };
460     Base = std::make_unique<RedactedLogger>(OS, LogLevel);
461   } else {
462     Base = std::make_unique<StreamLogger>(OS, LogLevel);
463   }
464 
465   if (LogPrefix.empty())
466     return Base;
467   class PrefixedLogger : public Logger {
468     std::string LogPrefix;
469     std::unique_ptr<Logger> Base;
470 
471   public:
472     PrefixedLogger(llvm::StringRef LogPrefix, std::unique_ptr<Logger> Base)
473         : LogPrefix(LogPrefix.str()), Base(std::move(Base)) {}
474     void log(Level L, const char *Fmt,
475              const llvm::formatv_object_base &Message) override {
476       Base->log(L, Fmt, llvm::formatv("[{0}] {1}", LogPrefix, Message));
477     }
478   };
479   return std::make_unique<PrefixedLogger>(LogPrefix, std::move(Base));
480 }
481 
482 } // namespace
483 } // namespace remote
484 } // namespace clangd
485 } // namespace clang
486 
487 using clang::clangd::elog;
488 
489 int main(int argc, char *argv[]) {
490   using namespace clang::clangd::remote;
491   llvm::cl::ParseCommandLineOptions(argc, argv, Overview);
492   llvm::sys::PrintStackTraceOnErrorSignal(argv[0]);
493   llvm::sys::SetInterruptFunction(&clang::clangd::requestShutdown);
494 
495   if (!llvm::sys::path::is_absolute(IndexRoot)) {
496     llvm::errs() << "Index root should be an absolute path.\n";
497     return -1;
498   }
499 
500   llvm::errs().SetBuffered();
501   // Don't flush stdout when logging for thread safety.
502   llvm::errs().tie(nullptr);
503   auto Logger = makeLogger(LogPrefix.getValue(), llvm::errs());
504   clang::clangd::LoggingSession LoggingSession(*Logger);
505 
506   llvm::Optional<llvm::raw_fd_ostream> TracerStream;
507   std::unique_ptr<clang::clangd::trace::EventTracer> Tracer;
508   if (!TraceFile.empty()) {
509     std::error_code EC;
510     TracerStream.emplace(TraceFile, EC,
511                          llvm::sys::fs::FA_Read | llvm::sys::fs::FA_Write);
512     if (EC) {
513       TracerStream.reset();
514       elog("Error while opening trace file {0}: {1}", TraceFile, EC.message());
515     } else {
516       // FIXME(kirillbobyrev): Also create metrics tracer to track latency and
517       // accumulate other request statistics.
518       Tracer = clang::clangd::trace::createJSONTracer(*TracerStream,
519                                                       /*PrettyPrint=*/false);
520       clang::clangd::vlog("Successfully created a tracer.");
521     }
522   }
523 
524   llvm::Optional<clang::clangd::trace::Session> TracingSession;
525   if (Tracer)
526     TracingSession.emplace(*Tracer);
527 
528   clang::clangd::RealThreadsafeFS TFS;
529   auto FS = TFS.view(llvm::None);
530   auto Status = FS->status(IndexPath);
531   if (!Status) {
532     elog("{0} does not exist.", IndexPath);
533     return Status.getError().value();
534   }
535 
536   auto SymIndex = clang::clangd::loadIndex(IndexPath);
537   if (!SymIndex) {
538     llvm::errs() << "Failed to open the index.\n";
539     return -1;
540   }
541   clang::clangd::SwapIndex Index(std::move(SymIndex));
542 
543   Monitor Monitor(Status->getLastModificationTime());
544 
545   std::thread HotReloadThread([&Index, &Status, &FS, &Monitor]() {
546     llvm::vfs::Status LastStatus = *Status;
547     static constexpr auto RefreshFrequency = std::chrono::seconds(30);
548     while (!clang::clangd::shutdownRequested()) {
549       hotReload(Index, llvm::StringRef(IndexPath), LastStatus, FS, Monitor);
550       std::this_thread::sleep_for(RefreshFrequency);
551     }
552   });
553 
554   runServerAndWait(Index, ServerAddress, IndexPath, Monitor);
555 
556   HotReloadThread.join();
557 }
558