proxy: per-backend automatic logging

Options similar to mcp.log_reqsample() are applied to responses as
they're read off the network.

Can be a lot faster than relying on callbacks to do logging, if you're
not doing logging inline with a route handler. Also simpler in some
cases (ie routelib).

Author: dormando

proto_proxy.c

@@ -1196,8 +1196,6 @@ mcp_resp_t *mcp_prep_bare_resobj(lua_State *L, LIBEVENT_THREAD *t) {
 
 void mcp_set_resobj(mcp_resp_t *r, mcp_request_t *rq, mcp_backend_t *be, LIBEVENT_THREAD *t) {
     memset(r, 0, sizeof(mcp_resp_t));
-    r->buf = NULL;
-    r->blen = 0;
     r->thread = t;
     assert(r->thread != NULL);
     gettimeofday(&r->start, NULL);
@@ -1224,20 +1222,7 @@ void mcp_set_resobj(mcp_resp_t *r, mcp_request_t *rq, mcp_backend_t *be, LIBEVEN
     }
 
     r->cmd = rq->pr.command;
-
-    strncpy(r->be_name, be->name, MAX_NAMELEN+1);
-    strncpy(r->be_port, be->port, MAX_PORTLEN+1);
-
-}
-
-mcp_resp_t *mcp_prep_resobj(lua_State *L, mcp_request_t *rq, mcp_backend_t *be, LIBEVENT_THREAD *t) {
-    mcp_resp_t *r = lua_newuserdatauv(L, sizeof(mcp_resp_t), 0);
-    mcp_set_resobj(r, rq, be, t);
-
-    luaL_getmetatable(L, "mcp.response");
-    lua_setmetatable(L, -2);
-
-    return r;
+    r->be = be;
 }
 
 void mcp_resp_set_elapsed(mcp_resp_t *r) {

proxy.h

@@ -227,6 +227,15 @@ struct proxy_tunables {
     bool down; // backend is forced into a down/bad state.
 };
 
+struct proxy_logging {
+    unsigned int deadline; // log if slower than N us (user specifies ms)
+    unsigned int rate; // sampling
+    bool all_errors; // always log on an error case
+    char *detail_hit;
+    char *detail_ok;
+    char *detail_err;
+};
+
 typedef STAILQ_HEAD(globalobj_head_s, mcp_globalobj_s) globalobj_head_t;
 typedef struct {
     lua_State *proxy_state; // main configuration vm
@@ -382,7 +391,9 @@ struct mcp_backend_label_s {
     char label[MAX_LABELLEN+1];
     size_t llen; // cache label length for small speedup in pool creation.
     int conncount; // number of sockets to make.
+    bool use_logging;
     struct proxy_tunables tunables;
+    struct proxy_logging logging;
 };
 
 // lua object wrapper meant to own a malloc'ed conn structure
@@ -435,13 +446,15 @@ struct mcp_backend_s {
     bool transferred; // if beconn has been shipped to owner thread.
     bool use_io_thread; // note if this backend is worker-local or not.
     bool stacked; // if backend already queued for syscalls.
+    bool use_logging; // if automatic logging is enabled.
     STAILQ_ENTRY(mcp_backend_s) beconn_next; // stack for connecting conns
     STAILQ_ENTRY(mcp_backend_s) be_next; // stack for backends
     iop_head_t iop_head; // stack of inbound requests.
+    struct proxy_logging logging;
+    struct proxy_tunables tunables; // this gets copied a few times for speed.
     char name[MAX_NAMELEN+1];
     char port[MAX_PORTLEN+1];
     char label[MAX_LABELLEN+1];
-    struct proxy_tunables tunables; // this gets copied a few times for speed.
     struct mcp_backendconn_s be[];
 };
 typedef STAILQ_HEAD(be_head_s, mcp_backend_s) be_head_t;
@@ -494,6 +507,7 @@ typedef struct {
     char *buf; // response line + potentially value.
     mc_resp *cresp; // client mc_resp object during extstore fetches.
     LIBEVENT_THREAD *thread; // cresp's owner thread needed for extstore cleanup.
+    mcp_backend_t *be; // backend that generated this response
     unsigned int blen; // total size of the value to read.
     struct timeval start; // time this object was created.
     long elapsed; // time elapsed once handled.
@@ -502,8 +516,6 @@ typedef struct {
     uint8_t cmd; // from parser (pr.command)
     uint8_t extra; // ascii multiget hack for memory accounting. extra blen.
     enum mcp_resp_mode mode; // reply mode (for noreply fixing)
-    char be_name[MAX_NAMELEN+1];
-    char be_port[MAX_PORTLEN+1];
 } mcp_resp_t;
 
 // re-cast an io_pending_t into this more descriptive structure.
@@ -600,7 +612,6 @@ void proxy_init_event_thread(proxy_event_thread_t *t, proxy_ctx_t *ctx, struct e
 void *proxy_event_thread(void *arg);
 void proxy_run_backend_queue(be_head_t *head);
 struct mcp_backendconn_s *proxy_choose_beconn(mcp_backend_t *be);
-mcp_resp_t *mcp_prep_resobj(lua_State *L, mcp_request_t *rq, mcp_backend_t *be, LIBEVENT_THREAD *t);
 mcp_resp_t *mcp_prep_bare_resobj(lua_State *L, LIBEVENT_THREAD *t);
 void mcp_resp_set_elapsed(mcp_resp_t *r);
 io_pending_proxy_t *mcp_queue_rctx_io(mcp_rcontext_t *rctx, mcp_request_t *rq, mcp_backend_t *be, mcp_resp_t *r);
@@ -787,7 +798,7 @@ int mcplib_funcgen_gc(lua_State *L);
 void mcp_funcgen_reference(lua_State *L);
 void mcp_funcgen_dereference(lua_State *L, mcp_funcgen_t *fgen);
 void mcp_rcontext_push_rqu_res(lua_State *L, mcp_rcontext_t *rctx, int handle);
-
+void mcplib_rqu_log(mcp_request_t *rq, mcp_resp_t *rs, int cfd);
 
 int mcplib_factory_command_new(lua_State *L);
 

proxy_internal.c

@@ -1719,8 +1719,9 @@ static inline int _mcplib_internal_run(LIBEVENT_THREAD *t, mcp_request_t *rq, mc
     }
 
     // in case someone logs this response it should make sense.
-    memcpy(r->be_name, "internal", strlen("internal"));
-    memcpy(r->be_port, "0", 1);
+    // FIXME: make a decision here.
+    //memcpy(r->be_name, "internal", strlen("internal"));
+    //memcpy(r->be_port, "0", 1);
 
     // TODO: r-> will need status/code/mode copied from resp.
     r->cresp = resp;

proxy_lua.c

@@ -285,7 +285,78 @@ static int mcplib_backend_wrap_gc(lua_State *L) {
 }
 
 static int mcplib_backend_gc(lua_State *L) {
-    return 0; // no-op.
+    mcp_backend_label_t *be = lua_touserdata(L, 1);
+    if (be->logging.detail_hit)
+        free(be->logging.detail_hit);
+    if (be->logging.detail_ok)
+        free(be->logging.detail_ok);
+    if (be->logging.detail_err)
+        free(be->logging.detail_err);
+
+    return 0;
+}
+
+static int _mcplib_backend_log(lua_State *L, mcp_backend_label_t *be) {
+    be->use_logging = true;
+
+    if (lua_getfield(L, -1, "deadline") != LUA_TNIL) {
+        int deadline = luaL_checkinteger(L, -1);
+        if (deadline < 0) {
+            proxy_lua_error(L, "backend log deadline must be >= 0");
+        }
+        // convert to milliseconds.
+        be->logging.deadline = deadline * 1000;
+    }
+    lua_pop(L, 1);
+
+    if (lua_getfield(L, -1, "rate") != LUA_TNIL) {
+        int rate = luaL_checkinteger(L, -1);
+        if (rate < 0) {
+            proxy_lua_error(L, "backend log sample rate must be >= 0");
+        }
+        be->logging.rate = rate;
+    }
+    lua_pop(L, 1);
+
+    if (lua_getfield(L, -1, "errors") != LUA_TNIL) {
+        luaL_checktype(L, -1, LUA_TBOOLEAN);
+        int errors = lua_toboolean(L, -1);
+        if (errors) {
+            be->logging.all_errors = true;
+        } else {
+            be->logging.all_errors = false;
+        }
+    }
+    lua_pop(L, 1);
+
+    if (lua_getfield(L, -1, "tag_hit") != LUA_TNIL) {
+        size_t tlen = 0;
+        const char *tag = luaL_checklstring(L, -1, &tlen);
+        be->logging.detail_hit = malloc(tlen+1);
+        memcpy(be->logging.detail_hit, tag, tlen);
+        be->logging.detail_hit[tlen] = '\0';
+    }
+    lua_pop(L, 1);
+
+    if (lua_getfield(L, -1, "tag_ok") != LUA_TNIL) {
+        size_t tlen = 0;
+        const char *tag = luaL_checklstring(L, -1, &tlen);
+        be->logging.detail_ok = malloc(tlen+1);
+        memcpy(be->logging.detail_ok, tag, tlen);
+        be->logging.detail_ok[tlen] = '\0';
+    }
+    lua_pop(L, 1);
+
+    if (lua_getfield(L, -1, "tag_err") != LUA_TNIL) {
+        size_t tlen = 0;
+        const char *tag = luaL_checklstring(L, -1, &tlen);
+        be->logging.detail_err = malloc(tlen+1);
+        memcpy(be->logging.detail_err, tag, tlen);
+        be->logging.detail_err[tlen] = '\0';
+    }
+    lua_pop(L, 1);
+
+    return 0;
 }
 
 // backend label object; given to pools which then find or create backend
@@ -298,15 +369,18 @@ static int mcplib_backend(lua_State *L) {
     size_t llen = 0;
     size_t nlen = 0;
     size_t plen = 0;
-    proxy_ctx_t *ctx = PROXY_GET_CTX(L);
-    mcp_backend_label_t *be = lua_newuserdatauv(L, sizeof(mcp_backend_label_t), 0);
-    memset(be, 0, sizeof(*be));
     const char *label;
     const char *name;
     const char *port;
+    proxy_ctx_t *ctx = PROXY_GET_CTX(L);
+    mcp_backend_label_t *be = lua_newuserdatauv(L, sizeof(mcp_backend_label_t), 0);
+    memset(be, 0, sizeof(*be));
     // copy global defaults for tunables.
     memcpy(&be->tunables, &ctx->tunables, sizeof(be->tunables));
     be->conncount = 1; // one connection per backend as default.
+    // set the metatable early so the GC handler can free partial allocations
+    luaL_getmetatable(L, "mcp.backend");
+    lua_setmetatable(L, -2); // set metatable to userdata.
 
     if (lua_istable(L, 1)) {
 
@@ -455,6 +529,14 @@ static int mcplib_backend(lua_State *L) {
         }
         lua_pop(L, 1);
 
+        if (lua_getfield(L, 1, "log") != LUA_TNIL) {
+            if (lua_istable(L, -1)) {
+                _mcplib_backend_log(L, be);
+            } else {
+                proxy_lua_error(L, "backend log option must be a table");
+            }
+        }
+        lua_pop(L, 1);
     } else {
         label = luaL_checklstring(L, 1, &llen);
         name = luaL_checklstring(L, 2, &nlen);
@@ -486,8 +568,6 @@ static int mcplib_backend(lua_State *L) {
     if (lua_istable(L, 1)) {
         lua_pop(L, 3); // drop label, name, port.
     }
-    luaL_getmetatable(L, "mcp.backend");
-    lua_setmetatable(L, -2); // set metatable to userdata.
 
     return 1; // return be object.
 }
@@ -530,12 +610,27 @@ static mcp_backend_wrap_t *_mcplib_make_backendconn(lua_State *L, mcp_backend_la
         proxy_lua_error(L, "out of memory allocating backend connection");
         return NULL;
     }
+
     bew->be = be;
 
     strncpy(be->name, bel->name, MAX_NAMELEN+1);
     strncpy(be->port, bel->port, MAX_PORTLEN+1);
     strncpy(be->label, bel->label, MAX_LABELLEN+1);
     memcpy(&be->tunables, &bel->tunables, sizeof(bel->tunables));
+    memcpy(&be->logging, &bel->logging, sizeof(bel->logging));
+    be->use_logging = bel->use_logging;
+    // TODO: check for errors.
+    // not really going to happen and if it does the tag just blanks out..
+    if (bel->logging.detail_hit) {
+        be->logging.detail_hit = strdup(bel->logging.detail_hit);
+    }
+    if (bel->logging.detail_ok) {
+        be->logging.detail_ok = strdup(bel->logging.detail_ok);
+    }
+    if (bel->logging.detail_err) {
+        be->logging.detail_err = strdup(bel->logging.detail_err);
+    }
+
     be->conncount = bel->conncount;
     STAILQ_INIT(&be->iop_head);
 
@@ -1397,8 +1492,8 @@ static int mcplib_log_req(lua_State *L) {
         rtype = rs->resp.type;
         rcode = rs->resp.code;
         rstatus = rs->status;
-        rname = rs->be_name;
-        rport = rs->be_port;
+        rname = rs->be->name;
+        rport = rs->be->port;
         elapsed = rs->elapsed;
     }
     size_t dlen = 0;
@@ -1432,6 +1527,57 @@ static uint32_t _mcp_nextrand(uint32_t *s) {
     return result;
 }
 
+void mcplib_rqu_log(mcp_request_t *rq, mcp_resp_t *rs, int cfd) {
+    LIBEVENT_THREAD *t = rs->thread;
+    logger *l = t->l;
+
+    long elapsed = 0;
+
+    int rtype = rs->resp.type;
+    int rcode = rs->resp.code;
+    int rstatus = rs->status;
+    elapsed = rs->elapsed;
+
+    bool do_log = false;
+    struct proxy_logging *pl = &rs->be->logging;
+    if (pl->all_errors && rstatus != MCMC_OK) {
+        do_log = true;
+    } else if (pl->deadline > 0 && elapsed > pl->deadline) {
+        do_log = true;
+    } else if (pl->rate > 0) {
+        // slightly biased random-to-rate without adding a loop, which is
+        // completely fine for this use case.
+        uint32_t rnd = (uint64_t)_mcp_nextrand(t->proxy_rng) * (uint64_t)pl->rate >> 32;
+        if (rnd == 0) {
+            do_log = true;
+        }
+    }
+
+    if (do_log) {
+        char *rname = rs->be->name;
+        char *rport = rs->be->port;
+        size_t dlen = 0;
+        const char *detail = NULL;
+
+        // FIXME: confirm this is what we want.
+        // I think this returns HIT in odd circumstances.
+        // might have to check for a "get class" cmd.
+        if (rstatus == MCMC_OK) {
+            if (rcode != MCMC_CODE_END) {
+                detail = rs->be->logging.detail_hit;
+            } else {
+                detail = rs->be->logging.detail_ok;
+            }
+        } else {
+            detail = rs->be->logging.detail_err;
+        }
+        if (detail) {
+            dlen = strlen(detail);
+        }
+
+        logger_log(l, LOGGER_PROXY_REQ, NULL, rq->pr.request, rq->pr.reqlen, elapsed, rtype, rcode, rstatus, cfd, detail, dlen, rname, rport);
+    }
+}
 
 // (milliseconds, sample_rate, allerrors, request, resp, "detail")
 static int mcplib_log_reqsample(lua_State *L) {
@@ -1459,8 +1605,8 @@ static int mcplib_log_reqsample(lua_State *L) {
         rtype = rs->resp.type;
         rcode = rs->resp.code;
         rstatus = rs->status;
-        rname = rs->be_name;
-        rport = rs->be_port;
+        rname = rs->be->name;
+        rport = rs->be->port;
         elapsed = rs->elapsed;
     }
     size_t dlen = 0;

proxy_luafgen.c

@@ -1084,6 +1084,10 @@ int mcp_process_rqueue_return(mcp_rcontext_t *rctx, int handle, mcp_resp_t *res)
         }
     }
 
+    if (res->be && res->be->use_logging) {
+        mcplib_rqu_log(rqu->rq, res, rctx->conn_fd);
+    }
+
     if (rqu->cb_ref) {
         lua_settop(rctx->Lc, 0);
         lua_rawgeti(rctx->Lc, LUA_REGISTRYINDEX, rqu->cb_ref);
@@ -1117,6 +1121,7 @@ int mcp_process_rqueue_return(mcp_rcontext_t *rctx, int handle, mcp_resp_t *res)
                                  // we settop _before_ calling cb's and
                                  // _before_ setting up for a coro resume.
     }
+
     rqu->flags |= flag;
     return rqu->flags;
 }
@@ -1150,10 +1155,7 @@ void mcp_run_rcontext_handle(mcp_rcontext_t *rctx, int handle) {
         if (rqu->obj_type == RQUEUE_TYPE_POOL) {
             mcp_request_t *rq = rqu->rq;
             mcp_backend_t *be = mcplib_pool_proxy_call_helper(rqu->obj, MCP_PARSER_KEY(rq->pr), rq->pr.klen);
-            // FIXME: queue requires conn because we're stacking objects
-            // into the connection for later submission, which means we
-            // absolutely cannot queue things once *c becomes invalid.
-            // need to assert/block this from happening.
+
             mcp_set_resobj(rqu->res_obj, rq, be, rctx->fgen->thread);
             io_pending_proxy_t *p = mcp_queue_rctx_io(rctx, rq, be, rqu->res_obj);
             p->return_cb = proxy_return_rqu_cb;