[synapse] enable generator config update from ZK

Gemfile.lock

@@ -1,9 +1,10 @@
 PATH
   remote: .
   specs:
-    synapse (0.14.7)
+    synapse (0.15.1)
       aws-sdk (~> 1.39)
       docker-api (~> 1.7)
+      hashdiff (~> 0.2.3)
       logging (~> 1.8)
       zk (~> 1.9.4)
 

lib/synapse/config_generator/haproxy.rb

@@ -5,6 +5,7 @@
 require 'socket'
 require 'digest/sha1'
 require 'set'
+require 'hashdiff'
 
 class Synapse::ConfigGenerator
   class Haproxy < BaseGenerator
@@ -801,6 +802,8 @@ class Haproxy < BaseGenerator
     # should be enough for anyone right (famous last words)?
     MAX_SERVER_ID = (2**16 - 1).freeze
 
+    attr_reader :server_id_map, :state_cache
+
     def initialize(opts)
       super(opts)
 
@@ -845,8 +848,11 @@ def initialize(opts)
       @backends_cache = {}
       @watcher_revisions = {}
 
-      @state_file_path = @opts['state_file_path']
-      @state_file_ttl = @opts.fetch('state_file_ttl', DEFAULT_STATE_FILE_TTL).to_i
+      @state_cache = HaproxyState.new(
+        @opts['state_file_path'],
+        @opts.fetch('state_file_ttl', DEFAULT_STATE_FILE_TTL).to_i,
+        self
+      )
 
       # For giving consistent orders, even if they are random
       @server_order_seed = @opts.fetch('server_order_seed', rand(2000)).to_i
@@ -907,15 +913,25 @@ def update_config(watchers)
       end
     end
 
+    def update_state_file(watchers)
+      @state_cache.update_state_file(watchers)
+    end
+
     # generates a new config based on the state of the watchers
     def generate_config(watchers)
       new_config = generate_base_config
       shared_frontend_lines = generate_shared_frontend
 
       watchers.each do |watcher|
         watcher_config = watcher.config_for_generator[name]
+        next if watcher_config.nil? || watcher_config.empty? || watcher_config['disabled']
         @watcher_configs[watcher.name] ||= parse_watcher_config(watcher)
-        next if watcher_config['disabled']
+
+        # if watcher_config is changed, trigger restart
+        config_diff = HashDiff.diff(@state_cache.config_for_generator(watcher.name), watcher_config)
+        if !config_diff.empty?
+          @restart_required = true
+        end
 
         regenerate = watcher.revision != @watcher_revisions[watcher.name] ||
                      @frontends_cache[watcher.name].nil? ||
@@ -1051,7 +1067,7 @@ def generate_backend_stanza(watcher, config)
 
       # The ordering here is important.  First we add all the backends in the
       # disabled state...
-      seen.fetch(watcher.name, []).each do |backend_name, backend|
+      @state_cache.backends(watcher).each do |backend_name, backend|
         backends[backend_name] = backend.merge('enabled' => false)
         # We remember the haproxy_server_id from a previous reload here.
         # Note though that if live servers below define haproxy_server_id
@@ -1308,74 +1324,112 @@ def construct_name(backend)
     ######################################
     # methods for managing the state file
     ######################################
-    def seen
-      # if we don't support the state file, return nothing
-      return {} if @state_file_path.nil?
+    class HaproxyState
+      include Synapse::Logging
 
-      # if we've never needed the backends, now is the time to load them
-      @seen = read_state_file if @seen.nil?
+      KEY_WATCHER_CONFIG_FOR_GENERATOR = "watcher_config_for_generator"
+      NON_BACKENDS_KEYS = [KEY_WATCHER_CONFIG_FOR_GENERATOR]
 
-      @seen
-    end
+      def initialize(state_file_path, state_file_ttl, haproxy)
+        @state_file_path = state_file_path
+        @state_file_ttl = state_file_ttl
+        @haproxy = haproxy
+      end
 
-    def update_state_file(watchers)
-      # if we don't support the state file, do nothing
-      return if @state_file_path.nil?
-
-      log.info "synapse: writing state file"
-      timestamp = Time.now.to_i
-
-      # Remove stale backends
-      seen.each do |watcher_name, backends|
-        backends.each do |backend_name, backend|
-          ts = backend.fetch('timestamp', 0)
-          delta = (timestamp - ts).abs
-          if delta > @state_file_ttl
-            log.info "synapse: expiring #{backend_name} with age #{delta}"
-            backends.delete(backend_name)
-          end
+      def backends(watcher_name)
+        if seen.key?(watcher_name)
+          seen[watcher_name].select { |section, data| !NON_BACKENDS_KEYS.include?(section) }
+        else
+          {}
         end
       end
 
-      # Remove any services which no longer have any backends
-      seen.reject!{|watcher_name, backends| backends.keys.length == 0}
+      def config_for_generator(watcher_name)
+        cache_config = {}
+        if seen.key?(watcher_name) && seen[watcher_name].key?(KEY_WATCHER_CONFIG_FOR_GENERATOR)
+          cache_config = seen[watcher_name][KEY_WATCHER_CONFIG_FOR_GENERATOR]
+        end
 
-      # Add backends from watchers
-      watchers.each do |watcher|
-        seen[watcher.name] ||= {}
+        cache_config
+      end
 
-        watcher.backends.each do |backend|
-          backend_name = construct_name(backend)
-          data = {
-            'timestamp' => timestamp,
-          }
-          server_id = @server_id_map[watcher.name][backend_name].to_i
-          if server_id && server_id > 0 && server_id <= MAX_SERVER_ID
-            data['haproxy_server_id'] = server_id
+      def update_state_file(watchers)
+        # if we don't support the state file, do nothing
+        return if @state_file_path.nil?
+
+        log.info "synapse: writing state file"
+        timestamp = Time.now.to_i
+
+        # Remove stale backends
+        seen.each do |watcher_name, data|
+          backends(watcher_name).each do |backend_name, backend|
+            ts = backend.fetch('timestamp', 0)
+            delta = (timestamp - ts).abs
+            if delta > @state_file_ttl
+              log.info "synapse: expiring #{backend_name} with age #{delta}"
+              data.delete(backend_name)
+            end
           end
+        end
 
-          seen[watcher.name][backend_name] = data.merge(backend)
+        # Remove any services which no longer have any backends
+        seen.reject!{|watcher_name, data| backends(watcher_name).keys.length == 0}
+
+        # Add backends and config from watchers
+        watchers.each do |watcher|
+          seen[watcher.name] ||= {}
+
+          watcher.backends.each do |backend|
+            backend_name = @haproxy.construct_name(backend)
+            data = {
+              'timestamp' => timestamp,
+            }
+            server_id = @haproxy.server_id_map[watcher.name][backend_name].to_i
+            if server_id && server_id > 0 && server_id <= MAX_SERVER_ID
+              data['haproxy_server_id'] = server_id
+            end
+
+            seen[watcher.name][backend_name] = data.merge(backend)
+          end
+
+          # Add config for generator from watcher
+          if watcher.config_for_generator.key?(@haproxy.name)
+            seen[watcher.name][KEY_WATCHER_CONFIG_FOR_GENERATOR] =
+              watcher.config_for_generator[@haproxy.name]
+          end
         end
+
+        # write the data!
+        write_data_to_state_file(seen)
       end
 
-      # write the data!
-      write_data_to_state_file(seen)
-    end
+      private
 
-    def read_state_file
-      # Some versions of JSON return nil on an empty file ...
-      JSON.load(File.read(@state_file_path)) || {}
-    rescue StandardError => e
-      # It's ok if the state file doesn't exist or contains invalid data
-      # The state file will be rebuilt automatically
-      {}
-    end
+      def seen
+        # if we don't support the state file, return nothing
+        return {} if @state_file_path.nil?
+
+        # if we've never needed the backends, now is the time to load them
+        @seen = read_state_file if @seen.nil?
+
+        @seen
+      end
 
-    # we do this atomically so the state file is always consistent
-    def write_data_to_state_file(data)
-      tmp_state_file_path = @state_file_path + ".tmp"
-      File.write(tmp_state_file_path, JSON.pretty_generate(data))
-      FileUtils.mv(tmp_state_file_path, @state_file_path)
+      def read_state_file
+        # Some versions of JSON return nil on an empty file ...
+        JSON.load(File.read(@state_file_path)) || {}
+      rescue StandardError => e
+        # It's ok if the state file doesn't exist or contains invalid data
+        # The state file will be rebuilt automatically
+        {}
+      end
+
+      # we do this atomically so the state file is always consistent
+      def write_data_to_state_file(data)
+        tmp_state_file_path = @state_file_path + ".tmp"
+        File.write(tmp_state_file_path, JSON.pretty_generate(data))
+        FileUtils.mv(tmp_state_file_path, @state_file_path)
+      end
     end
   end
 end

lib/synapse/service_watcher/base.rb

@@ -1,5 +1,6 @@
 require 'synapse/log'
 require 'set'
+require 'hashdiff'
 
 class Synapse::ServiceWatcher
   class BaseWatcher
@@ -152,7 +153,7 @@ def backends_filtered_by_labels
       end
     end
 
-    def set_backends(new_backends)
+    def set_backends(new_backends, new_config_for_generator = {})
       # Aggregate and deduplicate all potential backend service instances.
       new_backends = (new_backends + @default_servers) if @keep_default_servers
       # Substitute backend_port_override for the provided port
@@ -165,7 +166,20 @@ def set_backends(new_backends)
         [b['host'], b['port'], b.fetch('name', '')]
       }
 
+      backends_updated = update_backends(new_backends)
+      config_updated = update_config_for_generator(new_config_for_generator)
+
+      if backends_updated || config_updated
+        reconfigure!
+        return true
+      else
+        return false
+      end
+    end
+
+    def update_backends(new_backends)
       if new_backends.to_set == @backends.to_set
+        log.info "synapse: backends for service #{@name} do not change."
         return false
       end
 
@@ -192,11 +206,28 @@ def set_backends(new_backends)
         @backends = new_backends
       end
 
-      reconfigure!
-
       return true
     end
 
+    def update_config_for_generator(new_config_for_generator)
+      if new_config_for_generator.empty?
+        log.info "synapse: no config_for_generator data from #{name} for" \
+              " service #{@name}; keep existing config_for_generator: #{@config_for_generator.inspect}"
+        return false
+      else
+        log.info "synapse: discovered config_for_generator for service #{@name}"
+        diff = HashDiff.diff(new_config_for_generator, config_for_generator)
+
+        if diff.empty?
+          log.info "synapse: config_for_generator for service #{@name} does not change."
+          return false
+        else
+          @config_for_generator = new_config_for_generator
+          return true
+        end
+      end
+    end
+
     # Subclasses should not invoke this directly; it's only exposed so that it
     # can be overridden in subclasses.
     def reconfigure!

lib/synapse/service_watcher/zookeeper.rb

@@ -157,7 +157,15 @@ def discover
         end
       end
 
-      set_backends(new_backends)
+      node = @zk.get(@discovery['path'], :watch => true)
+      begin
+        new_config_for_generator = parse_service_config(node)
+      rescue StandardError => e
+        log.error "synapse: invalid config data in ZK node at #{@discovery['path']}: #{e}"
+        new_config_for_generator = {}
+      end
+
+      set_backends(new_backends, new_config_for_generator)
     end
 
     # sets up zookeeper callbacks if the data at the discovery path changes
@@ -260,6 +268,37 @@ def deserialize_service_instance(data)
 
       return host, port, name, weight, haproxy_server_options, labels
     end
+
+    def parse_service_config(data)
+      log.debug "synapse: deserializing process data"
+      if data.nil? || data.empty?
+        decoded = {}
+      else
+        decoded = @decode_method.call(data)
+      end
+
+      new_generator_config = {}
+      # validate the config. if the config is not empty:
+      #   each key should be named by one of the available generators
+      #   each value should be a hash (could be empty)
+      decoded.collect.each do |generator_name, generator_config|
+        if [email protected]_generators.keys.include?(generator_name)
+          log.error "synapse: invalid generator name in ZK node at #{@discovery['path']}:" \
+            " #{generator_name}"
+          next
+        else
+          if generator_config.nil? || !generator_config.is_a?(Hash)
+            log.warn "synapse: invalid generator config in ZK node at #{@discovery['path']}" \
+            " for generator #{generator_name}"
+            new_generator_config[generator_name] = {}
+          else
+            new_generator_config[generator_name] = generator_config
+          end
+        end
+      end
+
+      return new_generator_config
+    end
   end
 end
 

lib/synapse/version.rb

@@ -1,3 +1,3 @@
 module Synapse
-  VERSION = "0.14.7"
+  VERSION = "0.15.1"
 end