Allow registrations to be manifested on the file system and add 'use_previous_backends' option

README.md

@@ -198,6 +198,10 @@ If you do not list any default servers, no proxy will be created.  The
 `default_servers` will also be used in addition to discovered servers if the
 `keep_default_servers` option is set.
 
+If you do not list any `default_servers`, and all backends for a service
+disappear then the previous known backends will be used.  Disable this behavior
+by unsetting `use_previous_backends`.
+
 #### The `haproxy` Section ####
 
 This section is its own hash, which should contain the following keys:
@@ -335,5 +339,5 @@ end
 3. Implement the `start` and `validate_discovery_opts` methods
 4. Implement whatever additional methods your discovery requires
 
-When your watcher detects a list of new backends, they should be written to `@backends`.
-You should then call `@synapse.configure` to force synapse to update the HAProxy config.
+When your watcher detects a list of new backends, you should call `set_backends` to
+store the new backends and update the HAProxy config.

lib/synapse.rb

@@ -1,6 +1,7 @@
 require "synapse/version"
 require "synapse/service_watcher/base"
 require "synapse/haproxy"
+require "synapse/file_output"
 require "synapse/service_watcher"
 require "synapse/log"
 
@@ -17,9 +18,17 @@ def initialize(opts={})
       raise "specify a list of services to connect in the config" unless opts.has_key?('services')
       @service_watchers = create_service_watchers(opts['services'])
 
-      # create the haproxy object
+      # create objects that need to be notified of service changes
+      @config_generators = []
+      # create the haproxy config generator, this is mandatory
       raise "haproxy config section is missing" unless opts.has_key?('haproxy')
-      @haproxy = Haproxy.new(opts['haproxy'])
+      @config_generators << Haproxy.new(opts['haproxy'])
+
+      # possibly create a file manifestation for services that do not
+      # want to communicate via haproxy, e.g. cassandra
+      if opts.has_key?('file_output')
+        @config_generators << FileOutput.new(opts['file_output'])
+      end
 
       # configuration is initially enabled to configure on first loop
       @config_updated = true
@@ -47,8 +56,10 @@ def run
 
         if @config_updated
           @config_updated = false
-          log.info "synapse: regenerating haproxy config"
-          @haproxy.update_config(@service_watchers)
+          @config_generators.each do |config_generator|
+            log.info "synapse: regenerating #{config_generator.name} config"
+            config_generator.update_config(@service_watchers)
+          end
         else
           sleep 1
         end

lib/synapse/file_output.rb

@@ -0,0 +1,53 @@
+require 'synapse/log'
+require 'fileutils'
+require 'tempfile'
+
+module Synapse
+  class FileOutput
+    include Logging
+    attr_reader :opts, :name
+
+    def initialize(opts)
+      super()
+
+      unless opts.has_key?("output_directory")
+        raise ArgumentError, "flat file generation requires an output_directory key"
+      end
+
+      begin
+        FileUtils.mkdir_p(opts['output_directory'])
+      rescue SystemCallError => err
+        raise ArgumentError, "provided output directory #{opts['output_directory']} is not present or creatable"
+      end
+
+      @opts = opts
+      @name = 'file_output'
+    end
+
+    def update_config(watchers)
+      watchers.each do |watcher|
+        write_backends_to_file(watcher.name, watcher.backends)
+      end
+    end
+
+    def write_backends_to_file(service_name, new_backends)
+      data_path = File.join(@opts['output_directory'], "#{service_name}.json")
+      begin
+        old_backends = JSON.load(File.read(data_path))
+      rescue Errno::ENOENT
+        old_backends = nil
+      end
+
+      if old_backends == new_backends
+        return false
+      else
+        # Atomically write new sevice configuration file
+        temp_path = File.join(@opts['output_directory'],
+                              ".#{service_name}.json.tmp")
+        File.open(temp_path, 'w', 0644) {|f| f.write(new_backends.to_json)}
+        FileUtils.mv(temp_path, data_path)
+        return true
+      end
+    end
+  end
+end

lib/synapse/haproxy.rb

@@ -4,7 +4,7 @@
 module Synapse
   class Haproxy
     include Logging
-    attr_reader :opts
+    attr_reader :opts, :name
 
     # these come from the documentation for haproxy 1.5
     # http://haproxy.1wt.eu/download/1.5/doc/configuration.txt
@@ -43,6 +43,7 @@ class Haproxy
         "option abortonclose",
         "option accept-invalid-http-response",
         "option allbackups",
+        "option allredisp",
         "option checkcache",
         "option forceclose",
         "option forwardfor",
@@ -173,6 +174,7 @@ class Haproxy
         "option accept-invalid-http-request",
         "option accept-invalid-http-response",
         "option allbackups",
+        "option allredisp",
         "option checkcache",
         "option clitcpka",
         "option contstats",
@@ -386,6 +388,7 @@ class Haproxy
         "option accept-invalid-http-request",
         "option accept-invalid-http-response",
         "option allbackups",
+        "option allredisp",
         "option checkcache",
         "option clitcpka",
         "option contstats",
@@ -523,6 +526,7 @@ def initialize(opts)
       end
 
       @opts = opts
+      @name = 'haproxy'
 
       # how to restart haproxy
       @restart_interval = 2

lib/synapse/service_watcher/base.rb

@@ -1,3 +1,4 @@
+require 'set'
 require 'synapse/log'
 
 module Synapse
@@ -42,6 +43,10 @@ def initialize(opts={}, synapse)
 
       @keep_default_servers = opts['keep_default_servers'] || false
 
+      # If there are no default servers and a watcher reports no backends, then
+      # use the previous backends that we already know about.
+      @use_previous_backends = opts.fetch('use_previous_backends', true)
+
       # set a flag used to tell the watchers to exit
       # this is not used in every watcher
       @should_exit = false
@@ -95,13 +100,44 @@ def validate_discovery_opts
     end
 
     def set_backends(new_backends)
-      if @keep_default_servers
-        @backends = @default_servers + new_backends
+      # Aggregate and deduplicate all potential backend service instances.
+      new_backends = (new_backends + (@keep_default_servers ? @default_servers : [])).uniq {|b|
+        [b['host'], b['port'], b.fetch('name', '')]
+      }
+
+      if new_backends.to_set == @backends.to_set
+        return false
+      end
+
+      if new_backends.empty?
+        if @default_servers.empty?
+          if @use_previous_backends
+            # Discard this update
+            log.warn "synapse: no default servers for service #{@name};" \
+              "  using previous backends: #{@backends.inspect}"
+            return false
+          else
+            log.warn "synapse: no default servers for service #{@name} and" \
+              " 'use_previous_backends' is disabled; dropping all backends"
+            @backends.clear
+          end
+        else
+          log.warn "synapse: no backends for service #{@name};" \
+            " using default servers: #{@default_servers.inspect}"
+          @backends = @default_servers
+        end
       else
+        log.info "synapse: discovered #{new_backends.length} backends for service #{@name}"
         @backends = new_backends
       end
+
+      reconfigure!
+
+      return true
     end
 
+    # Subclasses should not invoke this directly; it's only exposed so that it
+    # can be overridden in subclasses.
     def reconfigure!
       @synapse.reconfigure!
     end

lib/synapse/service_watcher/dns.rb

@@ -89,21 +89,7 @@ def configure_backends(servers)
         end
       end
 
-      if new_backends.empty?
-        if @default_servers.empty?
-          log.warn "synapse: no backends and no default servers for service #{@name};" \
-            " using previous backends: #{@backends.inspect}"
-        else
-          log.warn "synapse: no backends for service #{@name};" \
-            " using default servers: #{@default_servers.inspect}"
-          @backends = @default_servers
-        end
-      else
-        log.info "synapse: discovered #{new_backends.length} backends for service #{@name}"
-        set_backends(new_backends)
-      end
-
-      reconfigure!
+      set_backends(new_backends)
     end
   end
 end

lib/synapse/service_watcher/docker.rb

@@ -23,16 +23,10 @@ def validate_discovery_opts
     end
 
     def watch
-      last_containers = []
       until @should_exit
         begin
           start = Time.now
-          current_containers = containers
-          unless last_containers == current_containers
-            last_containers = current_containers
-            configure_backends(last_containers)
-          end
-
+          set_backends(containers)
           sleep_until_next_check(start)
         rescue Exception => e
           log.warn "synapse: error in watcher thread: #{e.inspect}"
@@ -98,23 +92,5 @@ def containers
       log.warn "synapse: error while polling for containers: #{e.inspect}"
       []
     end
-
-    def configure_backends(new_backends)
-      if new_backends.empty?
-        if @default_servers.empty?
-          log.warn "synapse: no backends and no default servers for service #{@name};" \
-            " using previous backends: #{@backends.inspect}"
-        else
-          log.warn "synapse: no backends for service #{@name};" \
-            " using default servers: #{@default_servers.inspect}"
-          @backends = @default_servers
-        end
-      else
-        log.info "synapse: discovered #{new_backends.length} backends for service #{@name}"
-        set_backends(new_backends)
-      end
-      reconfigure!
-    end
-
   end
 end

lib/synapse/service_watcher/ec2tag.rb

@@ -54,20 +54,12 @@ def validate_discovery_opts
     end
 
     def watch
-      last_backends = []
       until @should_exit
         begin
           start = Time.now
-          current_backends = discover_instances
-
-          if last_backends != current_backends
+          if set_backends(discover_instances)
             log.info "synapse: ec2tag watcher backends have changed."
-            last_backends = current_backends
-            configure_backends(current_backends)
-          else
-            log.info "synapse: ec2tag watcher backends are unchanged."
           end
-
           sleep_until_next_check(start)
         rescue Exception => e
           log.warn "synapse: error in ec2tag watcher thread: #{e.inspect}"
@@ -111,23 +103,6 @@ def instances_with_tags(tag_name, tag_value)
         .tagged_values(tag_value)
         .select { |i| i.status == :running }
     end
-
-    def configure_backends(new_backends)
-      if new_backends.empty?
-        if @default_servers.empty?
-          log.warn "synapse: no backends and no default servers for service #{@name};" \
-            " using previous backends: #{@backends.inspect}"
-        else
-          log.warn "synapse: no backends for service #{@name};" \
-            " using default servers: #{@default_servers.inspect}"
-          @backends = @default_servers
-        end
-      else
-        log.info "synapse: discovered #{new_backends.length} backends for service #{@name}"
-        @backends = new_backends
-      end
-      @synapse.reconfigure!
-    end
   end
 end
 

lib/synapse/service_watcher/zookeeper.rb

@@ -45,7 +45,7 @@ def create(path)
       @zk.create(path, ignore: :node_exists)
     end
 
-    # find the current backends at the discovery path; sets @backends
+    # find the current backends at the discovery path
     def discover
       log.info "synapse: discovering backends for service #{@name}"
 
@@ -69,17 +69,7 @@ def discover
         end
       end
 
-      if new_backends.empty?
-        if @default_servers.empty?
-          log.warn "synapse: no backends and no default servers for service #{@name}; using previous backends: #{@backends.inspect}"
-        else
-          log.warn "synapse: no backends for service #{@name}; using default servers: #{@default_servers.inspect}"
-          @backends = @default_servers
-        end
-      else
-        log.info "synapse: discovered #{new_backends.length} backends for service #{@name}"
-        set_backends(new_backends)
-      end
+      set_backends(new_backends)
     end
 
     # sets up zookeeper callbacks if the data at the discovery path changes
@@ -103,8 +93,6 @@ def watcher_callback
         watch
         # Rediscover
         discover
-        # send a message to calling class to reconfigure
-        reconfigure!
       end
     end