frontend-grpc: initial access to the catalog

frontend_svc/gRPC/CMakeLists.txt

@@ -21,6 +21,7 @@ include_directories(${CMAKE_BINARY_DIR}/eos_cta
                     ${PROTOBUF3_INCLUDE_DIRS})
 
 add_executable(frontend-grpc-svc FrontendGRpcSvc.cpp)
+
 target_link_libraries(frontend-grpc-svc CtaGRpc
         ${PROTOBUF3_LIBRARIES} ${GRPC_GRPC++_LIBRARY}
         ctascheduler ctacommon ctaobjectstore ctacatalogue)

frontend_svc/gRPC/FrontendGRpcSvc.cpp

@@ -18,59 +18,62 @@
 #include "FrontendGRpcSvc.h"
 #include "version.h"
 
-using grpc::Server;
-using grpc::ServerBuilder;
-using grpc::ServerContext;
-using grpc::Status;
+Status CtaRpcImpl::Version(::grpc::ServerContext *context, const ::google::protobuf::Empty *request,
+                           ::cta::admin::Version *response) {
+    response->set_cta_version(CTA_VERSION);
+    response->set_xrootd_ssi_protobuf_interface_version("gPRC-frontend v0.0.1");
+    return Status::OK;
+}
 
-using cta::rpc::CtaRpc;
-
-#include "common/log/SyslogLogger.hpp"
-#include "common/log/StdoutLogger.hpp"
-#include "common/log/FileLogger.hpp"
-#include "common/log/LogLevel.hpp"
+void CtaRpcImpl::run(const std::string server_address) {
 
+    ServerBuilder builder;
+    // Listen on the given address without any authentication mechanism.
+    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
 
-class CtaRpcImpl final : public CtaRpc::Service {
+    // Register "service" as the instance through which we'll communicate with
+    // clients. In this case it corresponds to an *synchronous* service.
+    builder.RegisterService(this);
+    std::unique_ptr <Server> server(builder.BuildAndStart());
 
-    Status Version(::grpc::ServerContext* context, const ::google::protobuf::Empty* request, ::cta::admin::Version* response) override {
-        response->set_cta_version(CTA_VERSION);
-        response->set_xrootd_ssi_protobuf_interface_version("gPRC-frontend v0.0.1");
-        return Status::OK;
-    }
-};
+    m_log->log(cta::log::INFO, "Listening on socket address: " + server_address);
+    server->Wait();
+}
 
+CtaRpcImpl::CtaRpcImpl(cta::log::LogContext *lc, std::unique_ptr<cta::catalogue::Catalogue> &catalogue):
+    m_catalogue(std::move(catalogue)) {
+    m_log = lc;
+}
 
 using namespace cta;
 
 int main(const int argc, char *const *const argv) {
 
-    std::unique_ptr <cta::log::Logger> m_log;
-    m_log.reset(new log::StdoutLogger("cta-dev", "cta-grpc-frontend"));
+    std::unique_ptr <cta::log::Logger> logger;
+    logger.reset(new log::StdoutLogger("cta-dev", "cta-grpc-frontend"));
 
-    log::LogContext lc(*m_log);
+    log::LogContext lc(*logger);
 
     std::string server_address("0.0.0.0:17017");
 
-    CtaRpcImpl svc;
-
-//    grpc::EnableDefaultHealthCheckService(true);
-//    grpc::reflection::InitProtoReflectionServerBuilderPlugin();
-
-    ServerBuilder builder;
-    // Listen on the given address without any authentication mechanism.
-    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
-    // Register "service" as the instance through which we'll communicate with
-    // clients. In this case it corresponds to an *synchronous* service.
-    builder.RegisterService(&svc);
-    // Finally assemble the server.
-    std::unique_ptr <Server> server(builder.BuildAndStart());
-
-    lc.log(log::INFO, "Listening on socket address: " + server_address);
-
-    // Wait for the server to shutdown. Note that some other thread must be
-    // responsible for shutting down the server for this call to ever return.
-    server->Wait();
+    // Initialise the Catalogue
+    std::string catalogueConfigFile = "/etc/cta/cta-catalogue.conf";
+    const rdbms::Login catalogueLogin = rdbms::Login::parseFile(catalogueConfigFile);
+
+    const uint64_t nbArchiveFileListingConns = 2;
+    auto catalogueFactory = catalogue::CatalogueFactoryFactory::create(*logger, catalogueLogin,
+                                                                       10,
+                                                                       nbArchiveFileListingConns);
+    auto catalogue = catalogueFactory->create();
+    try {
+        catalogue->ping();
+        lc.log(log::INFO, "Connected to catalog " + catalogue->getSchemaVersion().getSchemaVersion<std::string>());
+    } catch (cta::exception::Exception &ex) {
+        lc.log(cta::log::CRIT, ex.getMessageValue());
+        exit(1);
+    }
 
+    CtaRpcImpl svc(&lc, catalogue);
+    svc.run(server_address);
 
 }
\ No newline at end of file

frontend_svc/gRPC/FrontendGRpcSvc.h

@@ -18,5 +18,31 @@
 
 #include "cta_rpc.grpc.pb.h"
 
+using cta::Scheduler;
+using cta::SchedulerDBInit_t;
+using cta::SchedulerDB_t;
+using cta::catalogue::Catalogue;
+using cta::rdbms::Login;
+using cta::rpc::CtaRpc;
+
+using grpc::Server;
+using grpc::ServerBuilder;
+using grpc::ServerContext;
+using grpc::Status;
+
+class CtaRpcImpl : public CtaRpc::Service {
+
+private:
+    std::unique_ptr <cta::catalogue::Catalogue> m_catalogue;
+    std::unique_ptr <cta::SchedulerDB_t> m_scheddb;
+    std::unique_ptr <cta::SchedulerDBInit_t> m_scheddb_init;
+    std::unique_ptr <cta::Scheduler> m_scheduler;
+    cta::log::LogContext  *m_log;
+
+public:
+    CtaRpcImpl(cta::log::LogContext *lc, std::unique_ptr<cta::catalogue::Catalogue> &catalogue);
+    void run(const std::string server_address);
+    Status Version(::grpc::ServerContext *context, const ::google::protobuf::Empty *request, ::cta::admin::Version *response);
+};
 
 #endif //CTA_FRONTENDGRPCSVC_H