nghttpx: Support per-backend mruby script

doc/nghttpx.h2r

@@ -299,9 +299,19 @@ server.  These hooks allows users to modify header fields, or common
 HTTP variables, like authority or request path, and even return custom
 response without forwarding request to backend servers.
 
-To specify mruby script file, use :option:`--mruby-file` option.  The
-script will be evaluated once per thread on startup, and it must
-instantiate object and evaluate it as the return value (e.g.,
+There are 2 levels of mruby script invocations: global and
+per-backend.  The global mruby script is set by :option:`--mruby-file`
+option and is called for all requests.  The per-backend mruby script
+is set by "mruby" parameter in :option:`-b` option.  It is invoked for
+a request which is forwarded to the particular backend.  The order of
+hook invocation is: global request phase hook, per-backend request
+phase hook, per-backend response phase hook, and finally global
+response phase hook.  If a hook returns a response, any later hooks
+are not invoked.  The global request hook is invoked before selecting
+backend, and changing request path may affect the backend selection.
+
+The all mruby script will be evaluated once per thread on startup, and
+it must instantiate object and evaluate it as the return value (e.g.,
 ``App.new``).  This object is called app object.  If app object
 defines ``on_req`` method, it is called with :rb:class:`Nghttpx::Env`
 object on request hook.  Similarly, if app object defines ``on_resp``

gennghttpxfun.py

@@ -169,6 +169,7 @@ OPTIONS = [
     "ocsp-startup",
     "no-verify-ocsp",
     "verify-client-tolerate-expired",
+    "ignore-per-backend-mruby-error",
 ]
 
 LOGVARS = [

src/shrpx.cc

@@ -1729,12 +1729,13 @@ Connections:
               The  parameters are  delimited  by  ";".  The  available
               parameters       are:      "proto=<PROTO>",       "tls",
               "sni=<SNI_HOST>",         "fall=<N>",        "rise=<N>",
-              "affinity=<METHOD>",  "dns", and  "redirect-if-not-tls".
-              The  parameter  consists   of  keyword,  and  optionally
-              followed by  "=" and value.  For  example, the parameter
-              "proto=h2"  consists of  the keyword  "proto" and  value
-              "h2".  The parameter "tls" consists of the keyword "tls"
-              without value.  Each parameter is described as follows.
+              "affinity=<METHOD>",    "dns",    "redirect-if-not-tls",
+              "upgrade-scheme",  and  "mruby=<PATH>".   The  parameter
+              consists of keyword, and  optionally followed by "=" and
+              value.  For  example, the parameter  "proto=h2" consists
+              of the  keyword "proto"  and value "h2".   The parameter
+              "tls" consists of the keyword "tls" without value.  Each
+              parameter is described as follows.
 
               The backend application protocol  can be specified using
               optional  "proto"   parameter,  and   in  the   form  of
@@ -1827,6 +1828,10 @@ Connections:
               server  which  requires  "https" :scheme  pseudo  header
               field on TLS encrypted connection.
 
+              "mruby=<PATH>"  parameter  specifies  a  path  to  mruby
+              script  file  which  is  invoked when  this  backend  is
+              selected.
+
               Since ";" and ":" are  used as delimiter, <PATTERN> must
               not  contain these  characters.  Since  ";" has  special
               meaning in shell, the option value must be quoted.
@@ -2749,6 +2754,10 @@ Process:
 Scripting:
   --mruby-file=<PATH>
               Set mruby script file
+  --ignore-per-backend-mruby-error
+              Ignore mruby compile error  for per-backend mruby script
+              file.  If error  occurred, it is treated as  if no mruby
+              file were specified for the backend.
 
 Misc:
   --conf=<PATH>
@@ -3424,6 +3433,8 @@ int main(int argc, char **argv) {
         {SHRPX_OPT_SINGLE_PROCESS.c_str(), no_argument, &flag, 159},
         {SHRPX_OPT_VERIFY_CLIENT_TOLERATE_EXPIRED.c_str(), no_argument, &flag,
          160},
+        {SHRPX_OPT_IGNORE_PER_BACKEND_MRUBY_ERROR.c_str(), no_argument, &flag,
+         161},
         {nullptr, 0, nullptr, 0}};
 
     int option_index = 0;
@@ -4190,6 +4201,11 @@ int main(int argc, char **argv) {
         cmdcfgs.emplace_back(SHRPX_OPT_VERIFY_CLIENT_TOLERATE_EXPIRED,
                              StringRef::from_lit("yes"));
         break;
+      case 161:
+        // --ignore-per-backend-mruby-error
+        cmdcfgs.emplace_back(SHRPX_OPT_IGNORE_PER_BACKEND_MRUBY_ERROR,
+                             StringRef::from_lit("yes"));
+        break;
       default:
         break;
       }

src/shrpx_config.cc

@@ -55,6 +55,9 @@
 #include "shrpx_log.h"
 #include "shrpx_tls.h"
 #include "shrpx_http.h"
+#ifdef HAVE_MRUBY
+#  include "shrpx_mruby.h"
+#endif // HAVE_MRUBY
 #include "util.h"
 #include "base64.h"
 #include "ssl_compat.h"
@@ -807,6 +810,7 @@ int parse_upstream_params(UpstreamParams &out, const StringRef &src_params) {
 
 struct DownstreamParams {
   StringRef sni;
+  StringRef mruby;
   AffinityConfig affinity;
   size_t fall;
   size_t rise;
@@ -921,6 +925,9 @@ int parse_downstream_params(DownstreamParams &out,
       out.redirect_if_not_tls = true;
     } else if (util::strieq_l("upgrade-scheme", param)) {
       out.upgrade_scheme = true;
+    } else if (util::istarts_with_l(param, "mruby=")) {
+      auto valstr = StringRef{first + str_size("mruby="), end};
+      out.mruby = valstr;
     } else if (!param.empty()) {
       LOG(ERROR) << "backend: " << param << ": unknown keyword";
       return -1;
@@ -1045,6 +1052,18 @@ int parse_mapping(Config *config, DownstreamAddrConfig &addr,
       if (params.redirect_if_not_tls) {
         g.redirect_if_not_tls = true;
       }
+      // All backends in the same group must have the same mruby path.
+      // If some backend does not specify mruby file, and there is at
+      // least one backend with mruby file, it is used for all backend
+      // in the group.
+      if (g.mruby_file.empty()) {
+        g.mruby_file = params.mruby;
+      } else if (g.mruby_file != params.mruby) {
+        LOG(ERROR) << "backend: mruby: multiple different mruby file found in "
+                      "a single group";
+        return -1;
+      }
+
       g.addrs.push_back(addr);
       continue;
     }
@@ -1065,6 +1084,7 @@ int parse_mapping(Config *config, DownstreamAddrConfig &addr,
       g.affinity.cookie.secure = params.affinity.cookie.secure;
     }
     g.redirect_if_not_tls = params.redirect_if_not_tls;
+    g.mruby_file = params.mruby;
 
     if (pattern[0] == '*') {
       // wildcard pattern
@@ -2179,6 +2199,9 @@ int option_lookup_token(const char *name, size_t namelen) {
       }
       break;
     case 'r':
+      if (util::strieq_l("ignore-per-backend-mruby-erro", name, 29)) {
+        return SHRPX_OPTID_IGNORE_PER_BACKEND_MRUBY_ERROR;
+      }
       if (util::strieq_l("strip-incoming-x-forwarded-fo", name, 29)) {
         return SHRPX_OPTID_STRIP_INCOMING_X_FORWARDED_FOR;
       }
@@ -3563,6 +3586,10 @@ int parse_config(Config *config, int optid, const StringRef &opt,
   case SHRPX_OPTID_VERIFY_CLIENT_TOLERATE_EXPIRED:
     config->tls.client_verify.tolerate_expired = util::strieq_l("yes", optarg);
 
+    return 0;
+  case SHRPX_OPTID_IGNORE_PER_BACKEND_MRUBY_ERROR:
+    config->ignore_per_backend_mruby_error = util::strieq_l("yes", optarg);
+
     return 0;
   case SHRPX_OPTID_CONF:
     LOG(WARN) << "conf: ignored";
@@ -3854,7 +3881,32 @@ int configure_downstream_group(Config *config, bool http2_proxy,
                   << (addr.tls ? ", tls" : "");
       }
     }
+#ifdef HAVE_MRUBY
+    // Try compile mruby script and catch compile error early.
+    if (!g.mruby_file.empty()) {
+      if (mruby::create_mruby_context(g.mruby_file) == nullptr) {
+        LOG(config->ignore_per_backend_mruby_error ? ERROR : FATAL)
+            << "backend: Could not compile mruby flie for pattern "
+            << g.pattern;
+        if (!config->ignore_per_backend_mruby_error) {
+          return -1;
+        }
+        g.mruby_file = StringRef{};
+      }
+    }
+#endif // HAVE_MRUBY
+  }
+
+#ifdef HAVE_MRUBY
+  // Try compile mruby script (--mruby-file) here to catch compile
+  // error early.
+  if (!config->mruby_file.empty()) {
+    if (mruby::create_mruby_context(config->mruby_file) == nullptr) {
+      LOG(FATAL) << "mruby-file: Could not compile mruby file";
+      return -1;
+    }
   }
+#endif // HAVE_MRUBY
 
   if (catch_all_group == -1) {
     LOG(FATAL) << "backend: No catch-all backend address is configured";

src/shrpx_config.h

@@ -345,6 +345,8 @@ constexpr auto SHRPX_OPT_OCSP_STARTUP = StringRef::from_lit("ocsp-startup");
 constexpr auto SHRPX_OPT_NO_VERIFY_OCSP = StringRef::from_lit("no-verify-ocsp");
 constexpr auto SHRPX_OPT_VERIFY_CLIENT_TOLERATE_EXPIRED =
     StringRef::from_lit("verify-client-tolerate-expired");
+constexpr auto SHRPX_OPT_IGNORE_PER_BACKEND_MRUBY_ERROR =
+    StringRef::from_lit("ignore-per-backend-mruby-error");
 
 constexpr size_t SHRPX_OBFUSCATED_NODE_LENGTH = 8;
 
@@ -483,6 +485,7 @@ struct DownstreamAddrGroupConfig {
       : pattern(pattern), affinity{AFFINITY_NONE}, redirect_if_not_tls(false) {}
 
   StringRef pattern;
+  StringRef mruby_file;
   std::vector<DownstreamAddrConfig> addrs;
   // Bunch of session affinity hash.  Only used if affinity ==
   // AFFINITY_IP.
@@ -915,6 +918,7 @@ struct Config {
         http2_proxy{false},
         single_process{false},
         single_thread{false},
+        ignore_per_backend_mruby_error{false},
         ev_loop_flags{0} {}
   ~Config();
 
@@ -959,6 +963,8 @@ struct Config {
   // handling is omitted.
   bool single_process;
   bool single_thread;
+  // Ignore mruby compile error for per-backend mruby script.
+  bool ignore_per_backend_mruby_error;
   // flags passed to ev_default_loop() and ev_loop_new()
   int ev_loop_flags;
 };
@@ -1063,6 +1069,7 @@ enum {
   SHRPX_OPTID_HTTP2_MAX_CONCURRENT_STREAMS,
   SHRPX_OPTID_HTTP2_NO_COOKIE_CRUMBLING,
   SHRPX_OPTID_HTTP2_PROXY,
+  SHRPX_OPTID_IGNORE_PER_BACKEND_MRUBY_ERROR,
   SHRPX_OPTID_INCLUDE,
   SHRPX_OPTID_INSECURE,
   SHRPX_OPTID_LISTENER_DISABLE_TIMEOUT,

src/shrpx_downstream.cc

@@ -188,6 +188,14 @@ Downstream::~Downstream() {
 #endif // HAVE_MRUBY
   }
 
+#ifdef HAVE_MRUBY
+  if (dconn_) {
+    const auto &group = dconn_->get_downstream_addr_group();
+    const auto &mruby_ctx = group->mruby_ctx;
+    mruby_ctx->delete_downstream(this);
+  }
+#endif // HAVE_MRUBY
+
   // DownstreamConnection may refer to this object.  Delete it now
   // explicitly.
   dconn_.reset();
@@ -217,6 +225,12 @@ void Downstream::detach_downstream_connection() {
     return;
   }
 
+#ifdef HAVE_MRUBY
+  const auto &group = dconn_->get_downstream_addr_group();
+  const auto &mruby_ctx = group->mruby_ctx;
+  mruby_ctx->delete_downstream(this);
+#endif // HAVE_MRUBY
+
   dconn_->detach_downstream(this);
 
   auto handler = dconn_->get_client_handler();
@@ -230,6 +244,16 @@ DownstreamConnection *Downstream::get_downstream_connection() {
 }
 
 std::unique_ptr<DownstreamConnection> Downstream::pop_downstream_connection() {
+#ifdef HAVE_MRUBY
+  if (!dconn_) {
+    return nullptr;
+  }
+
+  const auto &group = dconn_->get_downstream_addr_group();
+  const auto &mruby_ctx = group->mruby_ctx;
+  mruby_ctx->delete_downstream(this);
+#endif // HAVE_MRUBY
+
   return std::unique_ptr<DownstreamConnection>(dconn_.release());
 }
 

src/shrpx_http2_upstream.cc

@@ -461,6 +461,9 @@ void Http2Upstream::initiate_downstream(Downstream *downstream) {
     return;
   }
 
+#ifdef HAVE_MRUBY
+  auto dconn_ptr = dconn.get();
+#endif // HAVE_MRUBY
   rv = downstream->attach_downstream_connection(std::move(dconn));
   if (rv != 0) {
     // downstream connection fails, send error page
@@ -474,6 +477,25 @@ void Http2Upstream::initiate_downstream(Downstream *downstream) {
 
     return;
   }
+
+#ifdef HAVE_MRUBY
+  const auto &group = dconn_ptr->get_downstream_addr_group();
+  const auto &mruby_ctx = group->mruby_ctx;
+  if (mruby_ctx->run_on_request_proc(downstream) != 0) {
+    if (error_reply(downstream, 500) != 0) {
+      rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
+    }
+
+    downstream_queue_.mark_failure(downstream);
+
+    return;
+  }
+
+  if (downstream->get_response_state() == Downstream::MSG_COMPLETE) {
+    return;
+  }
+#endif // HAVE_MRUBY
+
   rv = downstream->push_request_headers();
   if (rv != 0) {
 
@@ -1611,6 +1633,22 @@ int Http2Upstream::on_downstream_header_complete(Downstream *downstream) {
 
 #ifdef HAVE_MRUBY
   if (!downstream->get_non_final_response()) {
+    auto dconn = downstream->get_downstream_connection();
+    const auto &group = dconn->get_downstream_addr_group();
+    const auto &dmruby_ctx = group->mruby_ctx;
+
+    if (dmruby_ctx->run_on_response_proc(downstream) != 0) {
+      if (error_reply(downstream, 500) != 0) {
+        return -1;
+      }
+      // Returning -1 will signal deletion of dconn.
+      return -1;
+    }
+
+    if (downstream->get_response_state() == Downstream::MSG_COMPLETE) {
+      return -1;
+    }
+
     auto worker = handler_->get_worker();
     auto mruby_ctx = worker->get_mruby_context();
 

src/shrpx_https_upstream.cc

@@ -431,12 +431,29 @@ int htp_hdrs_completecb(http_parser *htp) {
     return -1;
   }
 
+#ifdef HAVE_MRUBY
+  auto dconn_ptr = dconn.get();
+#endif // HAVE_MRUBY
   if (downstream->attach_downstream_connection(std::move(dconn)) != 0) {
     downstream->set_request_state(Downstream::CONNECT_FAIL);
 
     return -1;
   }
 
+#ifdef HAVE_MRUBY
+  const auto &group = dconn_ptr->get_downstream_addr_group();
+  const auto &dmruby_ctx = group->mruby_ctx;
+
+  if (dmruby_ctx->run_on_request_proc(downstream) != 0) {
+    resp.http_status = 500;
+    return -1;
+  }
+
+  if (downstream->get_response_state() == Downstream::MSG_COMPLETE) {
+    return 0;
+  }
+#endif // HAVE_MRUBY
+
   rv = downstream->push_request_headers();
 
   if (rv != 0) {
@@ -1021,6 +1038,8 @@ int HttpsUpstream::on_downstream_header_complete(Downstream *downstream) {
   const auto &req = downstream->request();
   auto &resp = downstream->response();
   auto &balloc = downstream->get_block_allocator();
+  auto dconn = downstream->get_downstream_connection();
+  assert(dconn);
 
   if (downstream->get_non_final_response() &&
       !downstream->supports_non_final_response()) {
@@ -1030,6 +1049,18 @@ int HttpsUpstream::on_downstream_header_complete(Downstream *downstream) {
 
 #ifdef HAVE_MRUBY
   if (!downstream->get_non_final_response()) {
+    const auto &group = dconn->get_downstream_addr_group();
+    const auto &dmruby_ctx = group->mruby_ctx;
+
+    if (dmruby_ctx->run_on_response_proc(downstream) != 0) {
+      error_reply(500);
+      return -1;
+    }
+
+    if (downstream->get_response_state() == Downstream::MSG_COMPLETE) {
+      return -1;
+    }
+
     auto worker = handler_->get_worker();
     auto mruby_ctx = worker->get_mruby_context();
 
@@ -1150,8 +1181,6 @@ int HttpsUpstream::on_downstream_header_complete(Downstream *downstream) {
   if (req.method != HTTP_CONNECT || !downstream->get_upgraded()) {
     auto affinity_cookie = downstream->get_affinity_cookie_to_send();
     if (affinity_cookie) {
-      auto dconn = downstream->get_downstream_connection();
-      assert(dconn);
       auto &group = dconn->get_downstream_addr_group();
       auto &shared_addr = group->shared_addr;
       auto &cookieconf = shared_addr->affinity.cookie;

src/shrpx_worker.cc

@@ -68,6 +68,10 @@ void proc_wev_cb(struct ev_loop *loop, ev_timer *w, int revents) {
 }
 } // namespace
 
+DownstreamAddrGroup::DownstreamAddrGroup() : retired{false} {}
+
+DownstreamAddrGroup::~DownstreamAddrGroup() {}
+
 // DownstreamKey is used to index SharedDownstreamAddr in order to
 // find the same configuration.
 using DownstreamKey = std::tuple<
@@ -185,6 +189,10 @@ void Worker::replace_downstream_config(
     dst = std::make_shared<DownstreamAddrGroup>();
     dst->pattern =
         ImmutableString{std::begin(src.pattern), std::end(src.pattern)};
+#ifdef HAVE_MRUBY
+    dst->mruby_ctx = mruby::create_mruby_context(src.mruby_file);
+    assert(dst->mruby_ctx);
+#endif // HAVE_MRUBY
 
     auto shared_addr = std::make_shared<SharedDownstreamAddr>();
 

src/shrpx_worker.h

@@ -183,7 +183,8 @@ struct SharedDownstreamAddr {
 };
 
 struct DownstreamAddrGroup {
-  DownstreamAddrGroup() : retired{false} {};
+  DownstreamAddrGroup();
+  ~DownstreamAddrGroup();
 
   DownstreamAddrGroup(const DownstreamAddrGroup &) = delete;
   DownstreamAddrGroup(DownstreamAddrGroup &&) = delete;
@@ -192,6 +193,9 @@ struct DownstreamAddrGroup {
 
   ImmutableString pattern;
   std::shared_ptr<SharedDownstreamAddr> shared_addr;
+#ifdef HAVE_MRUBY
+  std::unique_ptr<mruby::MRubyContext> mruby_ctx;
+#endif // HAVE_MRUBY
   // true if this group is no longer used for new request.  If this is
   // true, the connection made using one of address in shared_addr
   // must not be pooled.