fix(config_etcd): keep the previous value when a full reload gets invalid data

apisix/core/config_etcd.lua

@@ -546,9 +546,24 @@ local function sync_status_to_shdict(status)
 end
 
 
-local function load_full_data(self, dir_res, headers)
+local function get_prev_item(prev_values, prev_values_hash, key)
+    if not prev_values or not prev_values_hash then
+        return nil
+    end
+
+    -- deleted items are tombstoned as `false` in the values array and removed
+    -- from the hash by the watch path, so a hash hit is always a live item
+    local idx = prev_values_hash[key]
+    return idx and prev_values[idx] or nil
+end
+
+
+local function load_full_data(self, dir_res, headers, prev_values, prev_values_hash)
     local err
     local changed = false
+    -- previous items carried over because their new data failed the check;
+    -- their clean handlers must NOT be fired
+    local carried = {}
 
     if self.single_item then
         self.values = new_tab(1, 0)
@@ -583,6 +598,19 @@ local function load_full_data(self, dir_res, headers)
             if self.filter then
                 self.filter(item)
             end
+
+        elseif item.value ~= nil then
+            -- new data exists but is invalid: keep the previous value like the
+            -- incremental watch path does. An absent value (deleted key) must
+            -- not be resurrected, hence the `item.value ~= nil` guard.
+            local prev_item = get_prev_item(prev_values, prev_values_hash, self.key)
+            if prev_item then
+                log.warn("failed to check item data of [", self.key,
+                         "], keep the previous configuration, err: ", err)
+                insert_tab(self.values, prev_item)
+                self.values_hash[self.key] = #self.values
+                carried[prev_item] = true
+            end
         end
 
         self:upgrade_version(item.modifiedIndex)
@@ -601,8 +629,10 @@ local function load_full_data(self, dir_res, headers)
         for _, item in ipairs(values) do
             local key = short_key(self, item.key)
             local data_valid = true
+            err = nil
             if type(item.value) ~= "table" then
                 data_valid = false
+                err = "invalid item data, it should be an object"
                 log.error("invalid item data of [", self.key .. "/" .. key,
                           "], val: ", item.value,
                           ", it should be an object")
@@ -637,12 +667,35 @@ local function load_full_data(self, dir_res, headers)
                 if self.filter then
                     self.filter(item)
                 end
+
+            else
+                local prev_item = get_prev_item(prev_values, prev_values_hash, key)
+                if prev_item then
+                    -- keep serving with the last valid configuration instead of
+                    -- silently dropping the whole item on a full reload, see the
+                    -- incremental path in sync_data for the same semantics
+                    log.warn("failed to check item data of [", self.key, "/", key,
+                             "], keep the previous configuration, err: ", err)
+                    insert_tab(self.values, prev_item)
+                    self.values_hash[key] = #self.values
+                    carried[prev_item] = true
+                end
             end
 
             self:upgrade_version(item.modifiedIndex)
         end
     end
 
+    -- fire the clean handlers of the previous items that were not carried
+    -- over: they were either replaced by a new value or deleted from etcd
+    if prev_values then
+        for _, item in ipairs(prev_values) do
+            if item and not carried[item] then
+                config_util.fire_all_clean_handlers(item)
+            end
+        end
+    end
+
     if headers then
         self.prev_index = tonumber(headers["X-Etcd-Index"]) or 0
         self:upgrade_version(headers["X-Etcd-Index"])
@@ -691,16 +744,15 @@ local function sync_data(self)
         log.debug("readdir key: ", self.key, " res: ",
                   json.delay_encode(dir_res))
 
-        if self.values then
-            for i, val in ipairs(self.values) do
-                config_util.fire_all_clean_handlers(val)
-            end
-
-            self.values = nil
-            self.values_hash = nil
-        end
+        -- hand the previous values over to load_full_data so that an item whose
+        -- new data fails the validation can keep serving with its old value,
+        -- consistent with the incremental watch path. The clean handlers of the
+        -- replaced / deleted items are fired inside load_full_data.
+        local prev_values, prev_values_hash = self.values, self.values_hash
+        self.values = nil
+        self.values_hash = nil
 
-        load_full_data(self, dir_res, headers)
+        load_full_data(self, dir_res, headers, prev_values, prev_values_hash)
 
         return true
     end

apisix/plugin.lua

@@ -1017,8 +1017,16 @@ local function check_single_plugin_schema(name, plugin_conf, schema_type, skip_d
     if plugin_obj.check_schema then
         local ok, err = plugin_obj.check_schema(plugin_conf, schema_type)
         if not ok then
-            return false, "failed to check the configuration of plugin "
-                .. name .. " err: " .. err
+            if check_disable(plugin_conf) ~= true then
+                return false, "failed to check the configuration of plugin "
+                    .. name .. " err: " .. err
+            end
+
+            -- the plugin is disabled via _meta.disable so it will never be
+            -- executed: an environment dependent failure (e.g. proxy-cache
+            -- cache_zone not found on this node) must not invalidate the item
+            core.log.warn("failed to check the configuration of disabled plugin ",
+                          name, ", accepting it anyway")
         end
 
         if plugin_conf._meta then
@@ -1262,8 +1270,13 @@ local function stream_check_schema(plugins_conf, schema_type, skip_disabled_plug
         if plugin_obj.check_schema then
             local ok, err = plugin_obj.check_schema(plugin_conf, schema_type)
             if not ok then
-                return false, "failed to check the configuration of "
-                              .. "stream plugin [" .. name .. "]: " .. err
+                if check_disable(plugin_conf) ~= true then
+                    return false, "failed to check the configuration of "
+                                  .. "stream plugin [" .. name .. "]: " .. err
+                end
+
+                core.log.warn("failed to check the configuration of disabled ",
+                              "stream plugin [", name, "], accepting it anyway")
             end
         end