Line data    Source code

       1             : //
       2             : // Copyright (c) 2016 Juniper Networks, Inc. All rights reserved.
       3             : //
       4             : 
       5             : #include <cerrno>
       6             : #include <cstring>
       7             : 
       8             : #include <base/logging.h>
       9             : 
      10             : #include <zookeeper/zookeeper_client.h>
      11             : #include <zookeeper/zookeeper_client_impl.h>
      12             : #include <zookeeper/zookeeper_interface.h>
      13             : 
      14             : #define ZOO_LOG(_Level, _Msg)                                             \
      15             :     do {                                                                  \
      16             :         if (LoggingDisabled()) break;                                     \
      17             :         log4cplus::Logger logger = log4cplus::Logger::getRoot();          \
      18             :         LOG4CPLUS_##_Level(logger, __func__ << ":" << __FILE__ << ":" <<  \
      19             :             __LINE__ << ": " << _Msg);                                    \
      20             :     } while (false)
      21             : 
      22             : #define ZOO_LOG_ERR(_Msg)                                                 \
      23             :     do {                                                                  \
      24             :         LOG(ERROR, __func__ << ":" << __FILE__ << ":" << __LINE__ << ": " \
      25             :             << _Msg);                                                     \
      26             :     } while (false)
      27             : 
      28             : namespace zookeeper {
      29             : namespace interface {
      30             : 
      31             : class ZookeeperCBindings : public ZookeeperInterface {
      32             :  public:
      33           0 :     ZookeeperCBindings() {
      34           0 :     }
      35           0 :     virtual ~ZookeeperCBindings() {
      36           0 :     }
      37           0 :     virtual void ZooSetDebugLevel(ZooLogLevel logLevel) {
      38           0 :         zoo_set_debug_level(logLevel);
      39           0 :     }
      40           0 :     virtual zhandle_t* ZookeeperInit(const char *host, watcher_fn fn,
      41             :         int recv_timeout, const clientid_t *clientid, void *context,
      42             :         int flags) {
      43           0 :         return zookeeper_init(host, fn, recv_timeout, clientid, context,
      44           0 :             flags);
      45             :     }
      46           0 :     virtual int ZookeeperClose(zhandle_t *zh) {
      47           0 :         return zookeeper_close(zh);
      48             :     }
      49           0 :     virtual int ZooState(zhandle_t *zh) {
      50           0 :         return zoo_state(zh);
      51             :     }
      52           0 :     virtual int ZooCreate(zhandle_t *zh, const char *path, const char *value,
      53             :         int valuelen, const struct ACL_vector *acl, int flags,
      54             :         char *path_buffer, int path_buffer_len) {
      55           0 :         return zoo_create(zh, path, value, valuelen, acl, flags, path_buffer,
      56           0 :             path_buffer_len);
      57             :     }
      58           0 :     virtual int ZooDelete(zhandle_t *zh, const char *path, int version) {
      59           0 :         return zoo_delete(zh, path, version);
      60             :     }
      61           0 :     virtual int ZooGet(zhandle_t *zh, const char *path, int watch,
      62             :         char *buffer, int* buffer_len, struct Stat *stat) {
      63           0 :         return zoo_get(zh, path, watch, buffer, buffer_len, stat);
      64             :     }
      65           0 :     virtual int ZooExists(zhandle_t *zh, const char *path, int watch,
      66             :                                          struct Stat *stat) {
      67           0 :         return zoo_exists(zh, path, watch, stat);
      68             :     }
      69           0 :     virtual void ZooSetContext(zhandle_t * zh, void *context) {
      70           0 :         return zoo_set_context(zh, context);
      71             :     }
      72           0 :     virtual int ZooIsUnrecoverable(zhandle_t * zh) {
      73           0 :         return is_unrecoverable(zh);
      74             :     }
      75             : };
      76             : 
      77             : } // namespace interface
      78             : 
      79             : namespace client {
      80             : namespace impl {
      81             : 
      82           0 : void ZookeeperWatcher(zhandle_t* zh, int type, int state,
      83             :         const char* path, void* watcherCtx) {
      84           0 :     if (ZINVALIDSTATE == is_unrecoverable(zh)) {
      85           0 :         ZOO_LOG(DEBUG, "Zookeeper callback called with state " << state);
      86           0 :         ZookeeperClientImpl *zooImpl = (ZookeeperClientImpl *)watcherCtx;
      87           0 :         if (zooImpl && zooImpl->GetClient()) {
      88           0 :             if (((ZookeeperClient *)zooImpl->GetClient())->cb) {
      89           0 :                 ((ZookeeperClient *)zooImpl->GetClient())->cb();
      90             :             }
      91             :         }
      92             :     }
      93           0 : }
      94             : 
      95             : // ZookeeperClientImpl
      96           6 : ZookeeperClientImpl::ZookeeperClientImpl(const char *hostname,
      97           6 :     const char *servers, zookeeper::interface::ZookeeperInterface *zki) :
      98           6 :     hostname_(hostname),
      99           6 :     servers_(servers),
     100           6 :     zk_handle_(NULL),
     101           6 :     connected_(false),
     102          12 :     zki_(zki) {
     103             :     // Set loglevel
     104           6 :     zki_->ZooSetDebugLevel(ZOO_LOG_LEVEL_DEBUG);
     105           6 : }
     106             : 
     107          12 : ZookeeperClientImpl::~ZookeeperClientImpl() {
     108          12 : }
     109             : 
     110           7 : bool ZookeeperClientImpl::Connect() {
     111             :     while (true) {
     112           7 :         zk_handle_ = zki_->ZookeeperInit(servers_.c_str(),
     113             :                                          ZookeeperWatcher,
     114             :                                          kSessionTimeoutMSec_,
     115             :                                          NULL,
     116             :                                          NULL,
     117             :                                          0);
     118           7 :         if (zk_handle_ == NULL) {
     119           0 :             int zerrno(errno);
     120           0 :             ZOO_LOG_ERR("zookeeper_init FAILED: (" << zerrno <<
     121             :                 "): servers: " << servers_ << " retrying in 1 second");
     122           0 :             sleep(1);
     123           0 :             continue;
     124           0 :         }
     125           7 :         zki_->ZooSetContext(zk_handle_, this);
     126             :         // Block till session is connected
     127           8 :         while (!connected_) {
     128           8 :             int zstate(zki_->ZooState(zk_handle_));
     129           8 :             if (zstate == ZOO_CONNECTED_STATE) {
     130           7 :                 connected_ = true;
     131           7 :                 ZOO_LOG(DEBUG, "Session CONNECTED");
     132           7 :                 break;
     133             :             } else {
     134           1 :                 ZOO_LOG(DEBUG, "Session NOT CONNECTED: retrying in 1 second");
     135           1 :                 sleep(1);
     136           1 :                 continue;
     137           1 :             }
     138             :         }
     139           7 :         break;
     140           0 :     }
     141           7 :     assert(connected_);
     142           7 :     return true;
     143             : }
     144             : 
     145           7 : void ZookeeperClientImpl::Shutdown() {
     146           7 :     if (zk_handle_) {
     147           7 :         int rc(zki_->ZookeeperClose(zk_handle_));
     148           7 :         if (rc != ZOK) {
     149           0 :             int zerrno(errno);
     150           0 :             ZOO_LOG(WARN, "zookeeper_close FAILED (" << rc <<
     151             :                 "): errno: " << zerrno);
     152             :         }
     153           7 :         zk_handle_ = NULL;
     154             :     }
     155           7 :     connected_ = false;
     156           7 : }
     157             : 
     158           1 : bool ZookeeperClientImpl::Reconnect() {
     159           1 :     Shutdown();
     160           1 :     return Connect();
     161             : }
     162             : 
     163          19 : bool ZookeeperClientImpl::IsConnected() const {
     164          19 :     return connected_;
     165             : }
     166             : 
     167          20 : static inline bool IsZooErrorRecoverable(int zerror) {
     168          20 :     return zerror == ZCONNECTIONLOSS ||
     169          20 :         zerror == ZOPERATIONTIMEOUT;
     170             : }
     171             : 
     172          16 : static inline bool IsZooErrorUnrecoverable(int zerror) {
     173          16 :     return zerror == ZINVALIDSTATE;
     174             : }
     175             : 
     176           8 : int ZookeeperClientImpl::CreateNodeSync(const char *path, const char *value,
     177             :     int *err, int flag) {
     178             :     int rc;
     179           8 :  retry:
     180             :     do {
     181          10 :         rc = zki_->ZooCreate(zk_handle_, path, value, strlen(value),
     182             :             &ZOO_OPEN_ACL_UNSAFE, flag, NULL, -1);
     183          10 :     } while (IsZooErrorRecoverable(rc));
     184           8 :     if (IsZooErrorUnrecoverable(rc)) {
     185             :         // Reconnect
     186           1 :         Reconnect();
     187           1 :         goto retry;
     188             :     }
     189           7 :     if (rc != ZOK) {
     190           2 :         *err = errno;
     191             :     }
     192           7 :     return rc;
     193             : }
     194             : 
     195           0 : bool ZookeeperClientImpl::CreateNode(const char *path, const char *value,
     196             :                                      int flag) {
     197           0 :     int err = 0;
     198             :     int rc;
     199           0 :     if (!IsConnected()) {
     200           0 :         bool success(Connect());
     201           0 :         if (!success) {
     202           0 :             ZOO_LOG_ERR("Zookeeper Client Connect FAILED");
     203           0 :             return false;
     204             :         }
     205             :     }
     206           0 :     rc = CreateNodeSync(path, value, &err, flag);
     207           0 :     if (rc != ZOK && rc != ZNODEEXISTS) {
     208           0 :         ZOO_LOG_ERR("Creation of ZNODE(" << path << "): " << value
     209             :                     << ": FAILED: (" << rc << ") error: " << err);
     210           0 :         return false;
     211             :     }
     212           0 :     return true;
     213             : }
     214             : 
     215           2 : int ZookeeperClientImpl::GetNodeDataSync(const char *path, char *buf,
     216             :     int *buf_len, int *err) {
     217             :     int rc;
     218           2 :  retry:
     219             :     do {
     220           2 :         rc = zki_->ZooGet(zk_handle_, path, 0, buf, buf_len, NULL);
     221           2 :     } while (IsZooErrorRecoverable(rc));
     222           2 :     if (IsZooErrorUnrecoverable(rc)) {
     223             :         // Reconnect
     224           0 :         Reconnect();
     225           0 :         goto retry;
     226             :     }
     227           2 :     if (rc != ZOK) {
     228           0 :         *err = errno;
     229             :     }
     230           2 :     return rc;
     231             : }
     232             : 
     233           0 : bool ZookeeperClientImpl::CheckNodeExist(const char *path) {
     234           0 :     if (!IsConnected()) {
     235           0 :         bool success(Connect());
     236           0 :         if (!success) {
     237           0 :             ZOO_LOG_ERR("Zookeeper Client Connect FAILED");
     238           0 :             return false;
     239             :         }
     240             :     }
     241             : 
     242             :     struct Stat stat;
     243           0 :     int rc = zki_->ZooExists(zk_handle_, path, 0, &stat);
     244           0 :     return (rc == ZOK);
     245             : }
     246             : 
     247           6 : int ZookeeperClientImpl::DeleteNodeSync(const char *path, int *err) {
     248             :     int rc;
     249           6 :  retry:
     250             :     do {
     251           8 :         rc = zki_->ZooDelete(zk_handle_, path, -1);
     252           8 :     } while (IsZooErrorRecoverable(rc));
     253           6 :     if (IsZooErrorUnrecoverable(rc)) {
     254             :         // Reconnect
     255           0 :         Reconnect();
     256           0 :         goto retry;
     257             :     }
     258           6 :     if (rc != ZOK) {
     259           0 :         *err = errno;
     260             :     }
     261           6 :     return rc;
     262             : }
     263             : 
     264           0 : bool ZookeeperClientImpl::DeleteNode(const char *path) {
     265           0 :     int err = 0;
     266             :     int rc;
     267           0 :     if (!IsConnected()) {
     268           0 :         bool success(Connect());
     269           0 :         if (!success) {
     270           0 :             ZOO_LOG_ERR("Zookeeper Client Connect FAILED");
     271           0 :             return false;
     272             :         }
     273             :     }
     274             : 
     275           0 :     rc = DeleteNodeSync(path, &err);
     276           0 :     if (rc != ZOK) {
     277           0 :         ZOO_LOG_ERR("Deletion of ZNODE(" << path << "): "
     278             :                     << ": FAILED: (" << rc << ") error: " << err);
     279           0 :         return false;
     280             :     }
     281             : 
     282           0 :     return (rc == ZOK);
     283             : }
     284             : 
     285           6 : std::string ZookeeperClientImpl::Name() const {
     286           6 :     return hostname_;
     287             : }
     288             : 
     289             : } // namespace impl
     290             : 
     291             : // ZookeeperClient
     292           0 : ZookeeperClient::ZookeeperClient(const char *hostname, const char *servers) :
     293           0 :     impl_(new impl::ZookeeperClientImpl(hostname, servers,
     294           0 :         new zookeeper::interface::ZookeeperCBindings)) {
     295           0 : }
     296             : 
     297           6 : ZookeeperClient::ZookeeperClient(impl::ZookeeperClientImpl *impl) :
     298           6 :     impl_(impl) {
     299           6 : }
     300             : 
     301           0 : bool ZookeeperClient::CreateNode(const char *path, const char *value,
     302             :                                  int type) {
     303           0 :     int flag = 0;
     304           0 :     if (type == Z_NODE_TYPE_EPHEMERAL) {
     305           0 :         flag |= ZOO_EPHEMERAL;
     306             :     }
     307           0 :     if (type == Z_NODE_TYPE_SEQUENCE) {
     308           0 :         flag |= ZOO_SEQUENCE;
     309             :     }
     310           0 :     return impl_->CreateNode(path, value, flag);
     311             : }
     312             : 
     313           0 : bool ZookeeperClient::CheckNodeExist(const char *path) {
     314           0 :     return impl_->CheckNodeExist(path);
     315             : }
     316             : 
     317           0 : bool ZookeeperClient::DeleteNode(const char *path) {
     318           0 :     return impl_->DeleteNode(path);
     319             : }
     320             : 
     321           0 : void ZookeeperClient::Shutdown() {
     322           0 :     return impl_->Shutdown();
     323             : }
     324             : 
     325           0 : void ZookeeperClient::AddListener(ZooStateCallback callback) {
     326           0 :     cb = callback;
     327           0 :     if (impl_.get()) {
     328           0 :         impl_->SetClient(this);
     329             :     }
     330           0 : }
     331             : 
     332          12 : ZookeeperClient::~ZookeeperClient() {
     333          12 : }
     334             : 
     335             : // ZookeeperLockImpl
     336             : class ZookeeperLock::ZookeeperLockImpl {
     337             :  public:
     338           6 :     ZookeeperLockImpl(impl::ZookeeperClientImpl *clientImpl,
     339           6 :         const char *path) :
     340           6 :         clientImpl_(clientImpl),
     341           6 :         path_(path),
     342           6 :         is_acquired_(false) {
     343           6 :         id_ = clientImpl_->Name();
     344           6 :     }
     345             : 
     346           6 :     std::string Id() const {
     347           6 :         return id_;
     348             :     }
     349             : 
     350           6 :     bool Lock() {
     351           6 :         ZOO_LOG(INFO, "Trying (" << path_ << "): " << id_);
     352             :         while (true) {
     353             :             // Connect if not already done
     354           7 :             if (!clientImpl_->IsConnected()) {
     355           6 :                 bool success(clientImpl_->Connect());
     356           6 :                 if (!success) {
     357           0 :                     ZOO_LOG_ERR("Zookeeper Client Connect FAILED");
     358           6 :                     return success;
     359             :                 }
     360             :             }
     361             :             // Try creating the znode
     362             :             int err;
     363           7 :             int rc(clientImpl_->CreateNodeSync(path_.c_str(), id_.c_str(),
     364             :                 &err, 0));
     365           7 :             switch (rc) {
     366           5 :               case ZOK: {
     367             :                 // We acquired the lock
     368           5 :                 ZOO_LOG(INFO, "ACQUIRED (" << path_ << "): " << id_);
     369           5 :                 is_acquired_ = true;
     370           5 :                 return true;
     371             :               }
     372           2 :               case ZNODEEXISTS: {
     373             :                 // Node exists, get node data and check
     374             :                 char buf[256];
     375           2 :                 int buf_len(sizeof(buf));
     376             :                 int zerr;
     377           2 :                 int zrc(clientImpl_->GetNodeDataSync(path_.c_str(), buf,
     378             :                     &buf_len, &zerr));
     379           2 :                 if (zrc == ZOK) {
     380             :                     // Does it match our ID?
     381           2 :                     std::string mid(buf, buf_len);
     382           2 :                     if (id_ == mid) {
     383             :                         // We acquired the lock
     384           1 :                         ZOO_LOG(INFO, "ACQUIRED EEXIST (" << path_ << "): "
     385             :                             << id_);
     386           1 :                         is_acquired_ = true;
     387           1 :                         return true;
     388             :                     }
     389           1 :                     ZOO_LOG_ERR("EEXIST (" << path_ << "): " << mid <<
     390             :                         " , ours: " << id_);
     391           1 :                     sleep(1);
     392           1 :                     continue;
     393           2 :                 } else if (zrc == ZNONODE) {
     394           0 :                     ZOO_LOG(WARN, "GetNodeDataSync(" << path_ <<
     395             :                         "): Data: " << id_ <<
     396             :                         ": No Node EXISTS: retrying in 1 second");
     397           0 :                     sleep(1);
     398           0 :                     continue;
     399           0 :                 } else {
     400           0 :                     ZOO_LOG_ERR("GetNodeDataSync(" << path_ << "): " <<
     401             :                         id_ << ": FAILED: (" << zrc << ") error: " << zerr);
     402           0 :                     clientImpl_->Shutdown();
     403           0 :                     return false;
     404             :                 }
     405             :                 break;
     406             :               }
     407           0 :               default: {
     408           0 :                 ZOO_LOG_ERR("CreateNodeSync(" << path_ << "): " << id_
     409             :                     << ": FAILED: (" << rc << ") error: " << err);
     410           0 :                 clientImpl_->Shutdown();
     411           0 :                 return false;
     412             :               }
     413             :             }
     414           1 :         }
     415             :     }
     416             : 
     417           6 :     bool Release() {
     418             :         bool success;
     419             :         int err, rc;
     420           6 :         if (!is_acquired_) {
     421           0 :             ZOO_LOG_ERR("(" << path_ << "): " << id_ <<
     422             :                 ": Release WITHOUT Lock");
     423           0 :             success = false;
     424           0 :             goto cleanup;
     425             :         }
     426           6 :         is_acquired_ = false;
     427           6 :         rc = clientImpl_->DeleteNodeSync(path_.c_str(), &err);
     428           6 :         if (rc == ZOK) {
     429           6 :             ZOO_LOG(INFO, "RELEASED (" << path_ << "): " << id_);
     430           6 :             success = true;
     431           6 :             goto cleanup;
     432           0 :         } else if (rc == ZNONODE) {
     433           0 :             ZOO_LOG_ERR("DeleteNodeSync(" << path_ << "): " << id_ <<
     434             :                 ": No Node EXISTS(" << err <<
     435             :                 "): Possible concurrent execution");
     436           0 :             success = false;
     437           0 :             goto cleanup;
     438             :         } else {
     439           0 :             ZOO_LOG_ERR("DeleteNodeSync(" << path_ << "): " << id_ <<
     440             :                 ": FAILED (" << rc << "): error " << err);
     441           0 :             success = false;
     442           0 :             goto cleanup;
     443             :         }
     444           6 :      cleanup:
     445           6 :         clientImpl_->Shutdown();
     446           6 :         return success;
     447             :     }
     448             : 
     449             :  private:
     450             :     impl::ZookeeperClientImpl *clientImpl_;
     451             :     std::string path_;
     452             :     bool is_acquired_;
     453             :     std::string id_;
     454             : };
     455             : 
     456             : // ZookeeperLock
     457           6 : ZookeeperLock::ZookeeperLock(ZookeeperClient *client, const char *path) :
     458           6 :     impl_(new ZookeeperLockImpl(client->impl_.get(), path)) {
     459           6 : }
     460             : 
     461           6 : ZookeeperLock::~ZookeeperLock() {
     462           6 : }
     463             : 
     464           6 : std::string ZookeeperLock::Id() const {
     465           6 :     return impl_->Id();
     466             : }
     467             : 
     468           6 : bool ZookeeperLock::Lock() {
     469           6 :     return impl_->Lock();
     470             : }
     471             : 
     472           6 : bool ZookeeperLock::Release() {
     473           6 :     return impl_->Release();
     474             : }
     475             : 
     476             : } // namespace client
     477             : } // namespace zookeeper