MapDServer.cpp

/*
 * Copyright 2018 OmniSci, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "DataMgr/ForeignStorage/ForeignStorageInterface.h"
#include "ThriftHandler/DBHandler.h"

#ifdef HAVE_THRIFT_THREADFACTORY
#include <thrift/concurrency/ThreadFactory.h>
#else
#include <thrift/concurrency/PlatformThreadFactory.h>
#endif

#include <thrift/concurrency/ThreadManager.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TJSONProtocol.h>
#include <thrift/server/TThreadedServer.h>
#include <thrift/transport/TBufferTransports.h>
#include <thrift/transport/THttpServer.h>
#include <thrift/transport/TSSLServerSocket.h>
#include <thrift/transport/TSSLSocket.h>
#include <thrift/transport/TServerSocket.h>

#include "Logger/Logger.h"
#include "Shared/SystemParameters.h"
#include "Shared/file_delete.h"
#include "Shared/mapd_shared_mutex.h"
#include "Shared/mapd_shared_ptr.h"
#include "Shared/scope.h"

#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/trim.hpp>
#include <boost/filesystem.hpp>
#include <boost/locale/generator.hpp>
#include <boost/make_shared.hpp>
#include <boost/program_options.hpp>

#include <csignal>
#include <cstdlib>
#include <sstream>
#include <thread>
#include <vector>

#ifdef HAVE_AWS_S3
#include "DataMgr/OmniSciAwsSdk.h"
#endif
#include "MapDRelease.h"
#include "Shared/Compressor.h"
#include "Shared/SystemParameters.h"
#include "Shared/file_delete.h"
#include "Shared/mapd_shared_ptr.h"
#include "Shared/scope.h"
#include "ThriftHandler/ForeignTableRefreshScheduler.h"

using namespace ::apache::thrift;
using namespace ::apache::thrift::concurrency;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::server;
using namespace ::apache::thrift::transport;

extern bool g_enable_thrift_logs;

std::atomic<bool> g_running{true};
std::atomic<int> g_saw_signal{-1};

mapd_shared_mutex g_thrift_mutex;
TThreadedServer* g_thrift_http_server{nullptr};
TThreadedServer* g_thrift_buf_server{nullptr};

mapd::shared_ptr<DBHandler> g_warmup_handler =
    0;  // global "g_warmup_handler" needed to avoid circular dependency
// between "DBHandler" & function "run_warmup_queries"
mapd::shared_ptr<DBHandler> g_mapd_handler = 0;
std::once_flag g_shutdown_once_flag;

void shutdown_handler() {
  if (g_mapd_handler) {
    std::call_once(g_shutdown_once_flag, []() { g_mapd_handler->shutdown(); });
  }
}

void register_signal_handler(int signum, void (*handler)(int)) {
#ifdef _WIN32
  signal(signum, handler);
#else
  struct sigaction act;
  memset(&act, 0, sizeof(act));
  if (handler != SIG_DFL && handler != SIG_IGN) {
    // block all signal deliveries while inside the signal handler
    sigfillset(&act.sa_mask);
  }
  act.sa_handler = handler;
  sigaction(signum, &act, NULL);
#endif
}

// Signal handler to set a global flag telling the server to exit.
// Do not call other functions inside this (or any) signal handler
// unless you really know what you are doing. See also:
//   man 7 signal-safety
//   man 7 signal
//   https://en.wikipedia.org/wiki/Reentrancy_(computing)
void omnisci_signal_handler(int signum) {
  // Record the signal number for logging during shutdown.
  // Only records the first signal if called more than once.
  int expected_signal{-1};
  if (!g_saw_signal.compare_exchange_strong(expected_signal, signum)) {
    return;  // this wasn't the first signal
  }

  // This point should never be reached more than once.

  // Tell heartbeat() to shutdown by unsetting the 'g_running' flag.
  // If 'g_running' is already false, this has no effect and the
  // shutdown is already in progress.
  g_running = false;

  // Handle core dumps specially by pausing inside this signal handler
  // because on some systems, some signals will execute their default
  // action immediately when and if the signal handler returns.
  // We would like to do some emergency cleanup before core dump.
  if (signum == SIGABRT || signum == SIGSEGV || signum == SIGFPE
#ifndef _WIN32
      || signum == SIGQUIT
#endif
  ) {
    // Wait briefly to give heartbeat() a chance to flush the logs and
    // do any other emergency shutdown tasks.
    std::this_thread::sleep_for(std::chrono::seconds(2));

    // Explicitly trigger whatever default action this signal would
    // have done, such as terminate the process or dump core.
    // Signals are currently blocked so this new signal will be queued
    // until this signal handler returns.
    register_signal_handler(signum, SIG_DFL);
#ifdef _WIN32
    raise(signum);
#else
    kill(getpid(), signum);
#endif
    std::this_thread::sleep_for(std::chrono::seconds(5));

#ifndef __APPLE__
    // as a last resort, abort
    // primary used in Docker environments, where we can end up with PID 1 and fail to
    // catch unix signals
    quick_exit(signum);
#endif
  }
}

void register_signal_handlers() {
  register_signal_handler(SIGINT, omnisci_signal_handler);
#ifndef _WIN32
  register_signal_handler(SIGQUIT, omnisci_signal_handler);
  register_signal_handler(SIGHUP, omnisci_signal_handler);
#endif
  register_signal_handler(SIGTERM, omnisci_signal_handler);
  register_signal_handler(SIGSEGV, omnisci_signal_handler);
  register_signal_handler(SIGABRT, omnisci_signal_handler);
#ifndef _WIN32
  // Thrift secure socket can cause problems with SIGPIPE
  register_signal_handler(SIGPIPE, SIG_IGN);
#endif
}

void start_server(TThreadedServer& server, const int port) {
  try {
    server.serve();
    if (errno != 0) {
      throw std::runtime_error(std::string("Thrift server exited: ") +
                               std::strerror(errno));
    }
  } catch (std::exception& e) {
    LOG(ERROR) << "Exception: " << e.what() << ": port " << port << std::endl;
  }
}

void releaseWarmupSession(TSessionId& sessionId, std::ifstream& query_file) noexcept {
  query_file.close();
  if (sessionId != g_warmup_handler->getInvalidSessionId()) {
    try {
      g_warmup_handler->disconnect(sessionId);
    } catch (...) {
      LOG(ERROR) << "Failed to disconnect warmup session, possible failure to run warmup "
                    "queries.";
    }
  }
}

void run_warmup_queries(mapd::shared_ptr<DBHandler> handler,
                        std::string base_path,
                        std::string query_file_path) {
  // run warmup queries to load cache if requested
  if (query_file_path.empty()) {
    return;
  }
  if (handler->isAggregator()) {
    LOG(INFO) << "Skipping warmup query execution on the aggregator, queries should be "
                 "run directly on the leaf nodes.";
    return;
  }

  LOG(INFO) << "Running DB warmup with queries from " << query_file_path;
  try {
    g_warmup_handler = handler;
    std::string db_info;
    std::string user_keyword, user_name, db_name;
    std::ifstream query_file;
    Catalog_Namespace::UserMetadata user;
    Catalog_Namespace::DBMetadata db;
    TSessionId sessionId = g_warmup_handler->getInvalidSessionId();

    ScopeGuard session_guard = [&] { releaseWarmupSession(sessionId, query_file); };
    query_file.open(query_file_path);
    while (std::getline(query_file, db_info)) {
      if (db_info.length() == 0) {
        continue;
      }
      std::istringstream iss(db_info);
      iss >> user_keyword >> user_name >> db_name;
      if (user_keyword.compare(0, 4, "USER") == 0) {
        // connect to DB for given user_name/db_name with super_user_rights (without
        // password), & start session
        g_warmup_handler->super_user_rights_ = true;
        g_warmup_handler->connect(sessionId, user_name, "", db_name);
        g_warmup_handler->super_user_rights_ = false;

        // read and run one query at a time for the DB with the setup connection
        TQueryResult ret;
        std::string single_query;
        while (std::getline(query_file, single_query)) {
          boost::algorithm::trim(single_query);
          if (single_query.length() == 0 || single_query[0] == '-') {
            continue;
          }
          if (single_query[0] == '}') {
            single_query.clear();
            break;
          }
          if (single_query.find(';') == single_query.npos) {
            std::string multiline_query;
            std::getline(query_file, multiline_query, ';');
            single_query += multiline_query;
          }

          try {
            g_warmup_handler->sql_execute(ret, sessionId, single_query, true, "", -1, -1);
          } catch (...) {
            LOG(WARNING) << "Exception while executing '" << single_query
                         << "', ignoring";
          }
          single_query.clear();
        }

        // stop session and disconnect from the DB
        g_warmup_handler->disconnect(sessionId);
        sessionId = g_warmup_handler->getInvalidSessionId();
      } else {
        LOG(WARNING) << "\nSyntax error in the file: " << query_file_path.c_str()
                     << " Missing expected keyword USER. Following line will be ignored: "
                     << db_info.c_str() << std::endl;
      }
      db_info.clear();
    }
  } catch (const std::exception& e) {
    LOG(WARNING)
        << "Exception while executing warmup queries. "
        << "Warmup may not be fully completed. Will proceed nevertheless.\nError was: "
        << e.what();
  }
}

void heartbeat() {
#ifndef _WIN32
  // Block all signals for this heartbeat thread, only.
  sigset_t set;
  sigfillset(&set);
  int result = pthread_sigmask(SIG_BLOCK, &set, NULL);
  if (result != 0) {
    throw std::runtime_error("heartbeat() thread startup failed");
  }
#endif

  // Sleep until omnisci_signal_handler or anything clears the g_running flag.
  VLOG(1) << "heartbeat thread starting";
  while (::g_running) {
    using namespace std::chrono;
    std::this_thread::sleep_for(1s);
  }
  VLOG(1) << "heartbeat thread exiting";

  // Get the signal number if there was a signal.
  int signum = g_saw_signal;
  if (signum >= 1 && signum != SIGTERM) {
    LOG(INFO) << "Interrupt signal (" << signum << ") received.";
  }

  // if dumping core, try to do some quick stuff
  if (signum == SIGABRT || signum == SIGSEGV || signum == SIGFPE
#ifndef _WIN32
      || signum == SIGQUIT
#endif
  ) {
    if (g_mapd_handler) {
      std::call_once(g_shutdown_once_flag,
                     []() { g_mapd_handler->emergency_shutdown(); });
    }
    logger::shutdown();
    return;
    // core dump should begin soon after this, see omnisci_signal_handler()
  }

  // trigger an orderly shutdown by telling Thrift to stop serving
  {
    mapd_shared_lock<mapd_shared_mutex> read_lock(g_thrift_mutex);
    auto httpserv = g_thrift_http_server;
    if (httpserv) {
      httpserv->stop();
    }
    auto bufserv = g_thrift_buf_server;
    if (bufserv) {
      bufserv->stop();
    }
    // main() should return soon after this
  }
}

int startMapdServer(CommandLineOptions& prog_config_opts, bool start_http_server = true) {
  // try to enforce an orderly shutdown even after a signal
  register_signal_handlers();

  // register shutdown procedures for when a normal shutdown happens
  // be aware that atexit() functions run in reverse order
  atexit(&logger::shutdown);
  atexit(&shutdown_handler);

#ifdef HAVE_AWS_S3
  omnisci_aws_sdk::init_sdk();
  ScopeGuard aws_sdk_guard = [] { omnisci_aws_sdk::shutdown_sdk(); };
#endif

  // start background thread to clean up _DELETE_ME files
  const unsigned int wait_interval =
      3;  // wait time in secs after looking for deleted file before looking again
  std::thread file_delete_thread(file_delete,
                                 std::ref(g_running),
                                 wait_interval,
                                 prog_config_opts.base_path + "/mapd_data");
  std::thread heartbeat_thread(heartbeat);

  if (!g_enable_thrift_logs) {
    apache::thrift::GlobalOutput.setOutputFunction([](const char* msg) {});
  }

  if (g_enable_experimental_string_functions) {
    // Use the locale setting of the server by default. The generate parameter can be
    // updated appropriately if a locale override option is ever supported.
    boost::locale::generator generator;
    std::locale::global(generator.generate(""));
  }

  try {
    if (prog_config_opts.system_parameters.master_address.empty()) {
      g_mapd_handler =
          mapd::make_shared<DBHandler>(prog_config_opts.db_leaves,
                                       prog_config_opts.string_leaves,
                                       prog_config_opts.base_path,
                                       prog_config_opts.allow_multifrag,
                                       prog_config_opts.jit_debug,
                                       prog_config_opts.intel_jit_profile,
                                       prog_config_opts.read_only,
                                       prog_config_opts.allow_loop_joins,
                                       prog_config_opts.enable_rendering,
                                       prog_config_opts.renderer_use_vulkan_driver,
                                       prog_config_opts.enable_auto_clear_render_mem,
                                       prog_config_opts.render_oom_retry_threshold,
                                       prog_config_opts.render_mem_bytes,
                                       prog_config_opts.max_concurrent_render_sessions,
                                       prog_config_opts.reserved_gpu_mem,
                                       prog_config_opts.render_compositor_use_last_gpu,
                                       prog_config_opts.num_reader_threads,
                                       prog_config_opts.authMetadata,
                                       prog_config_opts.system_parameters,
                                       prog_config_opts.enable_legacy_syntax,
                                       prog_config_opts.idle_session_duration,
                                       prog_config_opts.max_session_duration,
                                       prog_config_opts.enable_runtime_udf,
                                       prog_config_opts.udf_file_name,
                                       prog_config_opts.udf_compiler_path,
                                       prog_config_opts.udf_compiler_options,
#ifdef ENABLE_GEOS
                                       prog_config_opts.libgeos_so_filename,
#endif
                                       prog_config_opts.disk_cache_config,
                                       false);
    } else {  // running ha server
      LOG(FATAL)
          << "No High Availability module available, please contact OmniSci support";
    }
  } catch (const std::exception& e) {
    LOG(FATAL) << "Failed to initialize service handler: " << e.what();
  }

  if (g_enable_fsi) {
    foreign_storage::ForeignTableRefreshScheduler::start(g_running);
  }

  mapd::shared_ptr<TServerSocket> serverSocket;
  mapd::shared_ptr<TServerSocket> httpServerSocket;
  if (!prog_config_opts.system_parameters.ssl_cert_file.empty() &&
      !prog_config_opts.system_parameters.ssl_key_file.empty()) {
    mapd::shared_ptr<TSSLSocketFactory> sslSocketFactory;
    sslSocketFactory =
        mapd::shared_ptr<TSSLSocketFactory>(new TSSLSocketFactory(SSLProtocol::SSLTLS));
    sslSocketFactory->loadCertificate(
        prog_config_opts.system_parameters.ssl_cert_file.c_str());
    sslSocketFactory->loadPrivateKey(
        prog_config_opts.system_parameters.ssl_key_file.c_str());
    if (prog_config_opts.system_parameters.ssl_transport_client_auth) {
      sslSocketFactory->authenticate(true);
    } else {
      sslSocketFactory->authenticate(false);
    }
    sslSocketFactory->ciphers("ALL:!ADH:!LOW:!EXP:!MD5:@STRENGTH");
    serverSocket = mapd::shared_ptr<TServerSocket>(new TSSLServerSocket(
        prog_config_opts.system_parameters.omnisci_server_port, sslSocketFactory));
    httpServerSocket = mapd::shared_ptr<TServerSocket>(
        new TSSLServerSocket(prog_config_opts.http_port, sslSocketFactory));
    LOG(INFO) << " OmniSci server using encrypted connection. Cert file ["
              << prog_config_opts.system_parameters.ssl_cert_file << "], key file ["
              << prog_config_opts.system_parameters.ssl_key_file << "]";
  } else {
    LOG(INFO) << " OmniSci server using unencrypted connection";
    serverSocket = mapd::shared_ptr<TServerSocket>(
        new TServerSocket(prog_config_opts.system_parameters.omnisci_server_port));
    httpServerSocket =
        mapd::shared_ptr<TServerSocket>(new TServerSocket(prog_config_opts.http_port));
  }

  ScopeGuard pointer_to_thrift_guard = [] {
    mapd_lock_guard<mapd_shared_mutex> write_lock(g_thrift_mutex);
    g_thrift_buf_server = g_thrift_http_server = nullptr;
  };

  mapd::shared_ptr<TProcessor> processor(
      new TrackingProcessor(g_mapd_handler, prog_config_opts.log_user_origin));
  mapd::shared_ptr<TTransportFactory> bufTransportFactory(
      new TBufferedTransportFactory());
  mapd::shared_ptr<TProtocolFactory> bufProtocolFactory(new TBinaryProtocolFactory());

  mapd::shared_ptr<TServerTransport> bufServerTransport(serverSocket);
  TThreadedServer bufServer(
      processor, bufServerTransport, bufTransportFactory, bufProtocolFactory);
  {
    mapd_lock_guard<mapd_shared_mutex> write_lock(g_thrift_mutex);
    g_thrift_buf_server = &bufServer;
  }

  std::thread bufThread(start_server,
                        std::ref(bufServer),
                        prog_config_opts.system_parameters.omnisci_server_port);

  // TEMPORARY
  auto warmup_queries = [&prog_config_opts]() {
    // run warm up queries if any exists
    run_warmup_queries(
        g_mapd_handler, prog_config_opts.base_path, prog_config_opts.db_query_file);
    if (prog_config_opts.exit_after_warmup) {
      g_running = false;
    }
  };

  mapd::shared_ptr<TServerTransport> httpServerTransport(httpServerSocket);
  mapd::shared_ptr<TTransportFactory> httpTransportFactory(
      new THttpServerTransportFactory());
  mapd::shared_ptr<TProtocolFactory> httpProtocolFactory(new TJSONProtocolFactory());
  TThreadedServer httpServer(
      processor, httpServerTransport, httpTransportFactory, httpProtocolFactory);
  if (start_http_server) {
    {
      mapd_lock_guard<mapd_shared_mutex> write_lock(g_thrift_mutex);
      g_thrift_http_server = &httpServer;
    }
    std::thread httpThread(
        start_server, std::ref(httpServer), prog_config_opts.http_port);

    warmup_queries();

    bufThread.join();
    httpThread.join();
  } else {
    warmup_queries();
    bufThread.join();
  }

  g_running = false;
  file_delete_thread.join();
  heartbeat_thread.join();

  if (g_enable_fsi) {
    foreign_storage::ForeignTableRefreshScheduler::stop();
  }

  int signum = g_saw_signal;
  if (signum <= 0 || signum == SIGTERM) {
    return 0;
  } else {
    return signum;
  }
}

int main(int argc, char** argv) {
  bool has_clust_topo = false;

  CommandLineOptions prog_config_opts(argv[0], has_clust_topo);

  try {
    if (auto return_code =
            prog_config_opts.parse_command_line(argc, argv, !has_clust_topo)) {
      return *return_code;
    }

    if (!has_clust_topo) {
      prog_config_opts.validate_base_path();
      prog_config_opts.validate();
      return (startMapdServer(prog_config_opts));
    }
  } catch (std::runtime_error& e) {
    std::cerr << "Can't start: " << e.what() << std::endl;
    return 1;
  } catch (boost::program_options::error& e) {
    std::cerr << "Usage Error: " << e.what() << std::endl;
    return 1;
  }
}