mirror of
https://github.com/lingble/chatwoot.git
synced 2025-10-29 18:22:53 +00:00
chore: add script to throttle bulkreindex job creation and increase meta timeouts(#12626)
- scripts to throttle reindex job creation and monitor progress ``` RAILS_ENV=production POSTGRES_STATEMENT_TIMEOUT=6000s bundle exec rails runner script/bulk_reindex_messages.rb RAILS_ENV=production bundle exec rails runner script/monitor_reindex.rb ``` --------- Co-authored-by: Pranav <pranavrajs@gmail.com>
This commit is contained in:
@@ -26,12 +26,12 @@ const fetchMetaData = async (commit, params) => {
|
|||||||
};
|
};
|
||||||
|
|
||||||
const debouncedFetchMetaData = debounce(fetchMetaData, 500, false, 1500);
|
const debouncedFetchMetaData = debounce(fetchMetaData, 500, false, 1500);
|
||||||
const longDebouncedFetchMetaData = debounce(fetchMetaData, 1000, false, 8000);
|
const longDebouncedFetchMetaData = debounce(fetchMetaData, 5000, false, 10000);
|
||||||
const superLongDebouncedFetchMetaData = debounce(
|
const superLongDebouncedFetchMetaData = debounce(
|
||||||
fetchMetaData,
|
fetchMetaData,
|
||||||
1500,
|
10000,
|
||||||
false,
|
false,
|
||||||
10000
|
20000
|
||||||
);
|
);
|
||||||
|
|
||||||
export const actions = {
|
export const actions = {
|
||||||
|
|||||||
@@ -27,6 +27,7 @@
|
|||||||
- purgable
|
- purgable
|
||||||
- housekeeping
|
- housekeeping
|
||||||
- async_database_migration
|
- async_database_migration
|
||||||
|
- bulk_reindex_low
|
||||||
- active_storage_analysis
|
- active_storage_analysis
|
||||||
- active_storage_purge
|
- active_storage_purge
|
||||||
- action_mailbox_incineration
|
- action_mailbox_incineration
|
||||||
|
|||||||
58
script/bulk_reindex_messages.rb
Normal file
58
script/bulk_reindex_messages.rb
Normal file
@@ -0,0 +1,58 @@
|
|||||||
|
# Bulk reindex all messages with throttling to prevent DB overload
|
||||||
|
# This creates jobs slowly to avoid overwhelming the database connection pool
|
||||||
|
# Usage: RAILS_ENV=production POSTGRES_STATEMENT_TIMEOUT=6000s bundle exec rails runner script/bulk_reindex_messages.rb
|
||||||
|
|
||||||
|
JOBS_PER_MINUTE = 50 # Adjust based on your DB capacity
|
||||||
|
BATCH_SIZE = 1000 # Messages per job
|
||||||
|
|
||||||
|
batch_count = 0
|
||||||
|
total_batches = (Message.count / BATCH_SIZE.to_f).ceil
|
||||||
|
start_time = Time.zone.now
|
||||||
|
|
||||||
|
index_name = Message.searchkick_index.name
|
||||||
|
|
||||||
|
puts '=' * 80
|
||||||
|
puts "Bulk Reindex Started at #{start_time}"
|
||||||
|
puts '=' * 80
|
||||||
|
puts "Total messages: #{Message.count}"
|
||||||
|
puts "Batch size: #{BATCH_SIZE}"
|
||||||
|
puts "Total batches: #{total_batches}"
|
||||||
|
puts "Index name: #{index_name}"
|
||||||
|
puts "Rate: #{JOBS_PER_MINUTE} jobs/minute (#{JOBS_PER_MINUTE * BATCH_SIZE} messages/minute)"
|
||||||
|
puts "Estimated time: #{(total_batches / JOBS_PER_MINUTE.to_f / 60).round(2)} hours"
|
||||||
|
puts '=' * 80
|
||||||
|
puts ''
|
||||||
|
|
||||||
|
sleep(15)
|
||||||
|
|
||||||
|
Message.find_in_batches(batch_size: BATCH_SIZE).with_index do |batch, index|
|
||||||
|
batch_count += 1
|
||||||
|
|
||||||
|
# Enqueue to low priority queue with proper format
|
||||||
|
Searchkick::BulkReindexJob.set(queue: :bulk_reindex_low).perform_later(
|
||||||
|
class_name: 'Message',
|
||||||
|
index_name: index_name,
|
||||||
|
batch_id: index,
|
||||||
|
record_ids: batch.map(&:id) # Keep as integers like Message.reindex does
|
||||||
|
)
|
||||||
|
|
||||||
|
# Throttle: wait after every N jobs
|
||||||
|
if (batch_count % JOBS_PER_MINUTE).zero?
|
||||||
|
elapsed = Time.zone.now - start_time
|
||||||
|
progress = (batch_count.to_f / total_batches * 100).round(2)
|
||||||
|
queue_size = Sidekiq::Queue.new('bulk_reindex_low').size
|
||||||
|
|
||||||
|
puts "[#{Time.zone.now.strftime('%Y-%m-%d %H:%M:%S')}] Progress: #{batch_count}/#{total_batches} (#{progress}%)"
|
||||||
|
puts " Queue size: #{queue_size}"
|
||||||
|
puts " Elapsed: #{(elapsed / 3600).round(2)} hours"
|
||||||
|
puts " ETA: #{((elapsed / batch_count * (total_batches - batch_count)) / 3600).round(2)} hours remaining"
|
||||||
|
puts ''
|
||||||
|
|
||||||
|
sleep(60)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
puts '=' * 80
|
||||||
|
puts "Done! Created #{batch_count} jobs"
|
||||||
|
puts "Total time: #{((Time.zone.now - start_time) / 3600).round(2)} hours"
|
||||||
|
puts '=' * 80
|
||||||
19
script/monitor_reindex.rb
Normal file
19
script/monitor_reindex.rb
Normal file
@@ -0,0 +1,19 @@
|
|||||||
|
# Monitor bulk reindex progress
|
||||||
|
# RAILS_ENV=production bundle exec rails runner script/monitor_reindex.rb
|
||||||
|
|
||||||
|
puts 'Monitoring bulk reindex progress (Ctrl+C to stop)...'
|
||||||
|
puts ''
|
||||||
|
|
||||||
|
loop do
|
||||||
|
bulk_queue = Sidekiq::Queue.new('bulk_reindex_low')
|
||||||
|
prod_queue = Sidekiq::Queue.new('async_database_migration')
|
||||||
|
retry_set = Sidekiq::RetrySet.new
|
||||||
|
|
||||||
|
puts "[#{Time.zone.now.strftime('%Y-%m-%d %H:%M:%S')}]"
|
||||||
|
puts " Bulk Reindex Queue: #{bulk_queue.size} jobs"
|
||||||
|
puts " Production Queue: #{prod_queue.size} jobs"
|
||||||
|
puts " Retry Queue: #{retry_set.size} jobs"
|
||||||
|
puts " #{('-' * 60)}"
|
||||||
|
|
||||||
|
sleep(30)
|
||||||
|
end
|
||||||
58
script/reindex_single_account.rb
Normal file
58
script/reindex_single_account.rb
Normal file
@@ -0,0 +1,58 @@
|
|||||||
|
# Reindex messages for a single account
|
||||||
|
# Usage: bundle exec rails runner script/reindex_single_account.rb ACCOUNT_ID [DAYS_BACK]
|
||||||
|
|
||||||
|
#account_id = ARGV[0]&.to_i
|
||||||
|
days_back = (ARGV[1] || 30).to_i
|
||||||
|
|
||||||
|
# if account_id.nil? || account_id.zero?
|
||||||
|
# puts "Usage: bundle exec rails runner script/reindex_single_account.rb ACCOUNT_ID [DAYS_BACK]"
|
||||||
|
# puts "Example: bundle exec rails runner script/reindex_single_account.rb 93293 30"
|
||||||
|
# exit 1
|
||||||
|
# end
|
||||||
|
|
||||||
|
# account = Account.find(account_id)
|
||||||
|
# puts "=" * 80
|
||||||
|
# puts "Reindexing messages for: #{account.name} (ID: #{account.id})"
|
||||||
|
# puts "=" * 80
|
||||||
|
|
||||||
|
# Enable feature if not already enabled
|
||||||
|
# unless account.feature_enabled?('advanced_search_indexing')
|
||||||
|
# puts "Enabling advanced_search_indexing feature..."
|
||||||
|
# account.enable_features(:advanced_search_indexing)
|
||||||
|
# account.save!
|
||||||
|
# end
|
||||||
|
|
||||||
|
# Get messages to index
|
||||||
|
# messages = Message.where(account_id: account.id)
|
||||||
|
# .where(message_type: [0, 1]) # incoming/outgoing only
|
||||||
|
# .where('created_at >= ?', days_back.days.ago)
|
||||||
|
|
||||||
|
messages = Message.where('created_at >= ?', days_back.days.ago)
|
||||||
|
|
||||||
|
puts "Found #{messages.count} messages to index (last #{days_back} days)"
|
||||||
|
puts ''
|
||||||
|
|
||||||
|
sleep(15)
|
||||||
|
|
||||||
|
# Create bulk reindex jobs
|
||||||
|
index_name = Message.searchkick_index.name
|
||||||
|
batch_count = 0
|
||||||
|
|
||||||
|
messages.find_in_batches(batch_size: 1000).with_index do |batch, index|
|
||||||
|
Searchkick::BulkReindexJob.set(queue: :bulk_reindex_low).perform_later(
|
||||||
|
class_name: 'Message',
|
||||||
|
index_name: index_name,
|
||||||
|
batch_id: index,
|
||||||
|
record_ids: batch.map(&:id)
|
||||||
|
)
|
||||||
|
|
||||||
|
batch_count += 1
|
||||||
|
print '.'
|
||||||
|
sleep(0.5) # Small delay
|
||||||
|
end
|
||||||
|
|
||||||
|
puts ''
|
||||||
|
puts '=' * 80
|
||||||
|
puts "Done! Created #{batch_count} bulk reindex jobs"
|
||||||
|
puts 'Messages will be indexed shortly via the bulk_reindex_low queue'
|
||||||
|
puts '=' * 80
|
||||||
Reference in New Issue
Block a user