Merge remote-tracking branch 'origin/bugfix/BB-466/zookeeper-expired-…

…session' into w/7.70/bugfix/BB-466/zookeeper-expired-session

extensions/lifecycle/conductor/LifecycleConductor.js

@@ -10,7 +10,7 @@ const BucketClient = require('bucketclient').RESTClient;
 
 const BackbeatProducer = require('../../../lib/BackbeatProducer');
 const BackbeatTask = require('../../../lib/tasks/BackbeatTask');
-const zookeeperHelper = require('../../../lib/clients/zookeeper');
+const ZookeeperManager = require('../../../lib/clients/ZookeeperManager');
 const KafkaBacklogMetrics = require('../../../lib/KafkaBacklogMetrics');
 const { authTypeAssumeRole } = require('../../../lib/constants');
 const VaultClientCache = require('../../../lib/clients/VaultClientCache');
@@ -604,9 +604,7 @@ class LifecycleConductor {
             process.nextTick(cb);
             return;
         }
-        this._zkClient = zookeeperHelper.createClient(
-            this.zkConfig.connectionString, this.zkConfig);
-        this._zkClient.connect();
+        this._zkClient = new ZookeeperManager(this.zkConfig.connectionString, this.zkConfig, this.logger);
         this._zkClient.once('error', cb);
         this._zkClient.once('ready', () => {
             // just in case there would be more 'error' events
@@ -670,7 +668,7 @@ class LifecycleConductor {
             // just in case there would be more 'error' events emitted
             this._kafkaBacklogMetrics.removeAllListeners('error');
             this._kafkaBacklogMetrics.on('error', err => {
-                this._log.error('error from kafka topic metrics', {
+                this.logger.error('error from kafka topic metrics', {
                     error: err.message,
                     method: 'LifecycleConductor._initKafkaBacklogMetrics',
                 });

extensions/notification/queueProcessor/QueueProcessor.js

@@ -7,7 +7,7 @@ const errors = require('arsenal').errors;
 
 const BackbeatConsumer = require('../../../lib/BackbeatConsumer');
 const NotificationDestination = require('../destination');
-const zookeeper = require('../../../lib/clients/zookeeper');
+const ZookeeperManager = require('../../../lib/clients/ZookeeperManager');
 const configUtil = require('../utils/config');
 const messageUtil = require('../utils/message');
 const NotificationConfigManager = require('../NotificationConfigManager');
@@ -72,10 +72,10 @@ class QueueProcessor extends EventEmitter {
             `${this.zkConfig.connectionString}${populatorZkPath}`;
         this.logger.info('opening zookeeper connection for reading ' +
             'bucket notification configuration', { zookeeperUrl });
-        this.zkClient = zookeeper.createClient(zookeeperUrl, {
+        this.zkClient = new ZookeeperManager(zookeeperUrl, {
             autoCreateNamespace: this.zkConfig.autoCreateNamespace,
-        });
-        this.zkClient.connect();
+        }, this.logger);
+
         this.zkClient.once('error', done);
         this.zkClient.once('ready', () => {
             // just in case there would be more 'error' events emitted

lib/KafkaBacklogMetrics.js

@@ -5,7 +5,7 @@ const zookeeper = require('node-zookeeper-client');
 const Logger = require('werelogs').Logger;
 const { errors, metrics } = require('arsenal');
 
-const zookeeperHelper = require('./clients/zookeeper');
+const ZookeeperManager = require('./clients/ZookeeperManager');
 const { readUInt64BE } = require('./util/buffer');
 const { promMetricNames } = require('./constants').kafkaBacklogMetrics;
 
@@ -58,8 +58,7 @@ class KafkaBacklogMetrics extends EventEmitter {
     }
 
     _initZookeeperClient() {
-        this._zookeeper = zookeeperHelper.createClient(this._zookeeperEndpoint);
-        this._zookeeper.connect();
+        this._zookeeper = new ZookeeperManager(this._zookeeperEndpoint, null, this._log);
         this._zookeeper.on('error', err => {
             this.emit('error', err);
         });

lib/api/BackbeatAPI.js

@@ -1,7 +1,6 @@
 'use strict'; // eslint-disable-line strict
 
 const async = require('async');
-const zookeeper = require('node-zookeeper-client');
 
 const { errors } = require('arsenal');
 const { RedisClient } = require('arsenal').metrics;
@@ -17,6 +16,7 @@ const Healthcheck = require('./Healthcheck');
 const routes = require('./routes');
 const { getSortedSetKey, getSortedSetMember } =
     require('../util/sortedSetHelper');
+const ZookeeperManager = require('../clients/ZookeeperManager');
 
 // StatsClient constant defaults
 // TODO: This should be moved to constants file
@@ -182,15 +182,6 @@ class BackbeatAPI {
             && this._crrStatusProducer.isReady();
     }
 
-    /**
-     * Check if Zookeeper and Producer are connected
-     * @return {boolean} true/false
-     */
-    isConnected() {
-        return this._zkClient.getState().name === 'SYNC_CONNECTED'
-            && this._checkProducersReady();
-    }
-
     /**
      * Get Kafka healthcheck
      * @param {object} details - route details from lib/api/routes.js
@@ -1079,13 +1070,12 @@ class BackbeatAPI {
         const zookeeperUrl =
             `${this._zkConfig.connectionString}${populatorZkPath}`;
 
-        const zkClient = zookeeper.createClient(zookeeperUrl, {
+        const zkClient = new ZookeeperManager(zookeeperUrl, {
             autoCreateNamespace: this._zkConfig.autoCreateNamespace,
-        });
-        zkClient.connect();
+        }, this._logger);
 
         zkClient.once('error', cb);
-        zkClient.once('connected', () => {
+        zkClient.once('ready', () => {
             zkClient.removeAllListeners('error');
             this._zkClient = zkClient;
             return cb();

lib/api/Healthcheck.js

@@ -11,7 +11,7 @@ class Healthcheck {
     /**
      * @constructor
      * @param {object} repConfig - extensions.replication configs
-     * @param {node-zookeeper-client.Client} zkClient - zookeeper client
+     * @param {ZookeeperManager} zkClient - zookeeper client manager
      * @param {BackbeatProducer} crrProducer - producer for CRR topic
      * @param {BackbeatProducer} crrStatusProducer - CRR status producer
      * @param {BackbeatProducer} metricProducer - producer for metric