log: Add a new watcher to watch for deletions.

`watch deletions`: would log all keys which are deleted using either `delete` or `md` command.
The log line would contain the command used, the key, the clsid and size of the deleted item.

Items which result in delete miss or are marked as stale wouldn't show up in the logs

6 files changed, 102 insertions, 3 deletions

diff --git a/doc/protocol.txt b/doc/protocol.txt
index 8a261ed..cd476db 100644
--- a/doc/protocol.txt
+++ b/doc/protocol.txt
@@ -1174,7 +1174,7 @@ Watchers are a way to connect to memcached and inspect what's going on
 internally. This is an evolving feature so new endpoints should show up over
 time.
 
-watch <fetchers|mutations|evictions|connevents>
+watch <fetchers|mutations|evictions|connevents|deletions>
 
 - Turn connection into a watcher. Options can be stacked and are
   space-separated. Logs will be sent to the watcher until it disconnects.
@@ -1223,6 +1223,9 @@ The arguments are:
 
 - "proxyuser": Emits log entries created from lua configuration files.
 
+- "deletions": Emits logs when an item is successfully deleted from the
+  cache using `delete` or `md` commands. delete misses wouldn't be logged
+
 Statistics
 ----------
 
diff --git a/logger.c b/logger.c
index 3f483ed..1ff7211 100644
--- a/logger.c
+++ b/logger.c
@@ -137,6 +137,18 @@ static void _logger_log_item_store(logentry *e, const entry_details *d, const vo
     e->size = sizeof(struct logentry_item_store) + nkey;
 }
 
+static void _logger_log_item_deleted(logentry *e, const entry_details *d, const void *entry, va_list ap) {
+    item *it = (item *)entry;
+    int comm = va_arg(ap, int);
+    struct logentry_deletion *le = (struct logentry_deletion *) e->data;
+    le->nkey = it->nkey;
+    le->cmd = comm;
+    le->nbytes = it->nbytes;
+    le->clsid = ITEM_clsid(it);
+    memcpy(le->key, ITEM_key(it), it->nkey);
+    e->size = sizeof(struct logentry_deletion) + le->nkey;
+}
+
 static void _logger_log_conn_event(logentry *e, const entry_details *d, const void *entry, va_list ap) {
     struct sockaddr_in6 *addr = va_arg(ap, struct sockaddr_in6 *);
     socklen_t addrlen = va_arg(ap, socklen_t);
@@ -251,6 +263,26 @@ static int _logger_parse_ee(logentry *e, char *scratch) {
     return total;
 }
 
+static int _logger_parse_ide(logentry *e, char *scratch) {
+    int total;
+    const char *cmd = "na";
+    const char * const cmd_map[] = {
+            "null", "delete", "md" };
+    char keybuf[KEY_MAX_URI_ENCODED_LENGTH];
+    struct logentry_deletion *le = (struct logentry_deletion *) e->data;
+    uriencode(le->key, keybuf, le->nkey, KEY_MAX_URI_ENCODED_LENGTH);
+
+    if (le->cmd <= 2)
+        cmd = cmd_map[le->cmd];
+
+    total = snprintf(scratch, LOGGER_PARSE_SCRATCH,
+                     "ts=%d.%d gid=%llu type=deleted key=%s cmd=%s clsid=%u size=%d\n",
+                     (int)e->tv.tv_sec, (int)e->tv.tv_usec, (unsigned long long) e->gid,
+                     keybuf, cmd, le->clsid,
+                     le->nbytes > 0 ? le->nbytes - 2 : 0); // CLRF
+    return total;
+}
+
 #ifdef EXTSTORE
 static int _logger_parse_extw(logentry *e, char *scratch) {
     int total;
@@ -445,6 +477,7 @@ static const entry_details default_entries[] = {
     },
     [LOGGER_CONNECTION_NEW] = {512, LOG_CONNEVENTS, _logger_log_conn_event, _logger_parse_cne, NULL},
     [LOGGER_CONNECTION_CLOSE] = {512, LOG_CONNEVENTS, _logger_log_conn_event, _logger_parse_cce, NULL},
+    [LOGGER_DELETIONS] = {512, LOG_DELETIONS, _logger_log_item_deleted, _logger_parse_ide, NULL},
 #ifdef EXTSTORE
     [LOGGER_EXTSTORE_WRITE] = {512, LOG_EVICTIONS, _logger_log_ext_write, _logger_parse_extw, NULL},
     [LOGGER_COMPACT_START] = {512, LOG_SYSEVENTS, _logger_log_text, _logger_parse_text,
diff --git a/logger.h b/logger.h
index 4e8f1bb..9d2630c 100644
--- a/logger.h
+++ b/logger.h
@@ -22,6 +22,7 @@ enum log_entry_type {
     LOGGER_SLAB_MOVE,
     LOGGER_CONNECTION_NEW,
     LOGGER_CONNECTION_CLOSE,
+    LOGGER_DELETIONS,
 #ifdef EXTSTORE
     LOGGER_EXTSTORE_WRITE,
     LOGGER_COMPACT_START,
@@ -108,6 +109,14 @@ struct logentry_item_store {
     char key[];
 };
 
+struct logentry_deletion {
+    int nbytes;
+    int cmd;
+    uint8_t nkey;
+    uint8_t clsid;
+    char key[];
+};
+
 struct logentry_conn_event {
     int transport;
     int reason;
@@ -164,6 +173,7 @@ struct _logentry {
 #define LOG_PROXYREQS  (1<<10) /* command logs from proxy */
 #define LOG_PROXYEVENTS (1<<11) /* error log stream from proxy */
 #define LOG_PROXYUSER (1<<12) /* user generated logs from proxy */
+#define LOG_DELETIONS (1<<13) /* see whats deleted */
 
 typedef struct _logger {
     struct _logger *prev;
diff --git a/memcached.h b/memcached.h
index 7ebe65f..3c54931 100644
--- a/memcached.h
+++ b/memcached.h
@@ -272,6 +272,9 @@ enum close_reasons {
 #define CAS_ALLOW_STALE true
 #define CAS_NO_STALE false
 
+#define LOG_TYPE_DELETE 1
+#define LOG_TYPE_META_DELETE 2
+
 enum store_item_type {
     NOT_STORED=0, STORED, EXISTS, NOT_FOUND, TOO_LARGE, NO_MEMORY
 };
diff --git a/proto_text.c b/proto_text.c
index 1f69ddc..58fd227 100644
--- a/proto_text.c
+++ b/proto_text.c
@@ -1684,6 +1684,7 @@ static void process_mdelete_command(conn *c, token_t *tokens, const size_t ntoke
             c->thread->stats.slab_stats[ITEM_clsid(it)].delete_hits++;
             pthread_mutex_unlock(&c->thread->stats.mutex);
 
+            LOGGER_LOG(NULL, LOG_DELETIONS, LOGGER_DELETIONS, it, LOG_TYPE_META_DELETE);
             do_item_unlink(it, hv);
             STORAGE_delete(c->thread->storage, it);
             if (c->noreply)
@@ -2176,7 +2177,7 @@ static void process_delete_command(conn *c, token_t *tokens, const size_t ntoken
         pthread_mutex_lock(&c->thread->stats.mutex);
         c->thread->stats.slab_stats[ITEM_clsid(it)].delete_hits++;
         pthread_mutex_unlock(&c->thread->stats.mutex);
-
+        LOGGER_LOG(NULL, LOG_DELETIONS, LOGGER_DELETIONS, it, LOG_TYPE_DELETE);
         do_item_unlink(it, hv);
         STORAGE_delete(c->thread->storage, it);
         do_item_remove(it);      /* release our reference */
@@ -2322,6 +2323,8 @@ static void process_watch_command(conn *c, token_t *tokens, const size_t ntokens
                 f |= LOG_PROXYEVENTS;
             } else if ((strcmp(tokens[x].value, "proxyuser") == 0)) {
                 f |= LOG_PROXYUSER;
+            } else if ((strcmp(tokens[x].value, "deletions") == 0)) {
+                f |= LOG_DELETIONS;
             } else {
                 out_string(c, "ERROR");
                 return;
diff --git a/t/watcher.t b/t/watcher.t
index f977c2a..09800a9 100755
--- a/t/watcher.t
+++ b/t/watcher.t
@@ -10,7 +10,7 @@ use FindBin qw($Bin);
 use lib "$Bin/lib";
 use MemcachedTest;
 
-plan tests => 30;
+plan tests => 43;
 
 my $server = new_memcached('-m 60 -o watcher_logbuf_size=8');
 my $client = $server->sock;
@@ -235,3 +235,50 @@ SKIP: {
 
     is($watchresult, "CLIENT_ERROR watch commands not allowed\r\n", "attempted watch gave client error with no_watch option set");
 }
+
+
+# test delete/meta-delete with value sizes
+{
+    my $watcher = $server->new_sock;
+    print $watcher "watch deletions\n";
+    is(<$watcher>, "OK\r\n", "deletions watcher enabled");
+
+    print $client "set vfoo 0 0 4\r\nvbar\r\n";
+    is(<$client>, "STORED\r\n", "stored the key");
+
+    # wouldn't be logged to watcher
+    print $client "md non-existing-key\r\n";
+    is(<$client>, "NF\r\n", "non-existing key can't be deleted");
+
+    print $client "delete vfoo\r\n";
+    is(<$client>, "DELETED\r\n", "key was deleted");
+
+    like(<$watcher>, qr/ts=\d+\.\d+\ gid=\d+ type=deleted key=vfoo cmd=delete .+ size=4/,
+        "delete command logged with correct size");
+
+    print $client "set vfoo 0 0 4\r\nvbar\r\n";
+    is(<$client>, "STORED\r\n", "stored the key");
+
+    print $client "md vfoo\r\n";
+    is(<$client>, "HD\r\n", "key was deleted");
+
+    like(<$watcher>, qr/ts=\d+\.\d+\ gid=\d+ type=deleted key=vfoo cmd=md .+ size=4/,
+        "meta-delete command logged with correct size");
+
+    print $client "set vfoo 0 0 4\r\nvbar\r\n";
+    is(<$client>, "STORED\r\n", "stored the key");
+
+    # shouldn't result in log line
+    print $client "md vfoo I\r\n";
+    is(<$client>, "HD\r\n", "key was marked stale");
+
+    print $client "mg vfoo\r\n";
+    is(<$client>, "HD X W\r\n", "mg shows key is stale and won recache");
+
+    # now do a delete and it will show up in the log, explicitly use delete instead of md to check the cmd= pair in log
+    print $client "delete vfoo\r\n";
+    is(<$client>, "DELETED\r\n", "key was deleted");
+
+    like(<$watcher>, qr/ts=\d+\.\d+\ gid=\d+ type=deleted key=vfoo cmd=delete .+ size=4/,
+        "delete command logged with correct size");
+}