Merge pull request #98 from minkikim89/minkikim89/failstop_config

FEATURE: failstop configuration

Author: jhpark816

arcus_zk.c

@@ -147,6 +147,9 @@ typedef struct {
     char    *mc_ipport;         // this memcached ip:port string
     char    *hostip;            // localhost server IP
     int     port;               // memcached port number
+#ifdef CONFIG_FAILSTOP
+    bool    mc_failstop;        // memcached automatic failstop on/off
+#endif
     int     hb_failstop;        // memcached heartbeat failstop
     int     hb_timeout;         // memcached heartbeat timeout
     int     zk_timeout;         // Zookeeper session timeout
@@ -172,6 +175,9 @@ arcus_zk_conf arcus_conf = {
     .svc            = NULL,
     .mc_ipport      = NULL,
     .hostip         = NULL,
+#ifdef CONFIG_FAILSTOP
+    .mc_failstop    = true,
+#endif
     .hb_failstop    = HEART_BEAT_DFT_FAILSTOP,
     .hb_timeout     = HEART_BEAT_DFT_TIMEOUT,
     .zk_timeout     = DEFAULT_ZK_TO,
@@ -243,6 +249,10 @@ struct sm {
     volatile bool timer_running;
 #endif
 
+#ifdef CONFIG_FAILSTOP
+    volatile bool mc_pause;
+#endif
+
     pthread_t state_tid;
 #ifdef ENABLE_SUICIDE_UPON_DISCONNECT
     /* A timer thread to fail-stop the server
@@ -336,6 +346,35 @@ static int wait_count(int timeout)
     return rc;
 }
 
+#ifdef CONFIG_FAILSTOP
+static int
+arcus_zk_client_init(zk_info_t *zinfo)
+{
+    inc_count(1);
+    zinfo->zh = zookeeper_init(zinfo->ensemble_list, arcus_zk_watcher,
+                               arcus_conf.zk_timeout, &zinfo->myid, zinfo, 0);
+    if (!zinfo->zh) {
+        arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+                "zookeeper_init() failed (%s error)\n", strerror(errno));
+        return EX_PROTOCOL;
+    }
+    /* wait until above init callback is called
+     * We need to wait until ZOO_CONNECTED_STATE
+     */
+    if (wait_count(arcus_conf.zk_timeout) != 0) {
+        /* zoo_state(zh) != ZOO_CONNECTED_STATE */
+        arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+                 "cannot to be ZOO_CONNECTED_STATE\n");
+        zookeeper_close(zinfo->zh);
+        zinfo->zh = NULL;
+        inc_count(-1);
+        return EX_PROTOCOL;
+    }
+
+    return 0;
+}
+#endif
+
 // Arcus zookeeper global watch callback routine
 static void
 arcus_zk_watcher(zhandle_t *wzh, int type, int state, const char *path, void *cxt)
@@ -371,6 +410,22 @@ arcus_zk_watcher(zhandle_t *wzh, int type, int state, const char *path, void *cx
         arcus_exit(wzh, EX_NOPERM);
     }
     else if (state == ZOO_EXPIRED_SESSION_STATE) {
+#ifdef CONFIG_FAILSTOP
+        if (arcus_conf.mc_failstop) {
+            // very likely that memcached process exited and restarted within
+            // session timeout
+            arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL, "Expired state. shutting down\n");
+            // send SMS here??
+            arcus_exit(wzh, EX_TEMPFAIL);
+        } else {
+            arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL, "Expired state. pausing memcached (mc_failstop: off)\n");
+
+            sm_lock();
+            sm_info.mc_pause = true;
+            sm_wakeup(true);
+            sm_unlock();
+        }
+#else
         // very likely that memcached process exited and restarted within
         // session timeout
         arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL, "Expired state. shutting down\n");
@@ -381,6 +436,7 @@ arcus_zk_watcher(zhandle_t *wzh, int type, int state, const char *path, void *cx
          * to avoid stale cache data access (rehashing already happened)
          */
         //arcus_zk_init(arcus_conf.ensemble_list, settings );
+#endif
     }
     else if (state == ZOO_ASSOCIATING_STATE || state == ZOO_CONNECTING_STATE) {
         /* we get these when connection to Ensemble is dropped and retrying
@@ -1181,6 +1237,14 @@ void arcus_zk_init(char *ensemble_list, int zk_to,
 
     gettimeofday(&start_time, 0);
 
+#ifdef CONFIG_FAILSTOP
+    rc = arcus_zk_client_init(zinfo);
+    if (rc != 0) {
+        arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+                               "Failed to initialize zk client\n");
+        arcus_exit(NULL, rc);
+    }
+#else
     inc_count(1);
     zinfo->zh = zookeeper_init(zinfo->ensemble_list, arcus_zk_watcher,
                                arcus_conf.zk_timeout, &zinfo->myid, zinfo, 0);
@@ -1198,6 +1262,7 @@ void arcus_zk_init(char *ensemble_list, int zk_to,
                  "cannot to be ZOO_CONNECTED_STATE\n");
         arcus_exit(zk_info.zh, EX_PROTOCOL);
     }
+#endif
 
     /* setting main zk */
     main_zk = zinfo;
@@ -1434,6 +1499,136 @@ int arcus_zk_get_ensemble(char *buf, int size)
     return ret;
 }
 
+#ifdef CONFIG_FAILSTOP
+int arcus_zk_rejoin_ensemble()
+{
+    int ret = 0;
+
+    pthread_mutex_lock(&zk_lock);
+
+    if (main_zk->zh != NULL) {
+        pthread_mutex_unlock(&zk_lock);
+        arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+            "Failed to rejoin ensemble. It's already a member of cloud.\n");
+        return -1;
+    }
+
+    assert(sm_info.mc_pause == true);
+
+    do {
+        struct timeval  start_time, end_time;
+        long            difftime_us;
+
+        /* initialize Arcus ZK stats */
+        memset(&azk_stat, 0, sizeof(azk_stat));
+
+        memset(&main_zk->myid, 0, sizeof(clientid_t));
+
+        assert(main_zk->ensemble_list);
+
+        gettimeofday(&start_time, 0);
+
+        ret = arcus_zk_client_init(main_zk);
+        if (ret != 0) {
+            arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+                                   "Failed to initialize zk client.\n");
+            break;
+        }
+
+        /* check zk root directory and get the serice code */
+        if (zk_root == NULL) {
+            zk_root = "/arcus"; /* set zk root directory */
+            ret = arcus_check_server_mapping(main_zk->zh, zk_root);
+            if (ret != 0) {
+                arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+                         "Failed to check server mapping for this cache node. "
+                         "(zk_root=%s)\n", zk_root);
+            }
+        }
+
+        /* create "/cache_list/{svc}/ip:port-hostname" ephemeral znode */
+        ret = arcus_create_ephemeral_znode(main_zk->zh, zk_root);
+        if (ret != 0) {
+            arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+                                   "arcus_create_ephemeral_znode() failed.\n");
+            break;
+        }
+
+        /* log this rejoin activity in /arcus/cache_server_log */
+        arcus_zk_log(main_zk->zh, "rejoin");
+
+        gettimeofday(&end_time, 0);
+        difftime_us = (end_time.tv_sec*1000000 + end_time.tv_usec) -
+                      (start_time.tv_sec*1000000 + start_time.tv_usec);
+
+        /* We have finished registering this memcached instance to Arcus cluster */
+        arcus_conf.logger->log(EXTENSION_LOG_INFO, NULL,
+                "Memcached rejoined Arcus cache cloud for \"%s\" service "
+                "(took %ld microsec)\n", arcus_conf.svc, difftime_us);
+
+        // "recv" timeout is actually the session timeout
+        // ZK client ping period is recv_timeout / 3.
+        arcus_conf.logger->log(EXTENSION_LOG_INFO, NULL,
+                "ZooKeeper session timeout: %d sec\n", zoo_recv_timeout(main_zk->zh)/1000);
+
+        assert(arcus_conf.ch);
+
+        struct String_vector strv = { 0, NULL };
+        /* 2nd argument, NULL means no watcher */
+        if (arcus_read_ZK_children(main_zk->zh, arcus_conf.cluster_path, NULL, &strv) <= 0) {
+            arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+                    "Failed to read cache list from ZK.\n");
+            ret = -1; break;
+        }
+
+        if (update_cluster_config(&strv) != 0) {
+            arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+                    "Failed to update cluster config.\n");
+            ret = -1; break;
+        }
+
+        deallocate_String_vector(&strv);
+
+        arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL,
+            "Memcached is watching Arcus cache cloud for \"%s\"\n", arcus_conf.cluster_path);
+
+
+        /* Wake up the state machine thread.
+         * Tell it to refresh the hash ring (cluster_config).
+         */
+        sm_lock();
+        sm_info.mc_pause = false;
+        /* Don't care if we just read the list above.  Do it again. */
+        sm_info.request.update_cache_list = true;
+        sm_wakeup(true);
+        sm_unlock();
+    } while(0);
+
+    if (ret != 0) {
+        if (main_zk->zh) {
+          zookeeper_close(main_zk->zh);
+          main_zk->zh = NULL;
+        }
+        ret = -1;
+    }
+    pthread_mutex_unlock(&zk_lock);
+
+    return ret;
+}
+
+void arcus_zk_set_mcfailstop(bool mcfailstop)
+{
+    pthread_mutex_lock(&arcus_conf.lock);
+    arcus_conf.mc_failstop = mcfailstop;
+    pthread_mutex_unlock(&arcus_conf.lock);
+}
+
+bool arcus_zk_get_mcfailstop(void)
+{
+    return arcus_conf.mc_failstop;
+}
+#endif
+
 int arcus_zk_set_hbtimeout(int hbtimeout)
 {
     int ret=0;
@@ -1535,6 +1730,9 @@ static void *sm_state_thread(void *arg)
     bool sm_retry = false;
     bool shutdown_by_me = false;
 
+#ifdef CONFIG_FAILSTOP
+    sm_info.mc_pause = false;
+#endif
     sm_info.state_running = true;
     while (!arcus_zk_shutdown)
     {
@@ -1566,6 +1764,23 @@ static void *sm_state_thread(void *arg)
         if (arcus_zk_shutdown)
             break;
 
+#ifdef CONFIG_FAILSTOP
+        if (sm_info.mc_pause == true) {
+            bool paused = false;
+            pthread_mutex_lock(&zk_lock);
+            if (sm_info.mc_pause == true) {
+                if (main_zk->zh != NULL) {
+                    zookeeper_close(main_zk->zh);
+                    main_zk->zh = NULL;
+                    arcus_conf.logger->log(EXTENSION_LOG_WARNING, NULL, "zk connection closed\n");
+                }
+                paused = true;
+            }
+            pthread_mutex_unlock(&zk_lock);
+            if (paused) continue;
+        }
+#endif
+
         /* Read the latest hash ring */
         if (smreq.update_cache_list) {
             struct String_vector strv_cache_list = {0, NULL};

arcus_zk.h

@@ -45,6 +45,12 @@ void arcus_zk_destroy(void);
 
 int  arcus_zk_set_ensemble(char *ensemble_list);
 int  arcus_zk_get_ensemble(char *buf, int size);
+#ifdef CONFIG_FAILSTOP
+int  arcus_zk_rejoin_ensemble(void);
+
+void arcus_zk_set_mcfailstop(bool failstop);
+bool arcus_zk_get_mcfailstop(void);
+#endif
 int  arcus_zk_set_hbtimeout(int hbtimeout);
 int  arcus_zk_get_hbtimeout(void);
 int  arcus_zk_set_hbfailstop(int hbfailstop);

doc/command-administration.md

@@ -266,6 +266,7 @@ Arcus cache server는 특정 configuration에 대해 동적으로 변경하거
 
 - verbosity
 - memlimit
+- mcfailstop
 - maxconns
 
 **config verbosity**
@@ -291,6 +292,16 @@ config memlimit [<memsize>]\r\n
 Arcus cache server가 현재 사용 중인 메모리 크기인 tatal_malloced 보다 큰 크기로만 설정이 가능하다.
 이 인자가 생략되면 현재 설정되어 있는 memory limit 값을 조회한다.
 
+**config mcfailstop**
+
+Arcus cache server의 automatic failstop 기능을 on 또는 off 한다.
+
+```
+config mcfailstop [on|off]\r\n
+```
+
+Network failure 상태에서 정상적인 서비스를 진행하지 못하는 cache server가 cache cloud에 그대로 존재할 경우, 해당 cache server가 담당하고 있는 data 범위에 대한 요청이 모두 실패하고 DB에 부담을 주게 된다. 또한 이후에 ZooKeeper에 재연결 되더라도 old data를 가지고 있을 가능성이 있으며 이로 인해 응용에 오동작을 발생시킬 수 있다. Arcus cache server는 이를 해결하기위해 ZooKeepeeper session timeout이 발생할 경우 failed cache server를 cache cloud에서 자동으로 제거하는 automatic failstop 기능을 기본적으로 제공한다.
+
 **config maxconns**
 
 Arcus cache server 구동 시에 -c 옵션으로 설정된 최대 연결 수를 동적으로(restart 없이) 변경/조회한다.
@@ -489,12 +500,15 @@ Arcus cache server가 연결되어 있는 ZooKeeper ensemble 설정에 대한 
 ```
 zkensemble set <ensemble_list>\r\n
 zkensemble get\r\n
+zkensemble rejoin\r\n
 ```
 
 set 명령은 ZK ensemble 주소를 변경한다. ensemble_list는 \<ip:port\>,...,\<ip:port\>  와 같은 ZK server 들의 list 형태 혹은 ZK ensemble의 도메인 주소 형태로 지정할 수 있다.
 
 get 명령은 ZK ensemble 주소를 조회한다. 조회 결과는 \<ip:port\>,...,\<ip:port\> 형식으로 확인할 수 있다.
 
+rejoin 명령은 cache server를 ZK ensemble에 다시 연결하도록 하는 명령이다. failstop이 off 된 상태에서 ZooKeeper session timeout이 일어날 경우 cache server는 ZooKeeper와의 연결을 끊고 cache cloud에서 빠져 대기한다. 이 때 이 rejoin 명령을 내리면 기존 data를 가진 그대로 cache cloud에 참여하게 된다.
+
 ### Help 명령
 
 Arcus cache server의 acsii command syntax를 조회한다.

include/memcached/types.h

@@ -31,6 +31,7 @@ struct iovec {
 #include <sys/uio.h>
 #endif
 
+#define CONFIG_FAILSTOP
 #define CHANGE_STANDARD_FREE_AVAIL
 #define SUPPORT_BOP_MGET
 #define SUPPORT_BOP_SMGET