Expose number of currently active watchers in stats1.6.11

The stat key `log_watchers` indicates the number of active connected
`watch` clients.

7 files changed, 43 insertions, 5 deletions

diff --git a/doc/protocol.txt b/doc/protocol.txt
index 70341a5..af97270 100644
--- a/doc/protocol.txt
+++ b/doc/protocol.txt
@@ -1339,6 +1339,7 @@ integers separated by a colon (treat this as a floating point number).
 | log_worker_written    | 64u     | Logs written by a worker, to be picked up |
 | log_watcher_skipped   | 64u     | Logs not sent to slow watchers.           |
 | log_watcher_sent      | 64u     | Logs written to watchers.                 |
+| log_watchers          | 64u     | Number of currently active watchers.      |
 | unexpected_napi_ids   | 64u     | Number of times an unexpected napi id is  |
 |                       |         | is received. See doc/napi_ids.txt         |
 | round_robin_fallback  | 64u     | Number of times napi id of 0 is received  |
diff --git a/logger.c b/logger.c
index a5a87ab..667f3c7 100644
--- a/logger.c
+++ b/logger.c
@@ -668,12 +668,13 @@ static int logger_thread_poll_watchers(int force_poll, int watcher) {
     return flushed;
 }
 
-static void logger_thread_sum_stats(struct logger_stats *ls) {
+static void logger_thread_flush_stats(struct logger_stats *ls) {
     STATS_LOCK();
     stats.log_worker_dropped  += ls->worker_dropped;
     stats.log_worker_written  += ls->worker_written;
     stats.log_watcher_skipped += ls->watcher_skipped;
     stats.log_watcher_sent    += ls->watcher_sent;
+    stats_state.log_watchers   = ls->watcher_count;
     STATS_UNLOCK();
 }
 
@@ -709,6 +710,10 @@ static void *logger_thread(void *arg) {
         }
 
         logger_thread_poll_watchers(1, WATCHER_ALL);
+
+        /* capture the current count within mutual exclusion of the lock */
+        ls.watcher_count = watcher_count;
+
         pthread_mutex_unlock(&logger_stack_lock);
 
         /* TODO: abstract into a function and share with lru_crawler */
@@ -722,7 +727,7 @@ static void *logger_thread(void *arg) {
             if (to_sleep < MIN_LOGGER_SLEEP)
                 to_sleep = MIN_LOGGER_SLEEP;
         }
-        logger_thread_sum_stats(&ls);
+        logger_thread_flush_stats(&ls);
     }
 
     return NULL;
diff --git a/logger.h b/logger.h
index 4015206..039f443 100644
--- a/logger.h
+++ b/logger.h
@@ -170,6 +170,7 @@ struct logger_stats {
     uint64_t worker_written;
     uint64_t watcher_skipped;
     uint64_t watcher_sent;
+    uint64_t watcher_count;
 };
 
 extern pthread_key_t logger_key;
diff --git a/memcached.c b/memcached.c
index c7c22a8..8bbdccd 100644
--- a/memcached.c
+++ b/memcached.c
@@ -1838,6 +1838,7 @@ void server_stats(ADD_STAT add_stats, conn *c) {
     APPEND_STAT("log_worker_written", "%llu", (unsigned long long)stats.log_worker_written);
     APPEND_STAT("log_watcher_skipped", "%llu", (unsigned long long)stats.log_watcher_skipped);
     APPEND_STAT("log_watcher_sent", "%llu", (unsigned long long)stats.log_watcher_sent);
+    APPEND_STAT("log_watchers", "%llu", (unsigned long long)stats_state.log_watchers);
     STATS_UNLOCK();
 #ifdef EXTSTORE
     storage_stats(add_stats, c);
diff --git a/memcached.h b/memcached.h
index 3289bdb..12385db 100644
--- a/memcached.h
+++ b/memcached.h
@@ -396,6 +396,7 @@ struct stats_state {
     unsigned int  conn_structs;
     unsigned int  reserved_fds;
     unsigned int  hash_power_level; /* Better hope it's not over 9000 */
+    unsigned int  log_watchers; /* number of currently active watchers */
     bool          hash_is_expanding; /* If the hash table is being expanded */
     bool          accepting_conns;  /* whether we are currently accepting */
     bool          slab_reassign_running; /* slab reassign in progress */
diff --git a/t/stats.t b/t/stats.t
index 65780fb..106b242 100755
--- a/t/stats.t
+++ b/t/stats.t
@@ -28,9 +28,9 @@ if (MemcachedTest::enabled_tls_testing()) {
     # when TLS is enabled, stats contains additional keys:
     #   - ssl_handshake_errors
     #   - time_since_server_cert_refresh
-    is(scalar(keys(%$stats)), 82, "expected count of stats values");
+    is(scalar(keys(%$stats)), 83, "expected count of stats values");
 } else {
-    is(scalar(keys(%$stats)), 80, "expected count of stats values");
+    is(scalar(keys(%$stats)), 81, "expected count of stats values");
 }
 
 # Test initial state
diff --git a/t/watcher.t b/t/watcher.t
index c624fb9..4ae100d 100644
--- a/t/watcher.t
+++ b/t/watcher.t
@@ -5,7 +5,7 @@ use strict;
 use warnings;
 use Socket qw/SO_RCVBUF/;
 
-use Test::More tests => 25;
+use Test::More tests => 30;
 use FindBin qw($Bin);
 use lib "$Bin/lib";
 use MemcachedTest;
@@ -192,6 +192,35 @@ SKIP: {
         "logged item get with correct size");
 }
 
+# test watcher stats
+{
+    my $stats_server = new_memcached('-m 60 -o watcher_logbuf_size=8');
+    my $stats_client = $stats_server->sock;
+    my $stats;
+
+    my $watcher1 = $stats_server->new_sock;
+    print $watcher1 "watch fetchers\n";
+    $res = <$watcher1>;
+    is($res, "OK\r\n", 'fetchers watcher enabled');
+    sleep 1;
+    $stats = mem_stats($stats_client);
+    is($stats->{log_watchers}, 1, 'tracked new fetchers watcher');
+
+    my $watcher2 = $stats_server->new_sock;
+    print $watcher2 "watch fetchers\n";
+    $res = <$watcher2>;
+    is($res, "OK\r\n", 'mutations watcher enabled');
+    sleep 1;
+    $stats = mem_stats($stats_client);
+    is($stats->{log_watchers}, 2, 'tracked new mutations watcher');
+
+    $watcher1->close();
+    $watcher2->close();
+    sleep 1;
+    $stats = mem_stats($stats_client);
+    is($stats->{log_watchers}, 0, 'untracked all watchers');
+}
+
 # test no_watch option
 {
     my $nowatch_server = new_memcached('-W');