logger endpoints for first release

swapping "RAWCMDS" for internal hooks on when items are fetched or stored.

This doesn't map 1:1 with commands, ie: a store is internally a fetch then
store, so two log lines are generated. An ascii multiget one make one log line
per key fetched.

It's a good place to start. Need to come back and refactor parts of logger.c
again, then convert all prints in the codebase to log entries.

6 files changed, 192 insertions, 32 deletions

diff --git a/items.c b/items.c
index dd08942..0984f16 100644
--- a/items.c
+++ b/items.c
@@ -693,7 +693,6 @@ item *do_item_get(const char *key, const size_t nkey, const uint32_t hv, conn *c
             fprintf(stderr, "> NOT FOUND ");
         } else {
             fprintf(stderr, "> FOUND KEY ");
-            was_found++;
         }
         for (ii = 0; ii < nkey; ++ii) {
             fprintf(stderr, "%c", key[ii]);
@@ -701,13 +700,15 @@ item *do_item_get(const char *key, const size_t nkey, const uint32_t hv, conn *c
     }
 
     if (it != NULL) {
+        was_found = 1;
         if (item_is_flushed(it)) {
             do_item_unlink(it, hv);
             do_item_remove(it);
             it = NULL;
-            if (was_found) {
+            if (settings.verbose > 2) {
                 fprintf(stderr, " -nuked by flush");
             }
+            was_found = 2;
         } else if (it->exptime != 0 && it->exptime <= current_time) {
             do_item_unlink(it, hv);
             do_item_remove(it);
@@ -715,9 +716,10 @@ item *do_item_get(const char *key, const size_t nkey, const uint32_t hv, conn *c
             pthread_mutex_lock(&c->thread->stats.mutex);
             c->thread->stats.get_expired++;
             pthread_mutex_unlock(&c->thread->stats.mutex);
-            if (was_found) {
+            if (settings.verbose > 2) {
                 fprintf(stderr, " -nuked by expire");
             }
+            was_found = 3;
         } else {
             it->it_flags |= ITEM_FETCHED|ITEM_ACTIVE;
             DEBUG_REFCNT(it, '+');
@@ -726,6 +728,8 @@ item *do_item_get(const char *key, const size_t nkey, const uint32_t hv, conn *c
 
     if (settings.verbose > 2)
         fprintf(stderr, "\n");
+    /* For now this is in addition to the above verbose logging. */
+    LOGGER_LOG(c->thread->l, LOG_FETCHERS, LOGGER_ITEM_GET, NULL, was_found, key, nkey);
 
     return it;
 }
diff --git a/logger.c b/logger.c
index 6040532..506f31c 100644
--- a/logger.c
+++ b/logger.c
@@ -44,7 +44,9 @@ static char logger_uriencode_str[768];
 /* Should this go somewhere else? */
 static const entry_details default_entries[] = {
     [LOGGER_ASCII_CMD] = {LOGGER_TEXT_ENTRY, 512, LOG_RAWCMDS, "<%d %s"},
-    [LOGGER_EVICTION] = {LOGGER_EVICTION_ENTRY, 512, LOG_EVICTIONS, NULL}
+    [LOGGER_EVICTION] = {LOGGER_EVICTION_ENTRY, 512, LOG_EVICTIONS, NULL},
+    [LOGGER_ITEM_GET] = {LOGGER_ITEM_GET_ENTRY, 512, LOG_FETCHERS, NULL},
+    [LOGGER_ITEM_STORE] = {LOGGER_ITEM_STORE_ENTRY, 512, LOG_MUTATIONS, NULL}
 };
 
 #define WATCHER_ALL -1
@@ -170,6 +172,9 @@ static void logger_set_flags(void) {
  *************************/
 
 /* Completes rendering of log line, copies to subscribed watchers */
+/* FIXME: This can be shortened considerably with a refactor for both the
+ * "skipped" writing and string conversion.
+ */
 #define LOGGER_PARSE_SCRATCH 4096
 static enum logger_parse_entry_ret logger_thread_parse_entry(logentry *e, struct logger_stats *ls) {
     int total = 0;
@@ -177,6 +182,7 @@ static enum logger_parse_entry_ret logger_thread_parse_entry(logentry *e, struct
     int x;
     char scratch[LOGGER_PARSE_SCRATCH];
     char scratch2[LOGGER_PARSE_SCRATCH];
+    char *status = "unknown";
 
     switch (e->event) {
         case LOGGER_TEXT_ENTRY:
@@ -193,6 +199,75 @@ static enum logger_parse_entry_ret logger_thread_parse_entry(logentry *e, struct
                     scratch2, (le->it_flags & ITEM_FETCHED) ? "yes" : "no",
                     (long long int)le->exptime, le->latime);
             break;
+        case LOGGER_ITEM_GET_ENTRY: ;
+            struct logentry_item_get *lig = (struct logentry_item_get *) e->data;
+            switch (lig->was_found) {
+                case 0:
+                    status = "not_found";
+                    break;
+                case 1:
+                    status = "found";
+                    break;
+                case 2:
+                    status = "flushed";
+                    break;
+                case 3:
+                    status = "expired";
+                    break;
+            }
+            logger_uriencode(lig->key, scratch2, lig->nkey, LOGGER_PARSE_SCRATCH);
+            total = snprintf(scratch, LOGGER_PARSE_SCRATCH,
+                    "ts=%d.%d gid=%llu type=item_get key=%s status=%s\n",
+                    (int)e->tv.tv_sec, (int)e->tv.tv_usec, (unsigned long long) e->gid,
+                    scratch2, status);
+            break;
+        case LOGGER_ITEM_STORE_ENTRY: ;
+            struct logentry_item_store *lis = (struct logentry_item_store *) e->data;
+            char *cmd = "na";
+            switch (lis->status) {
+                case STORED:
+                    status = "stored";
+                    break;
+                case EXISTS:
+                    status = "exists";
+                    break;
+                case NOT_FOUND:
+                    status = "not_found";
+                    break;
+                case TOO_LARGE:
+                    status = "too_large";
+                    break;
+                case NO_MEMORY:
+                    status = "no_memory";
+                    break;
+            }
+            switch (lis->cmd) {
+                case NREAD_ADD:
+                    cmd = "add";
+                    break;
+                case NREAD_SET:
+                    cmd = "set";
+                    break;
+                case NREAD_REPLACE:
+                    cmd = "replace";
+                    break;
+                case NREAD_APPEND:
+                    cmd = "append";
+                    break;
+                case NREAD_PREPEND:
+                    cmd = "prepend";
+                    break;
+                case NREAD_CAS:
+                    cmd = "cas";
+                    break;
+            }
+            logger_uriencode(lis->key, scratch2, lis->nkey, LOGGER_PARSE_SCRATCH);
+            total = snprintf(scratch, LOGGER_PARSE_SCRATCH,
+                    "ts=%d.%d gid=%llu type=item_store key=%s status=%s cmd=%s\n",
+                    (int)e->tv.tv_sec, (int)e->tv.tv_usec, (unsigned long long) e->gid,
+                    scratch2, status, cmd);
+            break;
+
     }
 
     if (total >= LOGGER_PARSE_SCRATCH || total <= 0) {
@@ -564,6 +639,29 @@ static void _logger_log_evictions(logentry *e, item *it) {
     e->size = sizeof(struct logentry_eviction) + le->nkey;
 }
 
+/* 0 == nf, 1 == found. 2 == flushed. 3 == expired.
+ * might be useful to store/print the flags an item has?
+ * could also collapse this and above code into an "item status" struct. wait
+ * for more endpoints to be written before making it generic, though.
+ */
+static void _logger_log_item_get(logentry *e, const int was_found, const char *key, const int nkey) {
+    struct logentry_item_get *le = (struct logentry_item_get *) e->data;
+    le->was_found = was_found;
+    le->nkey = nkey;
+    memcpy(le->key, key, nkey);
+    e->size = sizeof(struct logentry_item_get) + nkey;
+}
+
+static void _logger_log_item_store(logentry *e, const enum store_item_type status,
+        const int comm, char *key, const int nkey) {
+    struct logentry_item_store *le = (struct logentry_item_store *) e->data;
+    le->status = status;
+    le->cmd = comm;
+    le->nkey = nkey;
+    memcpy(le->key, key, nkey);
+    e->size = sizeof(struct logentry_item_store) + nkey;
+}
+
 /* Public function for logging an entry.
  * Tries to encapsulate as much of the formatting as possible to simplify the
  * caller's code.
@@ -588,6 +686,9 @@ enum logger_ret_type logger_log(logger *l, const enum log_entry_type event, cons
     }
     e->gid = logger_get_gid();
     e->event = d->subtype;
+    /* TODO: Could pass this down as an argument now that we're using
+     * LOGGER_LOG() macro.
+     */
     e->eflags = d->eflags;
     /* Noting time isn't optional. A feature may be added to avoid rendering
      * time and/or gid to a logger.
@@ -609,6 +710,22 @@ enum logger_ret_type logger_log(logger *l, const enum log_entry_type event, cons
         case LOGGER_EVICTION_ENTRY:
             _logger_log_evictions(e, (item *)entry);
             break;
+        case LOGGER_ITEM_GET_ENTRY:
+            va_start(ap, entry);
+            int was_found = va_arg(ap, int);
+            char *key = va_arg(ap, char *);
+            size_t nkey = va_arg(ap, size_t);
+            _logger_log_item_get(e, was_found, key, nkey);
+            va_end(ap);
+            break;
+        case LOGGER_ITEM_STORE_ENTRY:
+            va_start(ap, entry);
+            enum store_item_type status = va_arg(ap, enum store_item_type);
+            int comm = va_arg(ap, int);
+            char *skey = va_arg(ap, char *);
+            size_t snkey = va_arg(ap, size_t);
+            _logger_log_item_store(e, status, comm, skey, snkey);
+            break;
     }
 
     /* Push pointer forward by the actual amount required */
diff --git a/logger.h b/logger.h
index e750378..6d95d40 100644
--- a/logger.h
+++ b/logger.h
@@ -12,12 +12,16 @@
 
 enum log_entry_type {
     LOGGER_ASCII_CMD = 0,
-    LOGGER_EVICTION
+    LOGGER_EVICTION,
+    LOGGER_ITEM_GET,
+    LOGGER_ITEM_STORE
 };
 
 enum log_entry_subtype {
     LOGGER_TEXT_ENTRY = 0,
-    LOGGER_EVICTION_ENTRY
+    LOGGER_EVICTION_ENTRY,
+    LOGGER_ITEM_GET_ENTRY,
+    LOGGER_ITEM_STORE_ENTRY
 };
 
 enum logger_ret_type {
@@ -39,6 +43,7 @@ typedef const struct {
     char *format;
 } entry_details;
 
+/* log entry intermediary structures */
 struct logentry_eviction {
     long long int exptime;
     uint32_t latime;
@@ -47,6 +52,21 @@ struct logentry_eviction {
     char key[];
 };
 
+struct logentry_item_get {
+    uint8_t was_found;
+    uint8_t nkey;
+    char key[];
+};
+
+struct logentry_item_store {
+    int status;
+    int cmd;
+    uint8_t nkey;
+    char key[];
+};
+
+/* end intermediary structures */
+
 typedef struct _logentry {
     enum log_entry_subtype event;
     uint16_t eflags;
diff --git a/memcached.c b/memcached.c
index 435c9f3..cce005b 100644
--- a/memcached.c
+++ b/memcached.c
@@ -1262,6 +1262,8 @@ static void complete_update_bin(conn *c) {
         write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, NULL, 0);
         break;
     case NOT_STORED:
+    case TOO_LARGE:
+    case NO_MEMORY:
         if (c->cmd == NREAD_ADD) {
             eno = PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS;
         } else if(c->cmd == NREAD_REPLACE) {
@@ -2051,11 +2053,17 @@ static void process_bin_update(conn *c) {
             realtime(req->message.body.expiration), vlen+2);
 
     if (it == 0) {
+        enum store_item_type status;
         if (! item_size_ok(nkey, req->message.body.flags, vlen + 2)) {
             write_bin_error(c, PROTOCOL_BINARY_RESPONSE_E2BIG, NULL, vlen);
+            status = TOO_LARGE;
         } else {
             out_of_memory(c, "SERVER_ERROR Out of memory allocating item");
+            status = NO_MEMORY;
         }
+        /* FIXME: losing c->cmd since it's translated below. refactor? */
+        LOGGER_LOG(c->thread->l, LOG_MUTATIONS, LOGGER_ITEM_STORE,
+                NULL, status, 0, key, nkey);
 
         /* Avoid stale data persisting in cache because we failed alloc.
          * Unacceptable for SET. Anywhere else too? */
@@ -2356,6 +2364,7 @@ enum store_item_type do_store_item(item *it, int comm, conn *c, const uint32_t h
             stored = EXISTS;
         }
     } else {
+        int failed_alloc = 0;
         /*
          * Append - combine new and old record into single one. Here it's
          * atomic and thread-safe.
@@ -2380,29 +2389,26 @@ enum store_item_type do_store_item(item *it, int comm, conn *c, const uint32_t h
                 new_it = do_item_alloc(key, it->nkey, flags, old_it->exptime, it->nbytes + old_it->nbytes - 2 /* CRLF */, hv);
 
                 if (new_it == NULL) {
-                    /* SERVER_ERROR out of memory */
-                    if (old_it != NULL)
-                        do_item_remove(old_it);
-
-                    return NOT_STORED;
-                }
+                    failed_alloc = 1;
+                    stored = NOT_STORED;
+                } else {
+                    /* copy data from it and old_it to new_it */
 
-                /* copy data from it and old_it to new_it */
+                    if (comm == NREAD_APPEND) {
+                        memcpy(ITEM_data(new_it), ITEM_data(old_it), old_it->nbytes);
+                        memcpy(ITEM_data(new_it) + old_it->nbytes - 2 /* CRLF */, ITEM_data(it), it->nbytes);
+                    } else {
+                        /* NREAD_PREPEND */
+                        memcpy(ITEM_data(new_it), ITEM_data(it), it->nbytes);
+                        memcpy(ITEM_data(new_it) + it->nbytes - 2 /* CRLF */, ITEM_data(old_it), old_it->nbytes);
+                    }
 
-                if (comm == NREAD_APPEND) {
-                    memcpy(ITEM_data(new_it), ITEM_data(old_it), old_it->nbytes);
-                    memcpy(ITEM_data(new_it) + old_it->nbytes - 2 /* CRLF */, ITEM_data(it), it->nbytes);
-                } else {
-                    /* NREAD_PREPEND */
-                    memcpy(ITEM_data(new_it), ITEM_data(it), it->nbytes);
-                    memcpy(ITEM_data(new_it) + it->nbytes - 2 /* CRLF */, ITEM_data(old_it), old_it->nbytes);
+                    it = new_it;
                 }
-
-                it = new_it;
             }
         }
 
-        if (stored == NOT_STORED) {
+        if (stored == NOT_STORED && failed_alloc == 0) {
             if (old_it != NULL)
                 item_replace(old_it, it, hv);
             else
@@ -2422,6 +2428,8 @@ enum store_item_type do_store_item(item *it, int comm, conn *c, const uint32_t h
     if (stored == STORED) {
         c->cas = ITEM_get_cas(it);
     }
+    LOGGER_LOG(c->thread->l, LOG_MUTATIONS, LOGGER_ITEM_STORE, NULL,
+            stored, comm, ITEM_key(it), it->nkey);
 
     return stored;
 }
@@ -3126,10 +3134,16 @@ static void process_update_command(conn *c, token_t *tokens, const size_t ntoken
     it = item_alloc(key, nkey, flags, realtime(exptime), vlen);
 
     if (it == 0) {
-        if (! item_size_ok(nkey, flags, vlen))
+        enum store_item_type status;
+        if (! item_size_ok(nkey, flags, vlen)) {
             out_string(c, "SERVER_ERROR object too large for cache");
-        else
+            status = TOO_LARGE;
+        } else {
             out_of_memory(c, "SERVER_ERROR out of memory storing object");
+            status = NO_MEMORY;
+        }
+        LOGGER_LOG(c->thread->l, LOG_MUTATIONS, LOGGER_ITEM_STORE,
+                NULL, status, comm, key, nkey);
         /* swallow the data line */
         c->write_and_go = conn_swallow;
         c->sbytes = vlen;
@@ -3452,13 +3466,17 @@ static void process_watch_command(conn *c, token_t *tokens, const size_t ntokens
                 f |= LOG_RAWCMDS;
             } else if ((strcmp(tokens[x].value, "evictions") == 0)) {
                 f |= LOG_EVICTIONS;
+            } else if ((strcmp(tokens[x].value, "fetchers") == 0)) {
+                f |= LOG_FETCHERS;
+            } else if ((strcmp(tokens[x].value, "mutations") == 0)) {
+                f |= LOG_MUTATIONS;
             } else {
                 out_string(c, "ERROR");
                 return;
             }
         }
     } else {
-        f |= LOG_RAWCMDS;
+        f |= LOG_FETCHERS;
     }
 
     switch(logger_add_watcher(c, c->sfd, f)) {
@@ -3485,7 +3503,8 @@ static void process_command(conn *c, char *command) {
 
     MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes);
 
-    LOGGER_LOG(c->thread->l, LOG_RAWCMDS, LOGGER_ASCII_CMD, NULL, c->sfd, command);
+    if (settings.verbose > 1)
+        fprintf(stderr, "<%d %s\n", c->sfd, command);
 
     /*
      * for commands set/add/replace, we build an item and read the data
diff --git a/memcached.h b/memcached.h
index a64bcb8..01b15d1 100644
--- a/memcached.h
+++ b/memcached.h
@@ -212,7 +212,7 @@ enum pause_thread_types {
 #define NREAD_CAS 6
 
 enum store_item_type {
-    NOT_STORED=0, STORED, EXISTS, NOT_FOUND
+    NOT_STORED=0, STORED, EXISTS, NOT_FOUND, TOO_LARGE, NO_MEMORY
 };
 
 enum delta_result_type {
diff --git a/t/watcher.t b/t/watcher.t
index 8bccb85..1438cf8 100644
--- a/t/watcher.t
+++ b/t/watcher.t
@@ -40,7 +40,7 @@ while (my $log = <$watcher>) {
     last;
 }
 $res = <$watcher>;
-like($res, qr/ts=\d+\.\d+\ gid=\d+.*get foo/, "saw a real log line after a skip");
+like($res, qr/ts=\d+\.\d+\ gid=\d+ type=item_get/, "saw a real log line after a skip");
 
 # test combined logs
 # fill to evictions, then enable watcher, set again, and look for both lines
@@ -54,7 +54,7 @@ like($res, qr/ts=\d+\.\d+\ gid=\d+.*get foo/, "saw a real log line after a skip"
     }
 
     $watcher = $server->new_sock;
-    print $watcher "watch rawcmds evictions\n";
+    print $watcher "watch mutations evictions\n";
     $res = <$watcher>;
     is($res, "OK\r\n", "new watcher enabled");
     my $watcher2 = $server->new_sock;
@@ -66,8 +66,8 @@ like($res, qr/ts=\d+\.\d+\ gid=\d+.*get foo/, "saw a real log line after a skip"
     my $found_log = 0;
     my $found_ev  = 0;
     while (my $log = <$watcher>) {
-        $found_log = 1 if ($log =~ m/set bfoo/);
-        $found_ev = 1 if ($log =~ m/eviction/);
+        $found_log = 1 if ($log =~ m/type=item_store/);
+        $found_ev = 1 if ($log =~ m/type=eviction/);
         last if ($found_log && $found_ev);
     }
     is($found_log, 1, "found rawcmd log entry");