Implement internal restart / run_forever in monitor

mon/etcd_adapter.js

@@ -4,6 +4,8 @@
 const http = require('http');
 const WebSocket = require('ws');
 
+const MON_STOPPED = 'Monitor instance is stopped';
+
 class EtcdAdapter
 {
     constructor(mon)
@@ -66,10 +68,12 @@ class EtcdAdapter
         return this.selected_etcd_url;
     }
 
-    restart_watcher(cur_addr)
+    stop_watcher(cur_addr)
     {
+        cur_addr = cur_addr || this.selected_etcd_url;
         if (this.ws)
         {
+            console.log('Disconnected from etcd at '+this.ws_used_url);
             this.ws.close();
             this.ws = null;
         }
@@ -82,6 +86,11 @@ class EtcdAdapter
         {
             this.selected_etcd_url = null;
         }
+    }
+
+    restart_watcher(cur_addr)
+    {
+        this.stop_watcher(cur_addr);
         this.start_watcher(this.mon.config.etcd_mon_retries).catch(this.mon.die);
     }
 
@@ -104,15 +113,24 @@ class EtcdAdapter
                 now = Date.now();
             }
             tried[base] = now;
+            if (this.mon.stopped)
+            {
+                return;
+            }
             const ok = await new Promise(ok =>
             {
                 const timer_id = setTimeout(() =>
                 {
-                    this.ws.close();
+                    if (this.ws)
-                    this.ws = null;
+                    {
+                        console.log('Disconnected from etcd at '+this.ws_used_url);
+                        this.ws.close();
+                        this.ws = null;
+                    }
                     ok(false);
                 }, this.mon.config.etcd_mon_timeout);
                 this.ws = new WebSocket(base+'/watch');
+                this.ws_used_url = cur_addr;
                 const fail = () =>
                 {
                     ok(false);
@@ -135,13 +153,19 @@ class EtcdAdapter
         }
         if (!this.ws)
         {
-            this.mon.failconnect('Failed to open etcd watch websocket');
+            this.mon.die('Failed to open etcd watch websocket');
+            return;
+        }
+        if (this.mon.stopped)
+        {
+            this.stop_watcher();
+            return;
         }
         const cur_addr = this.selected_etcd_url;
         this.ws_alive = true;
         this.ws_keepalive_timer = setInterval(() =>
         {
-            if (this.ws_alive)
+            if (this.ws_alive && this.ws)
             {
                 this.ws_alive = false;
                 this.ws.send(JSON.stringify({ progress_request: {} }));
@@ -164,6 +188,11 @@ class EtcdAdapter
         }));
         this.ws.on('message', (msg) =>
         {
+            if (this.mon.stopped)
+            {
+                this.stop_watcher();
+                return;
+            }
             this.ws_alive = true;
             let data;
             try
@@ -183,15 +212,14 @@ class EtcdAdapter
                 if (data.result.compact_revision)
                 {
                     // we may miss events if we proceed
-                    console.error('Revisions before '+data.result.compact_revision+' were compacted by etcd, exiting');
+                    this.mon.die('Revisions before '+data.result.compact_revision+' were compacted by etcd, exiting');
-                    this.mon.on_stop(1);
                 }
-                console.error('Watch canceled by etcd, reason: '+data.result.cancel_reason+', exiting');
+                this.mon.die('Watch canceled by etcd, reason: '+data.result.cancel_reason+', exiting');
-                this.mon.on_stop(1);
             }
             else if (data.result.created)
             {
                 // etcd watch created
+                console.log('Successfully subscribed to etcd at '+this.selected_etcd_url+', revision '+data.result.header.revision);
             }
             else
             {
@@ -239,25 +267,33 @@ class EtcdAdapter
                 now = Date.now();
             }
             tried[base] = now;
+            if (this.mon.stopped)
+            {
+                throw new Error(MON_STOPPED);
+            }
             const res = await POST(base+path, body, timeout);
+            if (this.mon.stopped)
+            {
+                throw new Error(MON_STOPPED);
+            }
             if (res.error)
             {
                 if (this.selected_etcd_url == base)
                     this.selected_etcd_url = null;
-                console.error('failed to query etcd: '+res.error);
+                console.error('Failed to query etcd '+path+' (retry '+retry+'/'+retries+'): '+res.error);
                 continue;
             }
             if (res.json)
             {
                 if (res.json.error)
                 {
-                    console.error('etcd returned error: '+res.json.error);
+                    console.error(path+': etcd returned error: '+res.json.error);
                     break;
                 }
                 return res.json;
             }
         }
-        this.mon.failconnect();
+        throw new Error('Failed to query etcd ('+retries+' retries)');
     }
 }
 

mon/mon-main.js

@@ -23,4 +23,4 @@ for (let i = 2; i < process.argv.length; i++)
     }
 }
 
-new Mon(options).start().catch(e => { console.error(e); process.exit(1); });
+Mon.run_forever(options);

mon/mon.js

@@ -15,10 +15,38 @@ const { recheck_primary, save_new_pgs_txn, generate_pool_pgs } = require('./pg_g
 
 class Mon
 {
+    static run_forever(config)
+    {
+        let mon;
+        const run = () =>
+        {
+            console.log('Starting Monitor');
+            const my_mon = new Mon(config);
+            mon = my_mon;
+            my_mon.on_die = () =>
+            {
+                if (mon == my_mon)
+                {
+                    // Start a new instance
+                    run();
+                }
+            };
+            my_mon.start().catch(my_mon.die);
+        };
+        run();
+        const on_stop_cb = () => mon.on_stop().then(() => process.exit(0)).catch(err =>
+        {
+            console.error(err);
+            process.exit(0);
+        });
+        process.on('SIGINT', on_stop_cb);
+        process.on('SIGTERM', on_stop_cb);
+    }
+
     constructor(config)
     {
-        this.failconnect = (e) => this._die(e, 2);
+        this.stopped = false;
-        this.die = (e) => this._die(e, 1);
+        this.die = (e) => this._die(e);
         this.fileConfig = {};
         if (fs.existsSync(config.config_path||'/etc/vitastor/vitastor.conf'))
         {
@@ -29,8 +57,6 @@ class Mon
         this.check_config();
         this.state = JSON.parse(JSON.stringify(etcd_tree));
         this.prev_stats = { osd_stats: {}, osd_diff: {} };
-        this.signals_set = false;
-        this.on_stop_cb = () => this.on_stop(0).catch(console.error);
         this.recheck_pgs_active = false;
         this.etcd = new EtcdAdapter(this);
         this.etcd.parse_config(this.config);
@@ -162,6 +188,10 @@ class Mon
     // Schedule save_last_clean() to to run after a small timeout (1s) (to not spam etcd)
     schedule_save_last_clean()
     {
+        if (this.stopped)
+        {
+            return;
+        }
         if (!this.save_last_clean_timer)
         {
             this.save_last_clean_timer = setTimeout(() =>
@@ -239,27 +269,60 @@ class Mon
             lease: ''+this.etcd_lease_id
         }, this.config.etcd_start_timeout, 0);
         // Set refresh timer
-        this.lease_timer = setInterval(async () =>
+        this.lease_timer = setInterval(() =>
         {
-            const res = await this.etcd.etcd_call('/lease/keepalive', { ID: this.etcd_lease_id }, this.config.etcd_mon_timeout, this.config.etcd_mon_retries);
+            this.etcd.etcd_call('/lease/keepalive', { ID: this.etcd_lease_id }, this.config.etcd_mon_timeout, this.config.etcd_mon_retries)
-            if (!res.result.TTL)
+                .then(res =>
-            {
+                {
-                this.failconnect('Lease expired');
+                    if (!res.result.TTL)
-            }
+                        this.die('Lease expired');
+                })
+                .catch(this.die);
         }, this.config.etcd_mon_ttl*1000);
-        if (!this.signals_set)
-        {
-            process.on('SIGINT', this.on_stop_cb);
-            process.on('SIGTERM', this.on_stop_cb);
-            this.signals_set = true;
-        }
     }
 
-    async on_stop(status)
+    async on_stop()
     {
-        clearInterval(this.lease_timer);
+        console.log('Stopping Monitor');
-        await this.etcd.etcd_call('/lease/revoke', { ID: this.etcd_lease_id }, this.config.etcd_mon_timeout, this.config.etcd_mon_retries);
+        this.etcd.stop_watcher();
-        process.exit(status);
+        if (this.save_last_clean_timer)
+        {
+            clearTimeout(this.save_last_clean_timer);
+            this.save_last_clean_timer = null;
+        }
+        if (this.next_recheck_timer)
+        {
+            clearTimeout(this.next_recheck_timer);
+            this.next_recheck_timer = null;
+        }
+        if (this.recheck_timer)
+        {
+            clearTimeout(this.recheck_timer);
+            this.recheck_timer = null;
+        }
+        if (this.stats_timer)
+        {
+            clearTimeout(this.stats_timer);
+            this.stats_timer = null;
+        }
+        if (this.lease_timer)
+        {
+            clearInterval(this.lease_timer);
+            this.lease_timer = null;
+        }
+        let p = null;
+        if (this.etcd_lease_id)
+        {
+            const lease_id = this.etcd_lease_id;
+            this.etcd_lease_id = null;
+            p = this.etcd.etcd_call('/lease/revoke', { ID: lease_id }, this.config.etcd_mon_timeout, this.config.etcd_mon_retries);
+        }
+        // 'stopped' flag prevents all further etcd communications of this instance
+        this.stopped = true;
+        if (p)
+        {
+            await p;
+        }
     }
 
     async load_cluster_state()
@@ -333,6 +396,10 @@ class Mon
 
     async recheck_pgs()
     {
+        if (this.stopped)
+        {
+            return;
+        }
         if (this.recheck_pgs_active)
         {
             this.schedule_recheck();
@@ -502,6 +569,10 @@ class Mon
     // Schedule next recheck at least at <unixtime>
     schedule_next_recheck_at(unixtime)
     {
+        if (this.stopped)
+        {
+            return;
+        }
         this.next_recheck_at = !this.next_recheck_at || this.next_recheck_at > unixtime
             ? unixtime : this.next_recheck_at;
         const now = Date.now()/1000;
@@ -530,6 +601,10 @@ class Mon
     // This is required for multiple change events to trigger at most 1 recheck in 1s
     schedule_recheck()
     {
+        if (this.stopped)
+        {
+            return;
+        }
         if (!this.recheck_timer)
         {
             this.recheck_timer = setTimeout(() =>
@@ -600,7 +675,7 @@ class Mon
 
     schedule_update_stats()
     {
-        if (this.stats_timer)
+        if (this.stopped || this.stats_timer)
         {
             return;
         }
@@ -684,11 +759,12 @@ class Mon
         }
     }
 
-    _die(err, code)
+    _die(err)
     {
-        // In fact we can just try to rejoin
+        // Stop this instance of Monitor so we can restart
         console.error(err instanceof Error ? err : new Error(err || 'Cluster connection failed'));
-        process.exit(code || 2);
+        this.on_stop().catch(console.error);
+        this.on_die();
     }
 
     local_ips(all)

tests/run_3osds.sh

@@ -54,7 +54,7 @@ for i in $(seq 1 $OSD_COUNT); do
     start_osd $i
 done
 
-(while true; do set +e; node mon/mon-main.js --etcd_address $ETCD_URL --etcd_prefix "/vitastor" --verbose 1; if [[ $? -ne 2 ]]; then break; fi; done) >>./testdata/mon.log 2>&1 &
+node mon/mon-main.js --etcd_address $ETCD_URL --etcd_prefix "/vitastor" --verbose 1 >>./testdata/mon.log 2>&1 &
 MON_PID=$!
 
 if [ "$SCHEME" = "ec" ]; then

tests/test_vm_cont.sh

@@ -15,7 +15,7 @@ for i in $(seq 1 $OSD_COUNT); do
     eval OSD${i}_PID=$!
 done
 
-(while true; do node mon/mon-main.js --etcd_address $ETCD_URL --etcd_prefix "/vitastor" --verbose 1 || true; done) >>./testdata/mon.log 2>&1 &
+node mon/mon-main.js --etcd_address $ETCD_URL --etcd_prefix "/vitastor" --verbose 1 >>./testdata/mon.log 2>&1 &
 MON_PID=$!
 
 sleep 3