additional zk action handling

Author: terrywbrady

src-colladmin/actions/access_queue_action.rb

@@ -5,10 +5,6 @@
 
 # Collection Admin Task class - see config/actions.yml for description
 class AccessQueueAction < ZookeeperListAction
-  def initialize(config, action, path, myparams)
-    super(config, action, path, myparams)
-  end
-
   def perform_action
     jobs = []
     if ZookeeperListAction.migration_m3?

src-colladmin/actions/forward_to_ingest_action.rb

@@ -9,6 +9,13 @@ class ForwardToIngestAction < AdminAction
   def initialize(config, action, path, myparams, endpoint)
     super(config, action, path, myparams)
     @endpoint = endpoint
+
+    @zk = ZK.new(get_zookeeper_conn)
+    ZookeeperListAction.migration_level(@zk)
+  end
+
+  def get_zookeeper_conn
+    @config.fetch('zookeeper', '')
   end
 
   def get_body

src-colladmin/actions/ingest_batch_action.rb

@@ -7,7 +7,7 @@ class IngestBatchAction < ZookeeperListAction
   def initialize(config, action, path, myparams)
     @batch = myparams.fetch('batch', 'no-batch-provided')
     @batch_obj = Batch.new(@batch)
-    super(config, action, path, myparams, {batch: @batch})
+    super(config, action, path, myparams, { batch: @batch })
   end
 
   def get_title
@@ -22,7 +22,7 @@ def table_types
     @batch_obj.table_types
   end
 
-  def table_rows(body)
+  def table_rows(_body)
     queue_list = QueueList.new(@zk, { batch: @batch })
     queue_list.jobs.each do |qe|
       @batch_obj.add_queue_job(qe)

src-colladmin/actions/ingest_batch_folders_action.rb

@@ -8,12 +8,6 @@ def initialize(config, action, path, myparams)
     @days = myparams.fetch('days', '7').to_i
     @days = 60 if @days > 60
     super(config, action, path, myparams, "admin/bids/#{@days}")
-    @zk = ZK.new(get_zookeeper_conn)
-    ZookeeperListAction.migration_level(@zk)
-  end
-
-  def get_zookeeper_conn
-    @config.fetch('zookeeper', '')
   end
 
   def get_title

src-colladmin/actions/ingest_collection_locks_action.rb

@@ -15,8 +15,6 @@ def initialize(config, action, path, myparams)
     @collections = Collections.new(config)
 
     super(config, action, path, myparams, endpoint)
-    @zk = ZK.new(get_zookeeper_conn)
-    ZookeeperListAction.migration_level(@zk)
     @held_counts = {}
     ql = QueueList.new(@zk)
     ql.jobs.each do |qe|

src-colladmin/actions/inventory_queue_action.rb

@@ -5,10 +5,6 @@
 
 # Collection Admin Task class - see config/actions.yml for description
 class InventoryQueueAction < ZookeeperListAction
-  def initialize(config, action, path, myparams)
-    super(config, action, path, myparams)
-  end
-
   def perform_action
     jobs = ZookeeperListAction.migration_m1? ? [] : MerrittZK::LegacyInventoryJob.list_jobs(@zk)
     jobs.each do |po|

src-colladmin/actions/queue_action.rb

@@ -4,14 +4,6 @@
 require_relative 'forward_to_ingest_action'
 require_relative '../lib/http_post_json'
 
-# Collection Admin Task class - see config/actions.yml for description
-class QueueAction < PostToIngestAction
-  def initialize(config, action, path, myparams, endpoint)
-    qp = CGI.unescape(myparams.fetch('queue-path', 'na'))
-    super(config, action, path, myparams, "#{endpoint}#{qp}")
-  end
-end
-
 # Collection Admin Task class - see config/actions.yml for description
 class CollQueueAction < PostToIngestAction
   def initialize(config, action, path, myparams, endpoint)
@@ -22,58 +14,74 @@ def initialize(config, action, path, myparams, endpoint)
 end
 
 # Collection Admin Task class - see config/actions.yml for description
-class CollIterateQueueAction < PostToIngestAction
-  def initialize(config, action, path, myparams, endpoint)
-    coll = myparams.fetch('coll', '')
-    @it_endpoint = endpoint.gsub(/coll$/, coll) unless coll.empty?
-    super(config, action, path, myparams, 'admin/queues')
+class CollIterateQueueAction < ZookeeperAction
+  def initialize(config, action, path, myparams)
+    super(config, action, path, myparams)
+    @coll = myparams.fetch('coll', '')
   end
 
   def perform_action
-    resp = { status: 500 }
-    data = JSON.parse(get_body)
-    data = data.fetch('ingq:ingestQueueNameState', {})
-    data = data.fetch('ingq:ingestQueueName', {})
-    data.fetch('ingq:ingestQueue', []).each do |qjson|
-      node = qjson.fetch('ingq:node', '')
-      next if node.empty?
-
-      begin
-        resp = HttpPostJson.new(get_ingest_server, @it_endpoint.gsub('queue', node))
-      rescue StandardError => e
-        log(e.message)
-        log(e.backtrace)
+    if ZookeeperListAction.migration_m1?
+      ql = QueueList.new(@zk, { deletable: true })
+      puts "PROF: #{ql.profiles.keys}"
+    else
+      MerrittZK::LegacyIngestJob.list_jobs(@zk) do |job|
+        puts "TEST COLL #{@coll}: #{job}"
       end
     end
     { message: 'queue release submitted' }.to_json
   end
 end
 
 # Collection Admin Task class - see config/actions.yml for description
-class IterateQueueAction < PostToIngestAction
-  def initialize(config, action, path, myparams, endpoint)
+class IterateQueueAction < ZookeeperAction
+  def initialize(config, action, path, myparams)
+    super(config, action, path, myparams)
     @queue = myparams.fetch('queue', 'na')
     @reload_path = myparams.fetch('reload_path', 'na')
-    @it_endpoint = endpoint
-    super(config, action, path, myparams, "admin/#{@queue}")
+  end
+
+  def legacy_delete(job)
+    status = job.fetch(:status, '')
+    path = job.fetch(:path, '')
+    return unless %w[Completed Deleted].include?(status)
+    return if path.empty?
+
+    @zk.delete(path)
   end
 
   def perform_action
-    resp = { status: 500 }
-    data = JSON.parse(get_body)
-    data = data.fetch('ingq:ingestQueueNameState', {})
-    data = data.fetch('ingq:ingestQueueName', {})
-    MerrittJson.json_fetch_array_val(data, 'ingq:ingestQueue').each do |qjson|
-      node = qjson.fetch('ingq:node', '')
-      next if node.empty?
+    if @queue == 'queues-acc' && ZookeeperListAction.migration_m3?
+      MerrittZK::Access.list_jobs(@zk).each do |job|
+        qn = job.fetch(:queueNode, MerrittZK::Access::SMALL).gsub(%r{^/access/}, '')
+        j = MerrittZK::Access.new(qn, job.fetch(:id, ''))
+        j.load(@zk)
+        next unless j.status.deletable?
+
+        j.delete(@zk)
+      end
+    elsif @queue == 'queues-acc'
+      MerrittZK::LegacyAccessJob.list_jobs(@zk).each do |job|
+        legacy_delete(job)
+      end
+    elsif @queue == 'queues-inv' && ZookeeperListAction.migration_m1?
+      # no action
+    elsif @queue == 'queues-inv'
+      MerrittZK::LegacyInventoryJob.list_jobs(@zk).each do |job|
+        legacy_delete(job)
+      end
+    elsif ZookeeperListAction.migration_m1?
+      ql = QueueList.new(@zk, { deletable: true })
+      ql.batches.each_key do |bid|
+        batch = MerrittZK::Batch.find_batch_by_uuid(@zk, bid)
+        batch.load(@zk)
+        next unless batch.status.deletable?
 
-      begin
-        endpt = @it_endpoint.gsub('queue', node).gsub(%r{//}, '/')
-        resp = HttpPostJson.new(get_ingest_server, endpt)
-        return { message: "Status #{resp.status} for #{endpt}" }.to_json unless resp.status == 200
-      rescue StandardError => e
-        log(e.message)
-        log(e.backtrace)
+        batch.delete(@zk)
+      end
+    else
+      MerrittZK::LegacyIngestJob.list_jobs(@zk).each do |job|
+        legacy_delete(job)
       end
     end
     {

src-colladmin/actions/zookeeper_action.rb

@@ -42,8 +42,6 @@ def initialize(config, action, path, myparams, _filters = {})
   end
 
   def self.migration_level(zk)
-    return unless $migration.nil?
-
     $migration = []
     $migration << :m1 if zk.exists?('/migration/m1')
     $migration << :m3 if zk.exists?('/migration/m3')

src-colladmin/config/actions.yml

@@ -1011,7 +1011,6 @@ release-queue-item-legacy:
 cleanup-queue:
   link-title: Cleanup Deleted and Completed items from a Zookeeper Queue
   class: IterateQueueAction
-  params: ["admin/cleanupq/queue"]
   category: Queue Management
   sensitivity: irreversible change
   testing: manual

src-colladmin/lib/acc_queue.rb

@@ -121,4 +121,4 @@ def status
   def date
     get_value(:date)
   end
-end
+end

src-colladmin/lib/inv_queue.rb

@@ -81,6 +81,7 @@ def self.table_types
       type = 'status' if sym == :status
       type = 'datetime' if sym == :date
       if ZookeeperListAction.migration_m1?
+        # under m1 migration, there is no INV queue
         type = 'qdelete-mrtzk' if sym == :qdelete
         type = 'requeue-mrtzk' if sym == :requeue
       else

src-colladmin/lib/queue_json.rb

@@ -12,30 +12,13 @@ def path
     "#{get_queue_node}/#{get_value(:queueId, '')}"
   end
 
-  def get_queue_path(requeue: false)
+  def get_del_queue_path_m1
     st = get_value(:qstatus, '')
-    case st
-    when 'Consumed'
-      st = 'consume'
-    when 'Held'
-      return '' if requeue
-
-      st = 'held'
-    when 'Completed'
-      return '' if requeue
-
-      st = 'complete'
-    when 'Failed'
-      st = 'fail'
+    if ZookeeperListAction.migration_m1?
+      return '' unless %w[Failed Held].include?(st)
     else
-      return ''
+      return '' unless %w[Failed Completed Held].include?(st)
     end
-    "#{get_queue_node}/#{get_value(:queueId, '')}/#{st}"
-  end
-
-  def get_del_queue_path_m1
-    st = get_value(:qstatus, '')
-    return '' unless %w[Failed Completed Held].include?(st)
 
     path
   end
@@ -47,18 +30,6 @@ def get_requeue_path_m1
     path
   end
 
-  def get_hold_path(release: false)
-    st = get_value(:qstatus, '')
-    if st == 'Held' && release
-      'release'
-    elsif st == 'Pending' && !release
-      'hold'
-    else
-      return ''
-    end
-    path
-  end
-
   def get_hold_path_m1
     st = get_value(:qstatus, '')
     return '' unless %w[Pending].include?(st)