embedded memcached proxy

BUILD

@@ -1,3 +1,5 @@
+See below if building the proxy
+
 To build memcached in your machine from local repo you will have to install
 autotools, automake and libevent. In a debian based system that will look
 like this
@@ -22,3 +24,14 @@ You can telnet into that memcached to ensure it is up and running
 
 telnet 127.0.0.1 11211
 stats
+
+IF BUILDING PROXY, AN EXTRA STEP IS NECESSARY:
+
+cd memcached
+cd vendor
+./fetch.sh
+cd ..
+./autogen.sh
+./configure --enable-proxy
+make
+make test

Makefile.am

@@ -51,6 +51,10 @@ if ENABLE_SASL
 memcached_SOURCES += sasl_defs.c
 endif
 
+if ENABLE_PROXY
+memcached_SOURCES += proto_proxy.c proto_proxy.h vendor/mcmc/mcmc.h
+endif
+
 if ENABLE_EXTSTORE
 memcached_SOURCES += extstore.c extstore.h \
                      crc32c.c crc32c.h \
@@ -90,6 +94,16 @@ memcached_debug_DEPENDENCIES += memcached_debug_dtrace.o
 CLEANFILES += memcached_dtrace.o memcached_debug_dtrace.o
 endif
 
+if ENABLE_PROXY
+memcached_LDADD += vendor/lua/src/liblua.a vendor/mcmc/mcmc.o
+memcached_debug_LDADD += vendor/lua/src/liblua.a vendor/mcmc/mcmc.o
+endif
+
+if ENABLE_PROXY_URING
+memcached_LDADD += vendor/liburing/src/liburing.a
+memcached_debug_LDADD += vendor/liburing/src/liburing.a
+endif
+
 memcached_debug_CFLAGS += -DMEMCACHED_DEBUG
 
 memcached_dtrace.h: memcached_dtrace.d
@@ -108,6 +122,11 @@ memcached_debug_dtrace.o: $(memcached_debug_OBJECTS)
 SUBDIRS = doc
 DIST_DIRS = scripts
 EXTRA_DIST = doc scripts t memcached.spec memcached_dtrace.d version.m4 README.md LICENSE.bipbuffer
+EXTRA_DIST += vendor/Makefile vendor/lua vendor/mcmc
+
+if ENABLE_PROXY
+SUBDIRS += vendor
+endif
 
 MOSTLYCLEANFILES = *.gcov *.gcno *.gcda *.tcov
 

configure.ac

@@ -114,6 +114,12 @@ AC_ARG_ENABLE(static,
 AC_ARG_ENABLE(unix_socket,
   [AS_HELP_STRING([--disable-unix-socket], [Disable unix domain socket])])
 
+AC_ARG_ENABLE(proxy,
+  [AS_HELP_STRING([--enable-proxy], [Enable proxy code EXPERIMENTAL])])
+
+AC_ARG_ENABLE(proxy-uring,
+  [AS_HELP_STRING([--enable-proxy-uring], [Enable proxy io_uring code EXPERIMENTAL])])
+
 dnl **********************************************************************
 dnl DETECT_SASL_CB_GETCONF
 dnl
@@ -217,6 +223,18 @@ if test "x$enable_unix_socket" = "xno"; then
     AC_DEFINE([DISABLE_UNIX_SOCKET],1,[Set to nonzero if you want to disable unix domain socket])
 fi
 
+if test "x$enable_proxy" = "xyes"; then
+    AC_DEFINE([PROXY],1,[Set to nonzero if you want to enable proxy code])
+    CPPFLAGS="-Ivendor/lua/src -Ivendor/liburing/src/include $CPPFLAGS"
+    dnl lua needs math lib.
+    LIBS="$LIBS -lm -ldl"
+fi
+
+if test "x$enable_proxy_uring" = "xyes"; then
+    AC_DEFINE([HAVE_LIBURING],1,[Set to nonzero if you want to enable proxy uring handling])
+    CPPFLAGS="-Ivendor/liburing/src/include $CPPFLAGS"
+fi
+
 AM_CONDITIONAL([BUILD_DTRACE],[test "$build_dtrace" = "yes"])
 AM_CONDITIONAL([DTRACE_INSTRUMENT_OBJ],[test "$dtrace_instrument_obj" = "yes"])
 AM_CONDITIONAL([ENABLE_SASL],[test "$enable_sasl" = "yes"])
@@ -226,6 +244,8 @@ AM_CONDITIONAL([ENABLE_TLS],[test "$enable_tls" = "yes"])
 AM_CONDITIONAL([ENABLE_ASAN],[test "$enable_asan" = "yes"])
 AM_CONDITIONAL([ENABLE_STATIC],[test "$enable_static" = "yes"])
 AM_CONDITIONAL([DISABLE_UNIX_SOCKET],[test "$enable_unix_socket" = "no"])
+AM_CONDITIONAL([ENABLE_PROXY],[test "$enable_proxy" = "yes"])
+AM_CONDITIONAL([ENABLE_PROXY_URING],[test "$enable_proxy_uring" = "yes"])
 
 
 AC_SUBST(DTRACE)

logger.c

@@ -303,6 +303,34 @@ static int _logger_parse_cce(logentry *e, char *scratch) {
     return total;
 }
 
+#ifdef PROXY
+static void _logger_log_proxy_raw(logentry *e, const entry_details *d, const void *entry, va_list ap) {
+    struct timeval start = va_arg(ap, struct timeval);
+    char *cmd = va_arg(ap, char *);
+    unsigned short type = va_arg(ap, int);
+    unsigned short code = va_arg(ap, int);
+
+    struct logentry_proxy_raw *le = (void *)e->data;
+    struct timeval end;
+    gettimeofday(&end, NULL);
+    le->type = type;
+    le->code = code;
+    le->elapsed = (end.tv_sec - start.tv_sec) * 1000000 + (end.tv_usec - start.tv_usec);
+    memcpy(le->cmd, cmd, 9);
+}
+
+static int _logger_parse_prx_raw(logentry *e, char *scratch) {
+    int total;
+    struct logentry_proxy_raw *le = (void *)e->data;
+
+    total = snprintf(scratch, LOGGER_PARSE_SCRATCH,
+            "ts=%d.%d gid=%llu type=proxy_raw elapsed=%lu cmd=%s type=%d code=%d\n",
+            (int) e->tv.tv_sec, (int) e->tv.tv_usec, (unsigned long long) e->gid,
+            le->elapsed, le->cmd, le->type, le->code);
+    return total;
+}
+#endif
+
 /* Should this go somewhere else? */
 static const entry_details default_entries[] = {
     [LOGGER_ASCII_CMD] = {512, LOG_RAWCMDS, _logger_log_text, _logger_parse_text, "<%d %s"},
@@ -338,6 +366,15 @@ static const entry_details default_entries[] = {
         "type=compact_fraginfo ratio=%.2f bytes=%lu"
     },
 #endif
+#ifdef PROXY
+    [LOGGER_PROXY_CONFIG] = {512, LOG_SYSEVENTS, _logger_log_text, _logger_parse_text,
+        "type=proxy_conf status=%s"
+    },
+    [LOGGER_PROXY_RAW] = {512, LOG_RAWCMDS, _logger_log_proxy_raw, _logger_parse_prx_raw, NULL},
+    [LOGGER_PROXY_ERROR] = {512, LOG_SYSEVENTS, _logger_log_text, _logger_parse_text,
+        "type=proxy_error msg=%s"
+    },
+#endif
 };
 
 /*************************

logger.h

@@ -31,6 +31,11 @@ enum log_entry_type {
     LOGGER_COMPACT_END,
     LOGGER_COMPACT_FRAGINFO,
 #endif
+#ifdef PROXY
+    LOGGER_PROXY_CONFIG,
+    LOGGER_PROXY_RAW,
+    LOGGER_PROXY_ERROR,
+#endif
 };
 
 enum logger_ret_type {
@@ -106,7 +111,14 @@ struct logentry_conn_event {
     int sfd;
     struct sockaddr_in6 addr;
 };
-
+#ifdef PROXY
+struct logentry_proxy_raw {
+    unsigned short type;
+    unsigned short code;
+    long elapsed; // elapsed time in usec
+    char cmd[8];
+};
+#endif
 /* end intermediary structures */
 
 /* WARNING: cuddled items aren't compatible with warm restart. more code

memcached.c

@@ -56,6 +56,7 @@
 
 #include "proto_text.h"
 #include "proto_bin.h"
+#include "proto_proxy.h"
 
 #if defined(__FreeBSD__)
 #include <sys/sysctl.h>
@@ -85,7 +86,6 @@ static enum try_read_result try_read_udp(conn *c);
 
 static int start_conn_timeout_thread();
 
-
 /* stats */
 static void stats_init(void);
 static void conn_to_str(const conn *c, char *addr, char *svr_addr);
@@ -500,6 +500,11 @@ static const char *prot_text(enum protocol prot) {
         case negotiating_prot:
             rv = "auto-negotiate";
             break;
+#ifdef PROXY
+        case proxy_prot:
+            rv = "proxy";
+            break;
+#endif
     }
     return rv;
 }
@@ -797,6 +802,11 @@ conn *conn_new(const int sfd, enum conn_states init_state,
             case negotiating_prot:
                 c->try_read_command = try_read_command_negotiate;
                 break;
+#ifdef PROXY
+            case proxy_prot:
+                c->try_read_command = try_read_command_proxy;
+                break;
+#endif
         }
     }
 
@@ -1417,13 +1427,22 @@ static void reset_cmd_handler(conn *c) {
 
 static void complete_nread(conn *c) {
     assert(c != NULL);
+#ifdef PROXY
+    assert(c->protocol == ascii_prot
+           || c->protocol == binary_prot
+           || c->protocol == proxy_prot);
+#else
     assert(c->protocol == ascii_prot
            || c->protocol == binary_prot);
-
+#endif
     if (c->protocol == ascii_prot) {
         complete_nread_ascii(c);
     } else if (c->protocol == binary_prot) {
         complete_nread_binary(c);
+#ifdef PROXY
+    } else if (c->protocol == proxy_prot) {
+        complete_nread_proxy(c);
+#endif
     }
 }
 
@@ -1787,6 +1806,12 @@ void server_stats(ADD_STAT add_stats, conn *c) {
         APPEND_STAT("miss_from_extstore", "%llu", (unsigned long long)thread_stats.miss_from_extstore);
         APPEND_STAT("badcrc_from_extstore", "%llu", (unsigned long long)thread_stats.badcrc_from_extstore);
     }
+#endif
+#ifdef PROXY
+    if (settings.proxy_enabled) {
+        APPEND_STAT("proxy_conn_requests", "%llu", (unsigned long long)thread_stats.proxy_conn_requests);
+        APPEND_STAT("proxy_conn_errors", "%llu", (unsigned long long)thread_stats.proxy_conn_errors);
+    }
 #endif
     APPEND_STAT("delete_misses", "%llu", (unsigned long long)thread_stats.delete_misses);
     APPEND_STAT("delete_hits", "%llu", (unsigned long long)slab_stats.delete_hits);
@@ -1843,6 +1868,9 @@ void server_stats(ADD_STAT add_stats, conn *c) {
 #ifdef EXTSTORE
     storage_stats(add_stats, c);
 #endif
+#ifdef PROXY
+    proxy_stats(add_stats, c);
+#endif
 #ifdef TLS
     if (settings.ssl_enabled) {
         if (settings.ssl_session_cache) {
@@ -3801,6 +3829,9 @@ static void clock_handler(const evutil_socket_t fd, const short which, void *arg
         settings.sig_hup = false;
 
         authfile_load(settings.auth_file);
+#ifdef PROXY
+        proxy_start_reload(settings.proxy_ctx);
+#endif
     }
 
     evtimer_set(&clockevent, clock_handler, 0);
@@ -3999,6 +4030,9 @@ static void usage(void) {
            settings.ext_max_frag, settings.slab_automove_freeratio);
     verify_default("ext_item_age", settings.ext_item_age == UINT_MAX);
 #endif
+#ifdef PROXY
+    printf("   - proxy_config:        path to lua config file.\n");
+#endif
 #ifdef TLS
     printf("   - ssl_chain_cert:      certificate chain file in PEM format\n"
            "   - ssl_key:             private key, if not part of the -ssl_chain_cert\n"
@@ -4663,6 +4697,9 @@ int main (int argc, char **argv) {
         SSL_SESSION_CACHE,
         SSL_MIN_VERSION,
 #endif
+#ifdef PROXY
+        PROXY_CONFIG,
+#endif
 #ifdef MEMCACHED_DEBUG
         RELAXED_PRIVILEGES,
 #endif
@@ -4718,6 +4755,9 @@ int main (int argc, char **argv) {
         [SSL_SESSION_CACHE] = "ssl_session_cache",
         [SSL_MIN_VERSION] = "ssl_min_version",
 #endif
+#ifdef PROXY
+        [PROXY_CONFIG] = "proxy_config",
+#endif
 #ifdef MEMCACHED_DEBUG
         [RELAXED_PRIVILEGES] = "relaxed_privileges",
 #endif
@@ -5461,6 +5501,22 @@ int main (int argc, char **argv) {
                 }
                 settings.read_buf_mem_limit *= 1024 * 1024; /* megabytes */
                 break;
+#ifdef PROXY
+            case PROXY_CONFIG:
+                if (subopts_value == NULL) {
+                    fprintf(stderr, "Missing proxy_config file argument\n");
+                    return 1;
+                }
+                if (protocol_specified) {
+                    fprintf(stderr, "Cannot specify a protocol with proxy mode enabled\n");
+                    return 1;
+                }
+                settings.proxy_startfile = strdup(subopts_value);
+                settings.proxy_enabled = true;
+                settings.binding_protocol = proxy_prot;
+                protocol_specified = true;
+                break;
+#endif
 #ifdef MEMCACHED_DEBUG
             case RELAXED_PRIVILEGES:
                 settings.relaxed_privileges = true;
@@ -5868,6 +5924,14 @@ int main (int argc, char **argv) {
         exit(EX_OSERR);
     }
     /* start up worker threads if MT mode */
+#ifdef PROXY
+    if (settings.proxy_enabled) {
+        proxy_init();
+        if (proxy_load_config(settings.proxy_ctx) != 0) {
+            exit(EXIT_FAILURE);
+        }
+    }
+#endif
 #ifdef EXTSTORE
     slabs_set_storage(storage);
     memcached_thread_init(settings.num_threads, storage);