src,agents: add support for source code collection

agents/grpc/proto/command.proto

@@ -7,13 +7,15 @@ syntax = "proto3";
 
 import "profile.proto";
 import "reconfigure.proto";
+import "source_code.proto";
 
 package grpcagent;
 
 message CommandArgs {
   oneof args {
     ReconfigureBody reconfigure = 1;
     ProfileArgs profile = 2;
+    SourceCodeArgs source_code = 3;
   }
 }
 

agents/grpc/proto/nsolid_service.proto

@@ -8,6 +8,7 @@ import "info.proto";
 import "metrics.proto";
 import "packages.proto";
 import "reconfigure.proto";
+import "source_code.proto";
 import "startup_times.proto";
 
 package grpcagent;
@@ -22,6 +23,7 @@ service NSolidService {
   rpc ExportBlockedLoop (BlockedLoopEvent) returns (EventResponse) {}
   rpc ExportUnblockedLoop (UnblockedLoopEvent) returns (EventResponse) {}
   rpc ExportReconfigure (ReconfigureEvent) returns (EventResponse) {}
+  rpc ExportSourceCode (SourceCodeEvent) returns (EventResponse) {}
   rpc ExportStartupTimes (StartupTimesEvent) returns (EventResponse) {}
 }
 

agents/grpc/proto/source_code.proto

@@ -0,0 +1,19 @@
+syntax = "proto3";
+
+import "common.proto";
+
+package grpcagent;
+
+message SourceCodeArgs {
+  int64 thread_id = 1;
+  int32 script_id = 2;
+  string path = 3;
+}
+
+message SourceCodeEvent {
+  CommonResponse common = 1;
+  int64 thread_id = 2;
+  string path = 3;
+  string code = 4;
+  bool is_esm = 5;
+}

agents/grpc/src/grpc_agent.cc

@@ -1440,6 +1440,8 @@ void GrpcAgent::handle_command_request(CommandRequestStor&& req) {
     start_heap_sampling(request);
   } else if (cmd == "snapshot") {
     start_heap_snapshot(request);
+  } else if (cmd == "source_code") {
+    send_source_code_event(request);
   } else if (cmd == "startup_times") {
     send_startup_times_event(request.requestid().c_str());
   } else {
@@ -1725,6 +1727,56 @@ void GrpcAgent::send_reconfigure_event(const char* req_id) {
     });
 }
 
+void GrpcAgent::send_source_code_event(const grpcagent::CommandRequest& req) {
+  const grpcagent::SourceCodeArgs& args = req.args().source_code();
+  uint64_t thread_id = args.thread_id();
+  int script_id = args.script_id();
+  const std::string path = args.path();
+
+  ArenaOptions arena_options;
+  arena_options.initial_block_size = 1024;
+  arena_options.max_block_size = 65536;
+  std::unique_ptr<Arena> arena{new Arena{arena_options}};
+
+  auto source_code_event =
+      Arena::Create<grpcagent::SourceCodeEvent>(arena.get());
+  PopulateCommon(source_code_event->mutable_common(),
+                 "source_code",
+                 req.requestid().c_str());
+
+  source_code_event->set_thread_id(thread_id);
+  source_code_event->set_path(path);
+
+
+  auto envinst_sp = EnvInst::GetInst(thread_id);
+  if (envinst_sp == nullptr) {
+    Debug("Error getting EnvInst for thread: %ld\n", thread_id);
+    PopulateError(source_code_event->mutable_common(),
+                  ErrorType::EThreadGoneError);
+  } else {
+    int r = envinst_sp->GetSourceCode(script_id,
+                                      path,
+                                      source_code_event->mutable_code());
+    if (r != 0) {
+      Debug("Error reading file from: %s. Error: %d.\n", path.c_str(), r);
+      PopulateError(source_code_event->mutable_common(),
+                    ErrorType::ESourceCodeFileError);
+    }
+  }
+
+  auto context = GrpcClient::MakeClientContext(agent_id_, saas_);
+
+  GrpcClient::DelegateAsyncExport(
+    nsolid_service_stub_.get(), std::move(context), std::move(arena),
+    std::move(*source_code_event),
+    [](::grpc::Status,
+        std::unique_ptr<Arena>&&,
+        const grpcagent::SourceCodeEvent& info_event,
+        grpcagent::EventResponse*) {
+      return true;
+    });
+}
+
 void GrpcAgent::send_startup_times_event(const char* req_id) {
   ArenaOptions arena_options;
   arena_options.initial_block_size = 1024;

agents/grpc/src/grpc_agent.h

@@ -267,6 +267,8 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
 
   void send_reconfigure_event(const char* req_id);
 
+  void send_source_code_event(const grpcagent::CommandRequest& req);
+
   void send_startup_times_event(const char* req_id);
 
   void send_unblocked_loop_event(BlockedLoopStor&& stor);

agents/grpc/src/grpc_client.cc

@@ -273,6 +273,25 @@ int GrpcClient::DelegateAsyncExport(
 }
 
 
+int GrpcClient::DelegateAsyncExport(
+    NSolidService::StubInterface* stub,
+    std::unique_ptr<ClientContext>&& context,
+    std::unique_ptr<Arena>&& arena,
+    grpcagent::SourceCodeEvent&& event,
+    std::function<bool(Status,
+                       std::unique_ptr<Arena> &&,
+                       const grpcagent::SourceCodeEvent&,
+                       grpcagent::EventResponse*)>&& result_callback) noexcept {
+  return InternalDelegateAsyncExport<grpcagent::SourceCodeEvent>(
+      stub,
+      &NSolidService::StubInterface::async_interface::ExportSourceCode,
+      std::move(context),
+      std::move(arena),
+      std::move(event),
+      std::move(result_callback));
+}
+
+
 int GrpcClient::DelegateAsyncExport(
     NSolidService::StubInterface* stub,
     std::unique_ptr<ClientContext>&& context,

agents/grpc/src/grpc_client.h

@@ -113,6 +113,16 @@ class GrpcClient {
                         const grpcagent::ReconfigureEvent&,
                         grpcagent::EventResponse*)>&& result_callback) noexcept;
 
+  static int DelegateAsyncExport(
+    grpcagent::NSolidService::StubInterface* stub,
+    std::unique_ptr<::grpc::ClientContext>&& context,
+    std::unique_ptr<google::protobuf::Arena>&& arena,
+    grpcagent::SourceCodeEvent&& event,
+    std::function<bool(::grpc::Status,
+                        std::unique_ptr<google::protobuf::Arena> &&,
+                        const grpcagent::SourceCodeEvent&,
+                        grpcagent::EventResponse*)>&& result_callback) noexcept;
+
   static int DelegateAsyncExport(
     grpcagent::NSolidService::StubInterface* stub,
     std::unique_ptr<::grpc::ClientContext>&& context,

agents/grpc/src/grpc_errors.h

@@ -8,7 +8,8 @@
   X(EProfSnapshotError, 500, "Profile/Snapshot creation failure", 1003)        \
   X(ESnapshotDisabled, 500, "Heap Snapshots disabled", 1004)                   \
   X(ENoMemory, 500, "Internal Runtime Error", 1005)                            \
-  X(ENotAvailable, 404, "Resource not available", 1006)
+  X(ENotAvailable, 404, "Resource not available", 1006)                        \
+  X(ESourceCodeFileError, 500, "Internal Runtime Error", 1007)
 
 namespace node {
 namespace nsolid {