diff --git a/mon/etcd_adapter.js b/mon/etcd_adapter.js index 3d36e0e3..00c7bd5e 100644 --- a/mon/etcd_adapter.js +++ b/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(); - this.ws = null; + if (this.ws) + { + 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.on_stop(1); + this.mon.die('Revisions before '+data.result.compact_revision+' were compacted by etcd, exiting'); } - console.error('Watch canceled by etcd, reason: '+data.result.cancel_reason+', exiting'); - this.mon.on_stop(1); + this.mon.die('Watch canceled by etcd, reason: '+data.result.cancel_reason+', exiting'); } 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)'); } } diff --git a/mon/mon-main.js b/mon/mon-main.js index ff90b8ce..1e15850c 100755 --- a/mon/mon-main.js +++ b/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); diff --git a/mon/mon.js b/mon/mon.js index 019540d1..ea0fff74 100644 --- a/mon/mon.js +++ b/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.die = (e) => this._die(e, 1); + this.stopped = false; + 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); - if (!res.result.TTL) - { - this.failconnect('Lease expired'); - } + this.etcd.etcd_call('/lease/keepalive', { ID: this.etcd_lease_id }, this.config.etcd_mon_timeout, this.config.etcd_mon_retries) + .then(res => + { + 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); - await this.etcd.etcd_call('/lease/revoke', { ID: this.etcd_lease_id }, this.config.etcd_mon_timeout, this.config.etcd_mon_retries); - process.exit(status); + console.log('Stopping Monitor'); + this.etcd.stop_watcher(); + 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 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) diff --git a/tests/run_3osds.sh b/tests/run_3osds.sh index 8f7db143..40df4b9c 100644 --- a/tests/run_3osds.sh +++ b/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 diff --git a/tests/test_vm_cont.sh b/tests/test_vm_cont.sh index b146ae46..48b6d469 100755 --- a/tests/test_vm_cont.sh +++ b/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