Class: OpenC3::ReingestJob

Inherits:
Object show all
Defined in:
lib/openc3/utilities/reingest_job.rb

Overview

Replays raw .bin.gz log files from a bucket, decommutating each packet via DecomCommon.decom_and_publish(check_limits: false) so historical data reaches QuestDB without re-firing limits events.

Runs synchronously (caller wraps in a Thread). Tracks state in a ReingestJobModel. DEDUP is enabled on affected tables during the job and disabled in an ensure block on completion (or after a cooldown window so in-flight WAL commits are covered).

target_version:

- 'as_logged' (default): each file is decoded with the target config hash
that was in effect when the packets were originally logged. Files are
grouped by their embedded target_id and System is rebuilt per group.
- 'current': all files are decoded with the latest target config.
- <hash>: explicit hash, used for every file in the job.

Constant Summary collapse

STATUS_UPDATE_EVERY =

How often to persist progress during the ingest pass (write every N packets)

500
HEARTBEAT_INTERVAL_SEC =

How often to tick the heartbeat during the cooldown sleep

10
@@run_mutex =

Reingest rebuilds the process-global System singleton. Serialize all reingest jobs running in this process so they don't stomp each other.

Mutex.new

Instance Method Summary collapse

Constructor Details

#initialize(job_id:, files:, path:, bucket:, scope:, target_version: 'as_logged', dedup_cooldown_seconds: ENV.fetch('OPENC3_REINGEST_DEDUP_COOLDOWN', 60).to_i, logger: Logger) ⇒ ReingestJob

Returns a new instance of ReingestJob.



53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/openc3/utilities/reingest_job.rb', line 53

def initialize(job_id:, files:, path:, bucket:, scope:,
               target_version: 'as_logged',
               dedup_cooldown_seconds: ENV.fetch('OPENC3_REINGEST_DEDUP_COOLDOWN', 60).to_i,
               logger: Logger)
  @job_id = job_id
  @files = files
  @path = path
  @bucket_env = bucket
  @scope = scope
  @target_version = target_version
  @dedup_cooldown_seconds = dedup_cooldown_seconds
  @logger = logger
end

Instance Method Details

#run ⇒ Object 



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/openc3/utilities/reingest_job.rb', line 67

def run
  tmp_dir = Dir.mktmpdir
  job = load_job
  dedup_enabled_by_us = []
  db_shard = 0
  @@run_mutex.synchronize do
    begin
      mark(job, state: 'Running', progress_phase: 'downloading',
           started_at: Time.now.utc.iso8601,
           progress_total: @files.length)

      # Parse target from path, e.g. "DEFAULT/raw_logs/tlm/INST/20260421/"
      # → "INST". Fail fast if the path doesn't encode one — otherwise
      # ingest would run against whatever System was loaded in this process
      # from a prior job (or raise opaquely inside PacketLogReader), and
      # the job could mark Complete with rows written under the wrong
      # target config.
      path_parts = @path.to_s.split('/').reject(&:empty?)
      unless path_parts.length >= 4 && path_parts[1] == 'raw_logs'
        raise ReingestJobError, "Cannot determine target from path '#{@path}'; expected '{scope}/raw_logs/{tlm|cmd}/{target}/'"
      end
      target = path_parts[3]
      db_shard = QuestDBClient.db_shard_for_target(target, scope: @scope)

      local_files = download_and_uncompress(job, tmp_dir)

      # Pass 1: read raw (no System required) to discover table names and
      # each file's embedded target hash. File hashes are what the "as
      # logged" mode uses to pick the right target_version per file.
      mark(job, progress_phase: 'enabling_dedup', progress_current: 0,
           progress_total: 0)
      table_names, file_versions = discover_tables_and_versions(local_files)
      mark(job, table_names: table_names, progress_total: table_names.length)

      dedup_enabled_by_us, preexisting = enable_dedup(job, table_names, db_shard)
      mark(job,
           dedup_enabled_by_us: dedup_enabled_by_us,
           dedup_preexisting: preexisting,
           dedup_enabled_at: Time.now.utc.iso8601)

      # Pass 2: group files by the target_version we'll load for them,
      # then ingest each group under its own System instance.
      groups = group_files_by_version(local_files, file_versions)
      mark(job, versions_used: groups.keys,
           progress_phase: 'ingesting', progress_current: 0,
           progress_total: 0, packets_written: 0)
      ingest_all_groups(job, groups, target)

      mark(job, progress_phase: 'dedup_cooldown')
      cooldown(job)

      mark(job, progress_phase: 'disabling_dedup')
      disabled = disable_dedup(job, dedup_enabled_by_us, db_shard)
      mark(job, dedup_disabled_tables: disabled,
           dedup_disabled_at: Time.now.utc.iso8601,
           state: 'Complete',
           finished_at: Time.now.utc.iso8601)
    rescue Exception => e
      @logger.error("Reingest job #{@job_id} failed: #{e.message}\n#{e.backtrace.first(10).join("\n")}")
      # Always try to revert DEDUP even on crash so user tables are not left altered
      disabled_on_crash = []
      begin
        disabled_on_crash = disable_dedup(job, dedup_enabled_by_us, db_shard)
      rescue => de
        @logger.error("Reingest job #{@job_id} failed to disable DEDUP during crash cleanup: #{de.message}")
      end
      mark(job,
           dedup_disabled_tables: disabled_on_crash,
           dedup_disabled_at: Time.now.utc.iso8601,
           state: 'Crashed',
           error: e.message,
           finished_at: Time.now.utc.iso8601)
    ensure
      FileUtils.remove_entry_secure(tmp_dir, true) if tmp_dir && File.directory?(tmp_dir)
    end
  end
end