work in prog

Author: terrywbrady

src-colladmin/actions/access_queue_action.rb

@@ -6,7 +6,7 @@
 # Collection Admin Task class - see config/actions.yml for description
 class AccessQueueAction < ZookeeperListAction
   def initialize(config, action, path, myparams)
-    super(config, action, path, myparams, 'admin/queues-acc')
+    super(config, action, path, myparams)
   end
 
   def perform_action

src-colladmin/actions/ingest_batch_action.rb

@@ -3,11 +3,11 @@
 require_relative 'forward_to_ingest_action'
 
 # Collection Admin Task class - see config/actions.yml for description
-class IngestBatchAction < ForwardToIngestAction
+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, 'admin/queues')
+    super(config, action, path, myparams, {batch: @batch})
   end
 
   def get_title
@@ -23,7 +23,7 @@ def table_types
   end
 
   def table_rows(body)
-    queue_list = QueueList.new(get_ingest_server, body, { batch: @batch })
+    queue_list = QueueList.new(@zk, { batch: @batch })
     queue_list.jobs.each do |qe|
       @batch_obj.add_queue_job(qe)
     end

src-colladmin/actions/ingest_batch_folders_action.rb

@@ -8,6 +8,12 @@ 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
@@ -28,7 +34,7 @@ def init_status
 
   def table_rows(body)
     bflist = BatchFolderList.new(body)
-    bflist.apply_queue_list(QueueList.get_queue_list(get_ingest_server))
+    bflist.apply_queue_list(QueueList.new(@zk))
     bflist.apply_recent_ingests(RecentIngests.new(@config, @days))
     bflist.to_table
   end

src-colladmin/actions/ingest_collection_locks_action.rb

@@ -15,15 +15,21 @@ 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.get_queue_list(get_ingest_server)
+    ql = QueueList.new(@zk)
     ql.jobs.each do |qe|
       next if qe.qstatus != 'Held'
 
       @held_counts[qe.profile] = @held_counts.fetch(qe.profile, 0) + 1
     end
   end
 
+  def get_zookeeper_conn
+    @config.fetch('zookeeper', '')
+  end
+
   def specific_profile?
     @profile != ''
   end

src-colladmin/actions/inventory_queue_action.rb

@@ -6,7 +6,7 @@
 # Collection Admin Task class - see config/actions.yml for description
 class InventoryQueueAction < ZookeeperListAction
   def initialize(config, action, path, myparams)
-    super(config, action, path, myparams, 'admin/queues-inv')
+    super(config, action, path, myparams)
   end
 
   def perform_action

src-colladmin/actions/zookeeper_action.rb

@@ -33,20 +33,20 @@ def to_table
 
 ## Base class for actions that interact directly with Zookeeper using the mrt-zk library
 class ZookeeperListAction < AdminAction
-  def initialize(config, action, path, myparams, _filters)
+  def initialize(config, action, path, myparams, _filters = {})
     super(config, action, path, myparams)
     @filters = {}
     @zk = ZK.new(get_zookeeper_conn)
+    ZookeeperListAction.migration_level(@zk)
     @items = ZkList.new
-    migration_level
   end
 
-  def migration_level
+  def self.migration_level(zk)
     return unless $migration.nil?
 
     $migration = []
-    $migration << :m1 if @zk.exists?('/migration/m1')
-    $migration << :m3 if @zk.exists?('/migration/m3')
+    $migration << :m1 if zk.exists?('/migration/m1')
+    $migration << :m3 if zk.exists?('/migration/m3')
   end
 
   def self.migration_m1?
@@ -85,6 +85,7 @@ class ZookeeperAction < AdminAction
   def initialize(config, action, path, myparams)
     super(config, action, path, myparams)
     @zk = ZK.new(get_zookeeper_conn)
+    ZookeeperListAction.migration_level(@zk)
     @qpath = myparams.fetch('queue-path', '')
   end
 

src-colladmin/config/actions.yml

@@ -904,20 +904,6 @@ createProfile/sla:
   documentation: |
     Ingest: POST /admin/profile/sla
   migration: m5
-queue-delete:
-  link-title: Remove an item from a Zookeeper queue
-  class: QueueAction
-  params: ["admin/deleteq"]
-  category: Queue Management
-  sensitivity: irreversible change
-  testing: manual
-  description: |
-    Remove an item from a Zookeeper queue.
-  report-datatypes:
-  - qdelete
-  documentation: |
-    Ingest: POST /admin/deleteq
-  migration: m1, m2, m3
 queue-delete-mrtzk:
   link-title: Remove an item from a Zookeeper queue
   class: ZkDeleteM1Action
@@ -944,20 +930,6 @@ queue-delete-legacy:
   documentation: |
     Zk: tbd
   migration: m1
-requeue:
-  link-title: Re-queue an item from a Zookeeper queue
-  class: QueueAction
-  params: ["admin/requeue"]
-  category: Queue Management
-  sensitivity: irreversible change
-  testing: manual
-  description: |
-    Re-queue an item from a Zookeeper queue.
-  report-datatypes:
-  - requeue
-  documentation: |
-    Ingest: POST /admin/requeue
-  migration: m1, m2, m3
 requeue-mrtzk:
   link-title: Re-queue an item from a Zookeeper queue
   class: ZkRequeueM1Action
@@ -984,20 +956,6 @@ requeue-legacy:
   documentation: |
     ZK tbd
   migration: m1
-hold-queue-item:
-  link-title: Move a pending item from in Zookeeper queue to a Held status
-  class: QueueAction
-  params: ["admin/hold"]
-  category: Queue Management
-  sensitivity: irreversible change
-  testing: manual
-  description: |
-    Move a pending item from in Zookeeper queue to a Held status.
-  report-datatypes:
-  - hold
-  documentation: |
-    Ingest: POST /admin/hold
-  migration: m1, m2
 hold-queue-item-mrtzk:
   link-title: Move a pending item from in Zookeeper queue to a Held status
   class: ZkHoldM1Action
@@ -1024,20 +982,6 @@ hold-queue-item-legacy:
   documentation: |
     Zk: tbd
   migration: m1
-release-queue-item:
-  link-title: Release a held item from in Zookeeper queue to a Pending status
-  class: QueueAction
-  params: ["admin/release"]
-  category: Queue Management
-  sensitivity: irreversible change
-  testing: manual
-  description: |
-    Release a held item from in Zookeeper queue to a Pending status.
-  report-datatypes:
-  - release
-  documentation: |
-    Ingest: POST /admin/release
-  migration: m1, m2
 release-queue-item-mrtzk:
   link-title: Release a held item from in Zookeeper queue to a Pending status
   class: ZkReleaseM1Action

src-colladmin/lib/acc_queue.rb

@@ -121,70 +121,4 @@ def status
   def date
     get_value(:date)
   end
-end
-
-# access queue
-class AccQueue < MerrittJson
-  def initialize(queue_list, body)
-    data = JSON.parse(body)
-    data = fetch_hash_val(data, 'que:queueState')
-    data = fetch_hash_val(data, 'que:queueEntries')
-    list = fetch_array_val(data, 'que:queueEntryState')
-    list.each do |obj|
-      q = AccQueueEntry.new(obj)
-      queue_list.tokens.append(q)
-    end
-    super()
-  end
-end
-
-# list of available access queues
-class AccQueueList < MerrittJson
-  def initialize(ingest_server, body, _filter = {})
-    super()
-    @ingest_server = ingest_server
-    @body = body
-    @tokens = []
-    retrieve_queues
-  end
-
-  def self.get_queue_list(ingest_server, filter = {})
-    qjson = HttpGetJson.new(ingest_server, 'admin/queues-acc')
-    AccQueueList.new(ingest_server, qjson.body, filter)
-  end
-
-  def retrieve_queues
-    data = JSON.parse(@body)
-    data = fetch_hash_val(data, 'ingq:ingestQueueNameState')
-    data = fetch_hash_val(data, 'ingq:ingestQueueName')
-    fetch_array_val(data, 'ingq:ingestQueue').each do |qjson|
-      node = fetch_hash_val(qjson, 'ingq:node')
-      begin
-        qjson = HttpGetJson.new(@ingest_server, "admin/queue-acc#{node}")
-        next unless qjson.status == 200
-
-        AccQueue.new(self, qjson.body)
-      rescue StandardError => e
-        LambdaBase.log(e.message)
-        LambdaBase.log(e.backtrace)
-      end
-    end
-  end
-
-  attr_reader :tokens, :body
-
-  def to_table
-    table = []
-    ts = @tokens.sort do |a, b|
-      if a.status == b.status
-        b.date <=> a.date
-      else
-        AdminTask.status_sort_val(a.status) <=> AdminTask.status_sort_val(b.status)
-      end
-    end
-    ts.each_with_index do |q, _i|
-      table.append(q.to_table_row)
-    end
-    table
-  end
-end
+end

src-colladmin/lib/queue.rb

@@ -216,64 +216,26 @@ def num_jobs
   end
 end
 
-# representation of the ingest queue
-class IngestQueue < MerrittJson
-  def initialize(queue_list, body)
-    data = JSON.parse(body)
-    data = fetch_hash_val(data, 'que:queueState')
-    data = fetch_hash_val(data, 'que:queueEntries')
-    list = fetch_array_val(data, 'que:queueEntryState')
-    list.each do |obj|
-      q = QueueEntry.new(obj)
-      next unless q.check_filter(queue_list.filter)
-
-      qenrtylist = queue_list.batches.fetch(q.bid, QueueBatch.new(q.bid, q.user))
-      qenrtylist.add_job(q)
-      queue_list.batches[q.bid] = qenrtylist
-      queue_list.jobs.append(q)
-
-      # next if q.qstatus == "Completed" || q.qstatus == "Deleted"
-      k = "#{q.profile},#{q.qstatus}"
-      queue_list.profiles[k] = queue_list.profiles.fetch(k, [])
-      queue_list.profiles[k].append(q)
-      super()
-    end
-  end
-end
-
 # List of queues of a particular type - ingest once had separate queues for each worker
 class QueueList < MerrittJson
-  def initialize(ingest_server, body, filter = {})
+  def initialize(zk, filter = {})
     super()
-    @ingest_server = ingest_server
-    @body = body
     @batches = {}
     @jobs = []
     @profiles = {}
     @filter = filter
-    retrieve_queues
-  end
-
-  def self.get_queue_list(ingest_server, filter = {})
-    qjson = HttpGetJson.new(ingest_server, 'admin/queues')
-    QueueList.new(ingest_server, qjson.body, filter)
-  end
 
-  def retrieve_queues
-    data = JSON.parse(@body)
-    data = fetch_hash_val(data, 'ingq:ingestQueueNameState')
-    data = fetch_hash_val(data, 'ingq:ingestQueueName')
-    fetch_array_val(data, 'ingq:ingestQueue').each do |qjson|
-      node = fetch_hash_val(qjson, 'ingq:node').gsub(%r{^/}, '')
-      begin
-        qjson = HttpGetJson.new(@ingest_server, "admin/queue/#{node}")
-        next unless qjson.status == 200
-
-        IngestQueue.new(self, qjson.body)
-      rescue StandardError => e
-        LambdaBase.log(e.message)
-        LambdaBase.log(e.backtrace)
-      end
+    jobs = ZookeeperListAction.migration_m1? ? MerrittZK::Job.list_jobs(zk) : MerrittZK::LegacyIngestJob.list_jobs(zk)
+    jobs.each do |j|
+      job = QueueEntry.new(j)
+      @jobs << job
+      qb = @batches.fetch(job.bid, QueueBatch.new(job.bid, job.user))
+      qb.add_job(job)
+      @batches[job.bid] = qb
+
+      k = "#{job.profile},#{job.qstatus}"
+      profiles[k] = profiles.fetch(k, [])
+      profiles[k].append(job)
     end
   end