[backport -> release/3.3.x] fix(opentelemetry): migration - increase default queue batch size

kong-3.3.2-0.rockspec

@@ -254,6 +254,7 @@ build = {
     ["kong.db.migrations.operations.210_to_211"] = "kong/db/migrations/operations/210_to_211.lua",
     ["kong.db.migrations.operations.212_to_213"] = "kong/db/migrations/operations/212_to_213.lua",
     ["kong.db.migrations.operations.280_to_300"] = "kong/db/migrations/operations/280_to_300.lua",
+    ["kong.db.migrations.operations.331_to_332"] = "kong/db/migrations/operations/331_to_332.lua",
     ["kong.db.migrations.migrate_path_280_300"] = "kong/db/migrations/migrate_path_280_300.lua",
     ["kong.db.declarative.migrations"] = "kong/db/declarative/migrations/init.lua",
     ["kong.db.declarative.migrations.route_path"] = "kong/db/declarative/migrations/route_path.lua",
@@ -516,6 +517,8 @@ build = {
     ["kong.plugins.azure-functions.handler"] = "kong/plugins/azure-functions/handler.lua",
     ["kong.plugins.azure-functions.schema"]  = "kong/plugins/azure-functions/schema.lua",
 
+    ["kong.plugins.opentelemetry.migrations"] = "kong/plugins/opentelemetry/migrations/init.lua",
+    ["kong.plugins.opentelemetry.migrations.001_331_to_332"] = "kong/plugins/opentelemetry/migrations/001_331_to_332.lua",
     ["kong.plugins.opentelemetry.handler"] = "kong/plugins/opentelemetry/handler.lua",
     ["kong.plugins.opentelemetry.schema"]  = "kong/plugins/opentelemetry/schema.lua",
     ["kong.plugins.opentelemetry.proto"]  = "kong/plugins/opentelemetry/proto.lua",

kong/db/migrations/operations/331_to_332.lua

@@ -0,0 +1,120 @@
+-- Helper module for 331_to_332 migration operations.
+--
+-- Operations are versioned and specific to a migration so they remain
+-- fixed in time and are not modified for use in future migrations.
+--
+-- If you want to reuse these operations in a future migration,
+-- copy the functions over to a new versioned module.
+
+
+local function render(template, keys)
+  return (template:gsub("$%(([A-Z_]+)%)", keys))
+end
+
+
+--------------------------------------------------------------------------------
+-- Postgres operations for Workspace migration
+--------------------------------------------------------------------------------
+
+
+local postgres = {
+
+  up = {},
+
+  teardown = {
+
+    ------------------------------------------------------------------------------
+    -- General function to fixup a plugin configuration
+    fixup_plugin_config = function(_, connector, plugin_name, fixup_fn)
+      local pgmoon_json = require("pgmoon.json")
+      local select_plugin = render(
+        "SELECT id, name, config FROM plugins WHERE name = '$(NAME)'", {
+          NAME = plugin_name
+        })
+
+      local plugins, err = connector:query(select_plugin)
+      if not plugins then
+        return nil, err
+      end
+
+      for _, plugin in ipairs(plugins) do
+        local fix = fixup_fn(plugin.config)
+        if fix then
+          local sql = render(
+            "UPDATE plugins SET config = $(NEW_CONFIG)::jsonb WHERE id = '$(ID)'", {
+              NEW_CONFIG = pgmoon_json.encode_json(plugin.config),
+              ID = plugin.id,
+            })
+
+          local _, err = connector:query(sql)
+          if err then
+            return nil, err
+          end
+        end
+      end
+
+      return true
+    end,
+  },
+
+}
+
+--------------------------------------------------------------------------------
+-- Cassandra operations for Workspace migration
+--------------------------------------------------------------------------------
+
+
+local cassandra = {
+
+  up = {},
+
+  teardown = {
+
+    ------------------------------------------------------------------------------
+    -- General function to fixup a plugin configuration
+    fixup_plugin_config = function(_, connector, plugin_name, fixup_fn)
+      local coordinator = assert(connector:get_stored_connection())
+      local cassandra = require("cassandra")
+      local cjson = require("cjson")
+
+      for rows, err in coordinator:iterate("SELECT id, name, config FROM plugins") do
+        if err then
+          return nil, err
+        end
+
+        for i = 1, #rows do
+          local plugin = rows[i]
+          if plugin.name == plugin_name then
+            if type(plugin.config) ~= "string" then
+              return nil, "plugin config is not a string"
+            end
+            local config = cjson.decode(plugin.config)
+            local fix = fixup_fn(config)
+
+            if fix then
+              local _, err = coordinator:execute("UPDATE plugins SET config = ? WHERE id = ?", {
+                cassandra.text(cjson.encode(config)),
+                cassandra.uuid(plugin.id)
+              })
+              if err then
+                return nil, err
+              end
+            end
+          end
+        end
+      end
+
+      return true
+    end,
+
+  }
+
+}
+
+--------------------------------------------------------------------------------
+
+
+return {
+  postgres = postgres,
+  cassandra = cassandra,
+}

kong/plugins/opentelemetry/migrations/001_331_to_332.lua

@@ -0,0 +1,28 @@
+local operations = require "kong.db.migrations.operations.331_to_332"
+
+
+local function ws_migration_teardown(ops)
+  return function(connector)
+    return ops:fixup_plugin_config(connector, "opentelemetry", function(config)
+      if config.queue.max_batch_size == 1 then
+        config.queue.max_batch_size = 200
+        return true
+      end
+
+      return false
+    end)
+  end
+end
+
+
+return {
+  postgres = {
+    up = "",
+    teardown = ws_migration_teardown(operations.postgres.teardown),
+  },
+
+  cassandra = {
+    up = "",
+    teardown = ws_migration_teardown(operations.cassandra.teardown),
+  },
+}

kong/plugins/opentelemetry/migrations/init.lua

@@ -0,0 +1,3 @@
+return {
+  "001_331_to_332",
+}