modified on local at 2026-07-13 20:26:16
| 1 | 1 | #!/usr/bin/perl |
| 2 | 2 | #====================================================================== |
| 3 | 3 | # DriftSense -- schema scanner |
| 4 | 4 | # |
| 5 | 5 | # Cron entry point (NOT a CGI). Reads DATABASE_MONITOR_SETTINGS, connects |
| 6 | 6 | # to each configured DB, snapshots current DDL for every non-ignored |
| 7 | 7 | # table, diffs against previously stored DDL, writes new SCHEMA_CHANGE |
| 8 | 8 | # rows referencing BLOB_STORE for the full DDL snapshot. |
| 9 | 9 | # |
| 10 | 10 | # Optimizations baked in: |
| 11 | 11 | # * information_schema.TABLES.UPDATE_TIME early-exit -- skip tables |
| 12 | 12 | # that haven't been touched since our last scan of them. |
| 13 | 13 | # * Content-addressable BLOB_STORE dedup -- if the DDL is byte-for-byte |
| 14 | 14 | # identical to a prior snapshot (which almost always is if you |
| 15 | 15 | # re-scan an unchanged table), zero new bytes are stored. |
| 16 | 16 | # * gzip compression on stored snapshots. |
| 17 | 17 | # * Missing INT primary-key gap detection (per Shawn's own note file) |
| 18 | 18 | # -- captures when someone DELETEd rows without telling anyone. |
| 19 | 19 | # |
| 20 | 20 | # Config: /etc/drift_sense/drift_sense.conf (via MODS::Config). |
| 21 | 21 | # Log: /var/log/drift_sense/schema_scan.log (via cron redirect). |
| 22 | 22 | #====================================================================== |
| 23 | 23 | use strict; |
| 24 | 24 | use warnings; |
| 25 | 25 | use lib '/var/www/vhosts/3dshawn.com/site1'; |
| 26 | 26 | use POSIX (); |
| 27 | 27 | use DBI; |
| 28 | 28 | use Digest::SHA qw(sha256_hex); |
| 29 | 29 | use Compress::Zlib qw(compress); |
| 30 | 30 | use MODS::Config; |
| 31 | 31 | use MODS::DBConnect; |
| 32 | 32 | |
| 33 | 33 | $| = 1; |
| 34 | 34 | my $ts = POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime); |
| 35 | 35 | |
| 36 | 36 | my $cfg = MODS::Config->new; |
| 37 | 37 | my $db = MODS::DBConnect->new; |
| 38 | 38 | my $dbh = $db->db_connect or die "[$ts] cannot connect to DriftSense storage DB\n"; |
| 39 | 39 | |
| 40 | 40 | # ---- Fetch active database monitors ---- |
| 41 | 41 | my @monitors = $db->db_readwrite_multiple($dbh, q~ |
| 42 | 42 | SELECT database_list_id, db_name, hostname, port, username, password, |
| 43 | 43 | ignore_tables, track_row_count, notify_emails, server_id |
| 44 | 44 | FROM DATABASE_MONITOR_SETTINGS |
| 45 | 45 | WHERE status = 1 |
| 46 | 46 | ~, __FILE__, __LINE__); |
| 47 | 47 | |
| 48 | 48 | unless (@monitors) { |
| 49 | 49 | print "[$ts] no active monitors, exiting\n"; |
| 50 | 50 | exit 0; |
| 51 | 51 | } |
| 52 | 52 | |
| 53 | 53 | my $total_changes = 0; |
| 54 | 54 | my $total_scanned = 0; |
| 55 | 55 | my $total_failed = 0; |
| 56 | 56 | |
| 57 | 57 | foreach my $m (@monitors) { |
| 58 | 58 | my $dsn = "DBI:mysql:$m->{db_name}:$m->{hostname}:$m->{port}"; |
| 59 | 59 | my $target_dbh = eval { |
| 60 | 60 | DBI->connect($dsn, $m->{username}, $m->{password}, { |
| 61 | 61 | PrintError => 0, RaiseError => 0, mysql_connect_timeout => 5, |
| 62 | 62 | }); |
| 63 | 63 | }; |
| 64 | 64 | unless ($target_dbh) { |
| 65 | 65 | my $q_db = $dbh->quote($m->{db_name}); |
| 66 | 66 | $db->db_readwrite($dbh, qq~ |
| 67 | 67 | INSERT INTO FAILED_SCAN_LOGS (server_id, db_name, date_time) |
| 68 | 68 | VALUES ($m->{server_id}, $q_db, NOW()) |
| 69 | 69 | ~, __FILE__, __LINE__); |
| 70 | 70 | $total_failed++; |
| 71 | 71 | print "[$ts] $m->{db_name}: connect failed\n"; |
| 72 | 72 | next; |
| 73 | 73 | } |
| 74 | 74 | |
| 75 | 75 | # ---- Ignore-list parse (comma or newline separated) ---- |
| 76 | 76 | my %ignored; |
| 77 | 77 | if ($m->{ignore_tables}) { |
| 78 | 78 | for my $t (split /[\s,]+/, $m->{ignore_tables}) { |
| 79 | 79 | $t =~ s/^\s+|\s+$//g; |
| 80 | 80 | $ignored{$t} = 1 if length $t; |
| 81 | 81 | } |
| 82 | 82 | } |
| 83 | 83 | |
| 84 | 84 | # ---- Discover tables + UPDATE_TIME for early-exit ---- |
| 85 | 85 | my $tab_sth = $target_dbh->prepare(qq~ |
| 86 | 86 | SELECT TABLE_NAME, UPDATE_TIME |
| 87 | 87 | FROM information_schema.TABLES |
| 88 | 88 | WHERE TABLE_SCHEMA = ? |
| 89 | 89 | ~); |
| 90 | 90 | $tab_sth->execute($m->{db_name}) or do { |
| 91 | 91 | $total_failed++; |
| 92 | 92 | $target_dbh->disconnect; |
| 93 | 93 | print "[$ts] $m->{db_name}: information_schema query failed\n"; |
| 94 | 94 | next; |
| 95 | 95 | }; |
| 96 | 96 | my @tables; |
| 97 | 97 | while (my $r = $tab_sth->fetchrow_hashref) { |
| 98 | 98 | next if $ignored{$r->{TABLE_NAME}}; |
| 99 | 99 | push @tables, $r; |
| 100 | 100 | } |
| 101 | 101 | $tab_sth->finish; |
| 102 | 102 | |
| 103 | 103 | foreach my $t (@tables) { |
| 104 | 104 | $total_scanned++; |
| 105 | 105 | my $tname = $t->{TABLE_NAME}; |
| 106 | 106 | my $update_time = $t->{UPDATE_TIME} || ''; |
| 107 | 107 | |
| 108 | 108 | # ---- Fetch current DDL ---- |
| 109 | 109 | my $ddl_row = eval { |
| 110 | 110 | $target_dbh->selectrow_arrayref("SHOW CREATE TABLE `$m->{db_name}`.`$tname`"); |
| 111 | 111 | }; |
| 112 | 112 | unless ($ddl_row && $ddl_row->[1]) { |
| 113 | 113 | print "[$ts] $m->{db_name}.$tname: SHOW CREATE TABLE failed\n"; |
| 114 | 114 | next; |
| 115 | 115 | } |
| 116 | 116 | my $ddl = $ddl_row->[1]; |
| 117 | 117 | # Normalize noise before hashing so churn-only DDL bumps don't |
| 118 | 118 | # register as schema changes: |
| 119 | 119 | # * AUTO_INCREMENT=N in the table options -- bumps every time a |
| 120 | 120 | # row is inserted into the table (very noisy on log/audit tables) |
| 121 | 121 | # * ROW_FORMAT / STATS_* / ENCRYPTION default noise some MySQL |
| 122 | 122 | # builds emit inconsistently. |
| 123 | 123 | # Keep the DDL structure otherwise fully intact. |
| 124 | 124 | my $ddl_norm = $ddl; |
| 125 | 125 | $ddl_norm =~ s/\s*AUTO_INCREMENT=\d+//g; |
| 126 | 126 | my $sha = sha256_hex($ddl_norm); |
| 127 | 127 | |
| 128 | 128 | # ---- Compare against previously stored DDL for this table ---- |
| 129 | 129 | my $q_dbname = $dbh->quote($m->{db_name}); |
| 130 | 130 | my $q_tname = $dbh->quote($tname); |
| 131 | 131 | my $prev = $db->db_readwrite($dbh, qq~ |
| 132 | 132 | SELECT schema_blob_sha FROM SCHEMA_CHANGE |
| 133 | 133 | WHERE server_id = $m->{server_id} |
| 134 | 134 | AND database_name = $q_dbname |
| 135 | 135 | AND table_name = $q_tname |
| 136 | 136 | ORDER BY schema_change_id DESC LIMIT 1 |
| 137 | 137 | ~, __FILE__, __LINE__); |
| 138 | 138 | |
| 139 | 139 | my $prev_sha = $prev && $prev->{schema_blob_sha} ? $prev->{schema_blob_sha} : ''; |
| 140 | 140 | next if $prev_sha eq $sha; # unchanged -- skip |
| 141 | 141 | |
| 142 | 142 | # ---- Persist to BLOB_STORE (content-addressable, gzipped) ---- |
| 143 | 143 | _store_blob($db, $dbh, $sha, $ddl); |
| 144 | 144 | |
| 145 | 145 | # ---- Compute a compact diff summary (best-effort) ---- |
| 146 | 146 | my $prev_ddl = ''; |
| 147 | 147 | if ($prev_sha) { |
| 148 | 148 | $prev_ddl = _fetch_blob($db, $dbh, $prev_sha) || ''; |
| 149 | 149 | } |
| 150 | 150 | my $diff_summary = _short_diff($prev_ddl, $ddl); |
| 151 | 151 | |
| 152 | # First-ever capture of this table = baseline. Once we have a prev | |
| 153 | # SHA on record, any subsequent capture is a real DDL change. | |
| 154 | my $is_baseline = $prev_sha ? 0 : 1; | |
| 152 | 155 | my $q_diff = $dbh->quote($diff_summary); |
| 153 | 156 | my $q_sha = $dbh->quote($sha); |
| 154 | 157 | $db->db_readwrite($dbh, qq~ |
| 155 | 158 | INSERT INTO SCHEMA_CHANGE |
| 156 | (server_id, database_name, table_name, changes, | |
| 159 | (server_id, database_name, table_name, changes, is_baseline, | |
| 157 | 160 | schema_blob_sha, change_datetime) |
| 158 | 161 | VALUES |
| 159 | ($m->{server_id}, $q_dbname, $q_tname, $q_diff, $q_sha, NOW()) | |
| 162 | ($m->{server_id}, $q_dbname, $q_tname, $q_diff, $is_baseline, | |
| 163 | $q_sha, NOW()) | |
| 160 | 164 | ~, __FILE__, __LINE__); |
| 161 | 165 | $total_changes++; |
| 162 | 166 | print "[$ts] $m->{db_name}.$tname changed\n"; |
| 163 | 167 | } |
| 164 | 168 | |
| 165 | 169 | # ---- Row-count tracking + missing INT PK gap detection ---- |
| 166 | 170 | if ($m->{track_row_count}) { |
| 167 | 171 | foreach my $t (@tables) { |
| 168 | 172 | my $tname = $t->{TABLE_NAME}; |
| 169 | 173 | my $ct_row = eval { |
| 170 | 174 | $target_dbh->selectrow_arrayref("SELECT COUNT(*) FROM `$m->{db_name}`.`$tname`"); |
| 171 | 175 | }; |
| 172 | 176 | my $count = ($ct_row && $ct_row->[0]) ? $ct_row->[0] : 0; |
| 173 | 177 | |
| 174 | 178 | # Missing-INT-PK gap detection -- Shawn's own note: |
| 175 | 179 | # "look for missing INT numbers in the primary key to know |
| 176 | 180 | # if someone deleted data from the tables" |
| 177 | 181 | my $gap_ct = _pk_gap_count($target_dbh, $m->{db_name}, $tname); |
| 178 | 182 | |
| 179 | 183 | my $q_db = $dbh->quote($m->{db_name}); |
| 180 | 184 | my $q_t = $dbh->quote($tname); |
| 181 | 185 | $db->db_readwrite($dbh, qq~ |
| 182 | 186 | INSERT INTO TABLE_DATA_CHANGES |
| 183 | 187 | (server_id, db_name, table_name, record_count, |
| 184 | 188 | missing_pk_gap_count, date_time) |
| 185 | 189 | VALUES |
| 186 | 190 | ($m->{server_id}, $q_db, $q_t, $count, ~ |
| 187 | 191 | . ($gap_ct // 'NULL') . qq~, NOW()) |
| 188 | 192 | ~, __FILE__, __LINE__); |
| 189 | 193 | } |
| 190 | 194 | } |
| 191 | 195 | |
| 192 | 196 | # ---- Update last_connection stamp on the monitor ---- |
| 193 | 197 | $db->db_readwrite($dbh, qq~ |
| 194 | 198 | UPDATE DATABASE_MONITOR_SETTINGS |
| 195 | 199 | SET connection_status = 1, last_connection = NOW() |
| 196 | 200 | WHERE database_list_id = $m->{database_list_id} |
| 197 | 201 | ~, __FILE__, __LINE__); |
| 198 | 202 | |
| 199 | 203 | $target_dbh->disconnect; |
| 200 | 204 | } |
| 201 | 205 | |
| 202 | 206 | $db->db_disconnect($dbh); |
| 203 | 207 | |
| 204 | 208 | print "[$ts] schema scan complete: $total_scanned tables scanned, " |
| 205 | 209 | . "$total_changes changes captured, $total_failed monitors failed\n" |
| 206 | 210 | if $total_changes || $total_failed; |
| 207 | 211 | |
| 208 | 212 | exit 0; |
| 209 | 213 | |
| 210 | 214 | #--------------------------------------------------------------------- |
| 211 | 215 | # _store_blob -- content-addressable insert into BLOB_STORE. If the SHA |
| 212 | 216 | # already exists, just bump ref_count. Compression is inline gzip via |
| 213 | 217 | # Compress::Zlib (core module, no CPAN dep). |
| 214 | 218 | #--------------------------------------------------------------------- |
| 215 | 219 | sub _store_blob { |
| 216 | 220 | my ($db, $dbh, $sha, $content) = @_; |
| 217 | 221 | my $existing = $db->db_readwrite($dbh, |
| 218 | 222 | "SELECT blob_sha FROM BLOB_STORE WHERE blob_sha = " . $dbh->quote($sha) . " LIMIT 1", |
| 219 | 223 | __FILE__, __LINE__); |
| 220 | 224 | if ($existing && $existing->{blob_sha}) { |
| 221 | 225 | # already stored, just bump ref count |
| 222 | 226 | $db->db_readwrite($dbh, "UPDATE BLOB_STORE SET ref_count = ref_count + 1 WHERE blob_sha = " |
| 223 | 227 | . $dbh->quote($sha), __FILE__, __LINE__); |
| 224 | 228 | return; |
| 225 | 229 | } |
| 226 | 230 | my $gz = compress($content) // ''; |
| 227 | 231 | my $sth = $dbh->prepare(qq~ |
| 228 | 232 | INSERT INTO BLOB_STORE (blob_sha, content_gz, size_uncompressed, size_compressed, ref_count, first_seen_at) |
| 229 | 233 | VALUES (?, ?, ?, ?, 1, NOW()) |
| 230 | 234 | ~); |
| 231 | 235 | if ($sth) { |
| 232 | 236 | $sth->execute($sha, $gz, length($content), length($gz)); |
| 233 | 237 | $sth->finish; |
| 234 | 238 | } |
| 235 | 239 | } |
| 236 | 240 | |
| 237 | 241 | sub _fetch_blob { |
| 238 | 242 | my ($db, $dbh, $sha) = @_; |
| 239 | 243 | my $sth = $dbh->prepare("SELECT content_gz FROM BLOB_STORE WHERE blob_sha = ?"); |
| 240 | 244 | return undef unless $sth && $sth->execute($sha); |
| 241 | 245 | my $row = $sth->fetchrow_arrayref; |
| 242 | 246 | $sth->finish; |
| 243 | 247 | return undef unless $row && $row->[0]; |
| 244 | 248 | my $decompressed = eval { Compress::Zlib::uncompress($row->[0]) }; |
| 245 | 249 | return $decompressed; |
| 246 | 250 | } |
| 247 | 251 | |
| 248 | 252 | #--------------------------------------------------------------------- |
| 249 | 253 | # _short_diff -- compact human-readable summary of what changed. |
| 250 | 254 | # Used for the SCHEMA_CHANGE.changes text column so lists render |
| 251 | 255 | # without decompressing the full BLOB_STORE snapshot. |
| 252 | 256 | #--------------------------------------------------------------------- |
| 253 | 257 | sub _short_diff { |
| 254 | 258 | my ($old, $new) = @_; |
| 255 | 259 | $old //= ''; $new //= ''; |
| 256 | 260 | return 'first snapshot' unless length $old; |
| 257 | 261 | |
| 258 | 262 | my @old_lines = split /\n/, $old; |
| 259 | 263 | my @new_lines = split /\n/, $new; |
| 260 | 264 | my %old_set = map { $_ => 1 } @old_lines; |
| 261 | 265 | my %new_set = map { $_ => 1 } @new_lines; |
| 262 | 266 | |
| 263 | 267 | my @added = grep { !$old_set{$_} } @new_lines; |
| 264 | 268 | my @removed = grep { !$new_set{$_} } @old_lines; |
| 265 | 269 | |
| 266 | 270 | my $summary = ''; |
| 267 | 271 | $summary .= (scalar @added) . ' line(s) added; ' if @added; |
| 268 | 272 | $summary .= (scalar @removed) . ' line(s) removed; ' if @removed; |
| 269 | 273 | $summary =~ s/;\s*$//; |
| 270 | 274 | $summary ||= 'reordered / whitespace'; |
| 271 | 275 | |
| 272 | 276 | # Include a small preview of the first added line (safe truncation) |
| 273 | 277 | if (@added) { |
| 274 | 278 | my $first = $added[0]; $first =~ s/\s+/ /g; |
| 275 | 279 | $first = substr($first, 0, 80) . '...' if length($first) > 80; |
| 276 | 280 | $summary .= " -- e.g. +$first"; |
| 277 | 281 | } elsif (@removed) { |
| 278 | 282 | my $first = $removed[0]; $first =~ s/\s+/ /g; |
| 279 | 283 | $first = substr($first, 0, 80) . '...' if length($first) > 80; |
| 280 | 284 | $summary .= " -- e.g. -$first"; |
| 281 | 285 | } |
| 282 | 286 | return $summary; |
| 283 | 287 | } |
| 284 | 288 | |
| 285 | 289 | #--------------------------------------------------------------------- |
| 286 | 290 | # _pk_gap_count -- if the table has an AUTO_INCREMENT INT primary key, |
| 287 | 291 | # count missing IDs between MIN(id) and MAX(id). Non-zero = someone |
| 288 | 292 | # deleted rows. Returns undef if not applicable (no INT PK). |
| 289 | 293 | #--------------------------------------------------------------------- |
| 290 | 294 | sub _pk_gap_count { |
| 291 | 295 | my ($tdbh, $db_name, $tname) = @_; |
| 292 | 296 | my $pk_sth = eval { |
| 293 | 297 | $tdbh->prepare(qq~ |
| 294 | 298 | SELECT COLUMN_NAME, DATA_TYPE FROM information_schema.COLUMNS |
| 295 | 299 | WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? |
| 296 | 300 | AND COLUMN_KEY = 'PRI' AND EXTRA LIKE '%auto_increment%' |
| 297 | 301 | LIMIT 1 |
| 298 | 302 | ~); |
| 299 | 303 | }; |
| 300 | 304 | return undef unless $pk_sth && $pk_sth->execute($db_name, $tname); |
| 301 | 305 | my $pk = $pk_sth->fetchrow_hashref; |
| 302 | 306 | $pk_sth->finish; |
| 303 | 307 | return undef unless $pk && $pk->{COLUMN_NAME}; |
| 304 | 308 | return undef if $pk->{DATA_TYPE} !~ /int/i; |
| 305 | 309 | |
| 306 | 310 | my $col = $pk->{COLUMN_NAME}; |
| 307 | 311 | my $gap_row = eval { |
| 308 | 312 | $tdbh->selectrow_arrayref(qq~ |
| 309 | 313 | SELECT MAX(`$col`) - MIN(`$col`) + 1 - COUNT(*) AS gaps |
| 310 | 314 | FROM `$db_name`.`$tname` |
| 311 | 315 | ~); |
| 312 | 316 | }; |
| 313 | 317 | return undef unless $gap_row; |
| 314 | 318 | my $g = $gap_row->[0]; |
| 315 | 319 | return (defined $g && $g > 0) ? $g : 0; |
| 316 | 320 | } |