Merge pull request #206 from airbnb/resilient_zk_watcher

Resilient zk watcher

lib/synapse/service_watcher/zookeeper.rb

@@ -127,7 +127,15 @@ def discover
 
       new_backends = []
       @zk.children(@discovery['path'], :watch => true).each do |id|
-        node = @zk.get("#{@discovery['path']}/#{id}")
+        begin
+          node = @zk.get("#{@discovery['path']}/#{id}")
+        rescue ZK::Exceptions::NoNode => e
+          # This can happen when the registry unregisters a service node between
+          # the call to @zk.children and @zk.get(path). ZK does not guarantee
+          # a read to ``get`` of a child returned by ``children`` will succeed
+          log.error("synapse: #{@discovery['path']}/#{id} disappeared before it could be read: #{e}")
+          next
+        end
 
         begin
           # TODO: Do less munging, or refactor out this processing

spec/lib/synapse/service_watcher_zookeeper_spec.rb

@@ -29,11 +29,42 @@
   }
 
   context 'ZookeeperWatcher' do
-    let(:discovery) { { 'method' => 'zookeeper', 'hosts' => 'somehost','path' => 'some/path' } }
+    let(:discovery) { { 'method' => 'zookeeper', 'hosts' => 'somehost', 'path' => 'some/path' } }
+    let(:mock_zk) { double(ZK) }
+    let(:mock_node) do
+      node_double = double()
+      allow(node_double).to receive(:first).and_return(service_data_string)
+      node_double
+    end
+
     subject { Synapse::ServiceWatcher::ZookeeperWatcher.new(config, mock_synapse) }
     it 'decodes data correctly' do
       expect(subject.send(:deserialize_service_instance, service_data_string)).to eql(deserialized_service_data)
     end
+
+    it 'reacts to zk push events' do
+      expect(subject).to receive(:watch)
+      expect(subject).to receive(:discover).and_call_original
+      expect(mock_zk).to receive(:children).with('some/path', {:watch=>true}).and_return(
+        ["test_child_1"]
+      )
+      expect(mock_zk).to receive(:get).with('some/path/test_child_1').and_return(mock_node)
+      subject.instance_variable_set('@zk', mock_zk)
+      expect(subject).to receive(:set_backends).with([service_data.merge({'id' => 1})])
+      subject.send(:watcher_callback).call
+    end
+
+    it 'handles zk consistency issues' do
+      expect(subject).to receive(:watch)
+      expect(subject).to receive(:discover).and_call_original
+      expect(mock_zk).to receive(:children).with('some/path', {:watch=>true}).and_return(
+        ["test_child_1"]
+      )
+      expect(mock_zk).to receive(:get).with('some/path/test_child_1').and_raise(ZK::Exceptions::NoNode)
+      subject.instance_variable_set('@zk', mock_zk)
+      expect(subject).to receive(:set_backends).with([])
+      subject.send(:watcher_callback).call
+    end
   end
 
   context 'ZookeeperDnsWatcher' do