From 10f94415a968c9f20977af5136dfb1b558079aa8 Mon Sep 17 00:00:00 2001 From: David Lawrence Date: Mon, 27 Oct 2025 20:27:52 -0400 Subject: [PATCH] Bug 1967334 - BMO ETL: Refactor bmo etl export script to run as a service instead a cron job --- extensions/BMO/bin/bugzilla-etl.pl | 48 ++ extensions/BMO/bin/export_bmo_etl.pl | 1085 -------------------------- extensions/BMO/lib/Daemon.pm | 138 ++++ extensions/BMO/lib/ETL.pm | 84 ++ extensions/BMO/lib/ETL/Export.pm | 1062 +++++++++++++++++++++++++ extensions/BMO/t/bmo/bmo_etl.t | 10 +- 6 files changed, 1338 insertions(+), 1089 deletions(-) create mode 100644 extensions/BMO/bin/bugzilla-etl.pl delete mode 100644 extensions/BMO/bin/export_bmo_etl.pl create mode 100644 extensions/BMO/lib/Daemon.pm create mode 100644 extensions/BMO/lib/ETL.pm create mode 100644 extensions/BMO/lib/ETL/Export.pm diff --git a/extensions/BMO/bin/bugzilla-etl.pl b/extensions/BMO/bin/bugzilla-etl.pl new file mode 100644 index 0000000000..925cb5471e --- /dev/null +++ b/extensions/BMO/bin/bugzilla-etl.pl @@ -0,0 +1,48 @@ +#!/usr/bin/env perl + +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. +# +# This Source Code Form is "Incompatible With Secondary Licenses", as +# defined by the Mozilla Public License, v. 2.0. + +use strict; +use warnings; +use 5.10.1; + +use lib qw(. lib local/lib/perl5); + +BEGIN { + use Bugzilla; + Bugzilla->extensions; +} + +use Bugzilla::Extension::BMO::Daemon; +Bugzilla::Extension::BMO::Daemon->start(); + +=head1 NAME + +bugzilla-etl.pl - Export a daily snapshot of the BMO database to BigQuery. + +=head1 SYNOPSIS + + bugzilla-etl.pl [OPTIONS] COMMAND + + OPTIONS: + -f Run in the foreground (don't detach) + -d Output a lot of debugging information + -p file Specify the file where bugzilla-etl.pl should store its current + process id. Defaults to F. + --snapshot Use the provided snapshot date instead of current date. + --test Output the JSON to test files instead of sending to BigQuery. + --quiet Do not output any information while running. + + COMMANDS: + start Starts a new bugzilla-etl daemon if there isn't one running already + stop Stops a running bugzilla-etl daemon + restart Stops a running bugzilla-etl if one is running, and then + starts a new one. + once Execute only once and then exit. + check Report the current status of the daemon. + help Display this usage info diff --git a/extensions/BMO/bin/export_bmo_etl.pl b/extensions/BMO/bin/export_bmo_etl.pl deleted file mode 100644 index d08c49f039..0000000000 --- a/extensions/BMO/bin/export_bmo_etl.pl +++ /dev/null @@ -1,1085 +0,0 @@ -#!/usr/bin/env perl -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. -# -# This Source Code Form is "Incompatible With Secondary Licenses", as -# defined by the Mozilla Public License, v. 2.0. - -use 5.10.1; -use strict; -use warnings; -use lib qw(. lib local/lib/perl5); - -use Bugzilla; -use Bugzilla::Attachment; -use Bugzilla::Bug; -use Bugzilla::Constants; -use Bugzilla::Flag; -use Bugzilla::Group; -use Bugzilla::Logging; -use Bugzilla::User; -use Bugzilla::Extension::Review::FlagStateActivity; - -use HTTP::Headers; -use HTTP::Request; -use IO::Compress::Gzip qw(gzip $GzipError); -use IO::Uncompress::Gunzip qw(gunzip $GunzipError); -use List::Util qw(any); -use LWP::UserAgent::Determined; -use Mojo::File qw(path); -use Mojo::JSON qw(decode_json encode_json false true); -use Mojo::Util qw(getopt); -use Try::Tiny; - -# BigQuery API cannot handle payloads larger than 10MB so -# we will send data in blocks. -use constant API_BLOCK_COUNT => 1000; - -# Products which we should not send data to ETL such as Legal, etc. -use constant EXCLUDE_PRODUCTS => ('Legal',); - -# Log levels -use constant DEBUG_OUTPUT => 1; - -Bugzilla->usage_mode(USAGE_MODE_CMDLINE); -getopt - 't|test' => \my $test, - 'q|quiet' => \my $quiet, - 'd|debug' => \my $debug, - 's|snapshot-date=s' => \my $snapshot_date, - 'n|no-cache=s' => \my @no_cache; - -# Create a hash for faster checking later -my %no_cache = map { $_ => 1 } @no_cache; - -# Sanity checks -Bugzilla->params->{bmo_etl_enabled} || die "BMO ETL not enabled.\n"; - -my $base_url = Bugzilla->params->{bmo_etl_base_url}; -$base_url || die "Invalid BigQuery base URL.\n"; - -my $project_id = Bugzilla->params->{bmo_etl_project_id}; -$project_id || die "Invalid BigQuery product ID.\n"; - -my $dataset_id = Bugzilla->params->{bmo_etl_dataset_id}; -$dataset_id || die "Invalid BigQuery dataset ID.\n"; - -# Check to make sure another instance is not currently running -check_and_set_lock(); - -# Use dedicated replica1 -my $dbh = Bugzilla->switch_to_replica1_db; - -my $ua = LWP::UserAgent::Determined->new( - agent => 'Bugzilla', - keep_alive => 10, - requests_redirectable => [qw(GET HEAD DELETE PUT)], -); -$ua->timing('1,2,4,8,16,32'); -$ua->timeout(30); -if (my $proxy = Bugzilla->params->{proxy_url}) { - $ua->proxy(['https', 'http'], $proxy); -} - -# This date will be added to each object as it is being sent -if (!$snapshot_date) { - $snapshot_date = $dbh->selectrow_array( - 'SELECT ' . $dbh->sql_date_format('LOCALTIMESTAMP(0)', '%Y-%m-%d')); -} - -# Excluded bugs: List of bug ids that we should not send data for to ETL (i.e. Legal, etc.) -our %excluded_bugs = (); - -# Bugs that are private to one or more groups -our %private_bugs = (); - -# In order to avoid entering duplicate data, we will first query BigQuery -# to make sure other entries with this date are not already present. -check_for_duplicates(); - -# Process each table to be sent to ETL -process_bugs(); -process_attachments(); -process_flags(); -process_flag_state_activity(); -process_tracking_flags(); -process_keywords(); -process_see_also(); -process_users(); - -process_two_columns( - 'bug_mentors', 'bug_mentors', - ['bug_id', 'user_id'], - ['bug_id', 'user_id'] -); -process_two_columns( - 'dependencies', 'bug_dependencies', - ['blocked', 'dependson'], - ['bug_id', 'depends_on_id'] -); -process_two_columns( - 'regressions', 'bug_regressions', - ['regresses', 'regressed_by'], - ['bug_id', 'regresses_id'] -); -process_two_columns( - 'duplicates', 'bug_duplicates', - ['dupe', 'dupe_of'], - ['bug_id', 'duplicate_of_id'] -); - -# If we are done, remove the lock -delete_lock(); - -### Functions - -sub process_bugs { - my $table_name = 'bugs'; - my $total_count = 0; - my $last_offset = 0; - - logger("Processing $table_name"); - - my $sth = $dbh->prepare( - 'SELECT bug_id, delta_ts FROM bugs ORDER BY bug_id LIMIT ? OFFSET ?'); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($id, $mod_time) = @{$row}; - - logger("Processing id $id with mod_time of $mod_time."); - - # First check to see if we have a cached version with the same modification date - my $data = get_cache($id, $table_name, $mod_time); - - if (!$data) { - logger("$table_name id $id with time $mod_time not found in cache.", - DEBUG_OUTPUT); - - my $obj = Bugzilla::Bug->new($id); - - if (!$obj) { - logger("Object $id not loaded from database or no longer exists"); - next; - } - - logger("Object $id loaded from database", DEBUG_OUTPUT); - - my $bug_is_private = scalar @{$obj->groups_in}; - - if (any { $obj->product eq $_ } EXCLUDE_PRODUCTS) { - $excluded_bugs{$obj->id} = 1; - $total_count++; - next; - } - - $private_bugs{$obj->id} = 1 if $bug_is_private; - - # Standard non-sensitive fields - $data = { - id => $obj->id, - status => $obj->bug_status, - type => $obj->bug_type, - component => $obj->component, - creation_ts => $obj->creation_ts, - updated_ts => $obj->delta_ts, - op_sys => $obj->op_sys, - product => $obj->product, - platform => $obj->rep_platform, - reporter_id => $obj->reporter->id, - version => $obj->version, - team_name => $obj->component_obj->team_name, - classification => $obj->classification, - comment_count => $obj->comment_count, - vote_count => $obj->votes, - }; - - # Fields that require custom values based on criteria - $data->{assignee_id} - = $obj->assigned_to->login ne 'nobody@mozilla.org' - ? $obj->assigned_to->id - : undef; - $data->{url} - = (!$bug_is_private && $obj->bug_file_loc) ? $obj->bug_file_loc : undef; - $data->{severity} = $obj->bug_severity ne '--' ? $obj->bug_severity : undef; - $data->{crash_signature} - = (!$bug_is_private && $obj->cf_crash_signature) - ? $obj->cf_crash_signature - : undef; - $data->{priority} = $obj->priority ne '--' ? $obj->priority : undef; - $data->{resolution} = $obj->resolution ? $obj->resolution : undef; - $data->{summary} = !$bug_is_private ? $obj->short_desc : undef; - $data->{whiteboard} - = (!$bug_is_private && $obj->status_whiteboard) - ? $obj->status_whiteboard - : undef; - $data->{milestone} - = $obj->target_milestone ne '---' ? $obj->target_milestone : undef; - $data->{is_public} = $bug_is_private ? true : false; - $data->{cc_count} = scalar @{$obj->cc || []}; - - # If more than one group, then pick the one with the least of amount of members - if (!$bug_is_private) { - $data->{group} = undef; - } - elsif (scalar @{$obj->groups_in} == 1) { - my $groups = $obj->groups_in; - $data->{group} = $groups->[0]->name; - } - else { - $data->{group} = get_multi_group_value($obj); - } - - # Store a copy of the data for use in later executions - store_cache($obj->id, $table_name, $obj->delta_ts, $data); - } - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger("total count: $total_count, last offset: $last_offset", DEBUG_OUTPUT); - - # Send the rows to the server - send_data($table_name, \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -sub process_attachments { - my $table_name = 'attachments'; - my $total_count = 0; - my $last_offset = 0; - - logger("Processing $table_name."); - - my $sth - = $dbh->prepare( - 'SELECT attach_id, modification_time FROM attachments ORDER BY attach_id LIMIT ? OFFSET ?' - ); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($id, $mod_time) = @{$row}; - - logger("Processing id $id with mod_time of $mod_time."); - - # First check to see if we have a cached version with the same modification date - my $data = get_cache($id, $table_name, $mod_time); - - if (!$data) { - logger("$table_name id $id with time $mod_time not found in cache.", - DEBUG_OUTPUT); - - my $obj = Bugzilla::Attachment->new($id); - - if (!$obj) { - logger("Object $id not loaded from database or no longer exists"); - next; - } - - logger("Object $id loaded from database", DEBUG_OUTPUT); - - if ($excluded_bugs{$obj->bug_id}) { - $total_count++; - next; - } - - # Standard non-sensitive fields - $data = { - id => $obj->id, - bug_id => $obj->bug_id, - creation_ts => $obj->attached, - content_type => $obj->contenttype, - updated_ts => $obj->modification_time, - submitter_id => $obj->attacher->id, - is_obsolete => ($obj->isobsolete ? true : false), - }; - - # Fields that require custom values based on criteria - my $bug_is_private = exists $private_bugs{$obj->bug_id}; - $data->{description} = !$bug_is_private ? $obj->description : undef; - $data->{filename} = !$bug_is_private ? $obj->filename : undef; - - # Store a new copy of the data for use later - store_cache($obj->id, $table_name, $obj->modification_time, $data); - } - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger("total count: $total_count, last offset: $last_offset", DEBUG_OUTPUT); - - # Send the rows to the server - send_data($table_name, \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -sub process_flags { - my $table_name = 'flags'; - my $total_count = 0; - my $last_offset = 0; - - logger("Processing $table_name."); - - my $sth = $dbh->prepare( - 'SELECT id, modification_date FROM flags ORDER BY id LIMIT ? OFFSET ?'); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($id, $mod_time) = @{$row}; - - logger("Processing id $id with mod_time of $mod_time."); - - # First check to see if we have a cached version with the same modification date - my $data = get_cache($id, $table_name, $mod_time); - - if (!$data) { - logger("$table_name id $id with time $mod_time not found in cache.", - DEBUG_OUTPUT); - - my $obj = Bugzilla::Flag->new($id); - - if (!$obj) { - logger("Object $id not loaded from database or no longer exists"); - next; - } - - logger("Object $id loaded from database", DEBUG_OUTPUT); - - if ($excluded_bugs{$obj->bug_id}) { - $total_count++; - next; - } - - $data = { - id => $obj->id, - attachment_id => $obj->attach_id || undef, - bug_id => $obj->bug_id, - creation_ts => $obj->creation_date, - updated_ts => $obj->modification_date, - requestee_id => $obj->requestee_id, - setter_id => $obj->setter_id, - name => $obj->type->name, - value => $obj->status, - }; - - # Store a new copy of the data for use later - store_cache($obj->id, $table_name, $obj->modification_date, $data); - } - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger("total count: $total_count, last offset: $last_offset", DEBUG_OUTPUT); - - # Send the rows to the server - send_data($table_name, \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -# Process flags that were removed today using the flag_state_activity table -# These entries will also go into the flags table in BigQuery. -sub process_flag_state_activity { - my $table_name = 'flag_state_activity'; - my $total_count = 0; - my $last_offset = 0; - - logger("Processing $table_name."); - - my $sth - = $dbh->prepare( - 'SELECT id, flag_when FROM flag_state_activity WHERE status = \'X\' AND flag_when LIKE \'' - . $snapshot_date - . ' %\' ORDER BY id LIMIT ? OFFSET ?'); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($id, $mod_time) = @{$row}; - - logger("Processing id $id with mod_time of $mod_time."); - - # First check to see if we have a cached version with the same modification date - my $data = get_cache($id, $table_name, $mod_time); - - if (!$data) { - logger("$table_name id $id with time $mod_time not found in cache.", - DEBUG_OUTPUT); - - my $obj = Bugzilla::Extension::Review::FlagStateActivity->new($id); - - if (!$obj) { - logger("Object $id not loaded from database or no longer exists"); - next; - } - - logger("Object $id loaded from database", DEBUG_OUTPUT); - - if ($excluded_bugs{$obj->bug_id}) { - $total_count++; - next; - } - - $data = { - attachment_id => $obj->attachment_id || undef, - bug_id => $obj->bug_id, - creation_ts => $obj->flag_when, - updated_ts => $obj->flag_when, - requestee_id => $obj->requestee_id, - setter_id => $obj->setter_id, - name => $obj->type->name, - value => $obj->status, - }; - - # Store a new copy of the data for use later - store_cache($obj->id, $table_name, $obj->flag_when, $data); - } - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger("total count: $total_count, last offset: $last_offset", DEBUG_OUTPUT); - - # Send the rows to the server - send_data('flags', \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -sub process_tracking_flags { - my $table_name = 'tracking_flags'; - my $total_count = 0; - my $last_offset = 0; - - logger("Processing $table_name."); - - my $sth = $dbh->prepare( - 'SELECT tracking_flags_bugs.id, tracking_flags.name, tracking_flags_bugs.bug_id, tracking_flags_bugs.value - FROM tracking_flags_bugs - JOIN tracking_flags - ON tracking_flags_bugs.tracking_flag_id = tracking_flags.id - ORDER BY tracking_flags_bugs.id LIMIT ? OFFSET ?' - ); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($id, $name, $bug_id, $value) = @{$row}; - - if ($excluded_bugs{$bug_id}) { - $total_count++; - next; - } - - # Standard fields - my $data = {id => $id, bug_id => $bug_id}; - - # Fields that require custom values based on other criteria - if (exists $private_bugs{$bug_id}) { - $data->{name} = undef; - $data->{value} = undef; - } - else { - $data->{name} = $name; - $data->{value} = $value; - } - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger("total count: $total_count last offset: $last_offset", DEBUG_OUTPUT); - - # Send the rows to the server - send_data($table_name, \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -sub process_keywords { - my $table_name = 'keywords'; - my $total_count = 0; - my $last_offset = 0; - - logger("Processing $table_name."); - - my $sth = $dbh->prepare( - 'SELECT bug_id, keyworddefs.name - FROM keywords - JOIN keyworddefs - ON keywords.keywordid = keyworddefs.id - ORDER BY bug_id LIMIT ? OFFSET ?' - ); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($bug_id, $keyword) = @{$row}; - - if ($excluded_bugs{$bug_id}) { - $total_count++; - next; - } - - # Standard fields - my $data = {bug_id => $bug_id}; - - # Fields that require custom values based on other criteria - $data->{keyword} = !exists $private_bugs{$bug_id} ? $keyword : undef; - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger( - "total count: $total_count, count: $total_count, last offset: $last_offset", - DEBUG_OUTPUT); - - # Send the rows to the server - send_data($table_name, \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -sub process_see_also { - my $table_name = 'see_also'; - my $total_count = 0; - my $last_offset = 0; - - logger("Processing $table_name."); - - my $sth - = $dbh->prepare( - 'SELECT bug_id, value, class FROM bug_see_also ORDER BY bug_id LIMIT ? OFFSET ?' - ); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($bug_id, $value, $class) = @{$row}; - - if ($excluded_bugs{$bug_id}) { - $total_count++; - next; - } - - # Standard fields - my $data = {bug_id => $bug_id,}; - - # Fields that require custom values based on other criteria - if ($private_bugs{$bug_id}) { - $data->{url} = undef; - } - elsif ($class =~ /::Local/) { - $data->{url} = Bugzilla->localconfig->urlbase . 'show_bug.cgi?id=' . $value; - } - else { - $data->{url} = $value; - } - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger("total count: $total_count, last offset: $last_offset", DEBUG_OUTPUT); - - # Send the rows to the server - send_data($table_name, \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -sub process_users { - my $table_name = 'users'; - my $total_count = 0; - my $last_offset = 0; - - logger("Processing $table_name."); - - my $sth - = $dbh->prepare( - 'SELECT userid, modification_ts FROM profiles ORDER BY userid LIMIT ? OFFSET ?' - ); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($id, $mod_time) = @{$row}; - - logger("Processing id $id with mod_time of $mod_time."); - - # Set the mod time to an arbitrary value for caching purposes if its - # real mod time is not yet been set to a real value. - $mod_time = '1970-01-01 12:00:00' if !$mod_time; - - # First check to see if we have a cached version with the same modification date - my $data = get_cache($id, $table_name, $mod_time); - - if (!$data) { - logger("$table_name id $id with time $mod_time not found in cache.", - DEBUG_OUTPUT); - - my $obj = Bugzilla::User->new($id); - - if (!$obj) { - logger("Object $id not loaded from database or no longer exists"); - next; - } - - logger("Object $id loaded from database", DEBUG_OUTPUT); - - # Standard fields - $data = { - id => $obj->id, - last_seen => - ($obj->last_seen_date ? $obj->last_seen_date . ' 00:00:00' : undef), - email => $obj->email, - is_new => ($obj->is_new ? true : false), - }; - - # Fields that require custom values based on criteria - $data->{nick} = $obj->nick ? $obj->nick : undef; - $data->{name} = $obj->name ? $obj->name : undef; - $data->{is_staff} - = $obj->in_group('mozilla-employee-confidential') ? true : false; - $data->{is_trusted} = $obj->in_group('editbugs') ? true : false; - $data->{ldap_email} = $obj->ldap_email ? $obj->ldap_email : undef; - - # Store a new copy of the data for use later - store_cache($obj->id, $table_name, $obj->modification_ts, $data); - } - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger("total count: $total_count, last offset: $last_offset", DEBUG_OUTPUT); - - # Send the rows to the server - send_data($table_name, \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -sub process_two_columns { - my ($table_name, $bq_name, $column_names, $data_names) = @_; - my $total_count = 0; - my $last_offset = 0; - my $columns_string = join ', ', @{$column_names}; - my $order_by = $column_names->[0]; - - logger("Processing $table_name."); - - my $sth = $dbh->prepare( - "SELECT $columns_string FROM $table_name ORDER BY $order_by LIMIT ? OFFSET ?"); - - $sth->execute(API_BLOCK_COUNT, $last_offset); - - while (my @rows = @{$sth->fetchall_arrayref() || []}) { - my @results; - - foreach my $row (@rows) { - my ($value1, $value2) = @{$row}; - - if ($excluded_bugs{$value1}) { - $total_count++; - next; - } - - logger("Processing values $value1, $value2 for $table_name."); - - my $data = {$data_names->[0] => $value1, $data_names->[1] => $value2,}; - - push @results, $data; - - $total_count++; - } - - # Send a batch of data and then run query for the next batch of rows - logger("total count: $total_count, last offset: $last_offset", DEBUG_OUTPUT); - - # Send the rows to the server - send_data($bq_name, \@results, $total_count); - - $last_offset += API_BLOCK_COUNT; - $sth->execute(API_BLOCK_COUNT, $last_offset); - } -} - -sub get_cache { - my ($id, $table, $timestamp) = @_; - - if ($no_cache{$table}) { - logger("Retrieving cached data is disabled for $table."); - return undef; - } - - logger("Retreiving data from $table for $id with time $timestamp.", - DEBUG_OUTPUT); - - try { - # Retrieve compressed JSON from cache table if it exists - my $gzipped_data = $dbh->selectrow_array( - 'SELECT data FROM bmo_etl_cache WHERE id = ? AND table_name = ? AND snapshot_date = ?', - undef, $id, $table, $timestamp - ); - return undef if !$gzipped_data; - - # First uncompress the JSON and then decode it back to Perl data - my $data; - unless (gunzip \$gzipped_data => \$data) { - delete_lock(); - die "gunzip failed: $GunzipError\n"; - } - return decode_json($data); - } - catch { - # Log the failure and return undef - WARN("ERROR: Unable to retrieve cached data from database: $_"); - return undef; - } -} - -sub store_cache { - my ($id, $table, $timestamp, $data) = @_; - - if ($no_cache{$table}) { - logger("Storing cached data is disabled for $table."); - return undef; - } - - logger("Storing data into $table for $id with time $timestamp.", DEBUG_OUTPUT); - - # Encode the perl data into JSON - $data = encode_json($data); - - # Compress the JSON to save space in the DB - my $gzipped_data; - unless (gzip \$data => \$gzipped_data) { - delete_lock(); - die "gzip failed: $GzipError\n"; - } - - # We need to use the main DB for write operations - my $main_dbh = Bugzilla->dbh_main; - - try { - # Clean out outdated JSON - $main_dbh->do('DELETE FROM bmo_etl_cache WHERE id = ? AND table_name = ?', - undef, $id, $table); - - # Enter new cached JSON - $main_dbh->do( - 'INSERT INTO bmo_etl_cache (id, table_name, snapshot_date, data) VALUES (?, ?, ?, ?)', - undef, $id, $table, $timestamp, $gzipped_data - ); - } - catch { - # Log the failure - WARN("ERROR: Unable to store cache data in database: $_"); - } -} - -sub send_data { - my ($table, $all_rows, $current_count) = @_; - - logger( - 'Sending ' . scalar @{$all_rows} . " rows to table $table using BigQuery API"); - - # Add the same snapshot date to every row sent - foreach my $row (@{$all_rows}) { - $row->{snapshot_date} = $snapshot_date; - } - - my @json_rows = (); - foreach my $row (@{$all_rows}) { - push @json_rows, {json => $row}; - } - - my $big_query = {rows => \@json_rows}; - - if ($test) { - my $filename - = bz_locations()->{'datadir'} . '/' - . $snapshot_date . '-' - . $table . '-' - . $current_count . '.json'; - - logger("Writing data to $filename."); - - my $fh = path($filename)->open('>>'); - print $fh encode_json($big_query) . "\n"; - unless (close $fh) { - delete_lock(); - die "Could not close $filename: $!\n"; - } - - return; - } - - my $http_headers = HTTP::Headers->new; - - # Do not attempt to get access token if running in test environment - if ($base_url !~ /^http:\/\/[^\/]+:9050/) { - my $access_token = _get_access_token(); - $http_headers->header(Authorization => 'Bearer ' . $access_token); - } - - my $full_path = sprintf 'projects/%s/datasets/%s/tables/%s/insertAll', - $project_id, $dataset_id, $table; - - logger("Sending to $base_url/$full_path", DEBUG_OUTPUT); - - my $request = HTTP::Request->new('POST', "$base_url/$full_path", $http_headers); - $request->header('Content-Type' => 'application/json'); - - logger('Encoding content into JSON.', DEBUG_OUTPUT); - - $request->content(encode_json($big_query)); - - logger('Sending request', DEBUG_OUTPUT); - - my $response = $ua->request($request); - - logger($response->content, DEBUG_OUTPUT); - - my $result = decode_json($response->content); - - if (!$response->is_success - || (exists $result->{insertErrors} && @{$result->{insertErrors}})) - { - delete_lock(); - die "Google Big Query insert failure:\nRequest:\n" - . $request->content - . "\n\nResponse:\n" - . $response->content . "\n"; - } -} - -sub _get_access_token { - state $access_token; # We should only need to get this once - state $token_expiry; - - logger('Checking for acess token', DEBUG_OUTPUT); - - # If we already have a token and it has not expired yet, just return it - if ($access_token && time < $token_expiry) { - logger('Previous access token found', DEBUG_OUTPUT); - return $access_token; - } - -# Google Kubernetes allows for the use of Workload Identity. This allows -# us to link two service accounts together and give special access for applications -# running under Kubernetes. We use the special access to get an OAuth2 access_token -# that can then be used for accessing the the Google API such as BigQuery. - my $url - = sprintf - 'http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/%s/token', - Bugzilla->params->{bmo_etl_service_account}; - - my $http_headers = HTTP::Headers->new; - $http_headers->header('Metadata-Flavor' => 'Google'); - - my $request = HTTP::Request->new('GET', $url, $http_headers); - - my $res = $ua->request($request); - - if (!$res->is_success) { - delete_lock(); - die 'Google access token failure: ' . $res->content . "\n"; - } - - my $result = decode_json($res->decoded_content); - $access_token = $result->{access_token}; - $token_expiry = time + $result->{expires_in}; - - logger('New access token returned', DEBUG_OUTPUT); - - return $access_token; -} - -# If a previous process is performing an export to BigQuery, then -# we must check the lock table and exit if true. -sub check_and_set_lock { - return if $test; # No need if just dumping test files - - logger('Checking for previous lock or setting new one', DEBUG_OUTPUT); - - my $dbh_main = Bugzilla->dbh_main; - - # Clear out any locks that are greater than 24h old - $dbh_main->do('DELETE FROM bmo_etl_locked WHERE creation_ts < ' - . $dbh_main->sql_date_math('NOW()', '-', 24, 'HOUR')); - - # Now check for any pre-existing locks and do not proceed if one found - my $locked = $dbh_main->selectrow_array('SELECT COUNT(*) FROM bmo_etl_locked'); - if ($locked) { - die "Another process has set a lock. Exiting\n"; - } - - logger('Previous lock not found. Setting new one.', DEBUG_OUTPUT); - - $dbh_main->do( - 'INSERT INTO bmo_etl_locked (value, creation_ts) VALUES (?, NOW())', - undef, 'locked'); -} - -# Delete lock from bmo_etl_locked -sub delete_lock { - logger("Deleting lock in database."); - Bugzilla->dbh_main->do('DELETE FROM bmo_etl_locked'); -} - -sub check_for_duplicates { - return if $test; # no need if just dumping test files - - logger("Checking for duplicate data for snapshot date $snapshot_date."); - - my $http_headers = HTTP::Headers->new; - - # Do not attempt to get access token if running in test environment - if ($base_url !~ /^http:\/\/[^\/]+:9050/) { - my $access_token = _get_access_token(); - $http_headers->header(Authorization => 'Bearer ' . $access_token); - } - - my $full_path = "projects/$project_id/queries"; - - logger("Querying $base_url/$full_path", DEBUG_OUTPUT); - - my $query = { - query => - "SELECT count(*) FROM ${project_id}.${dataset_id}.bugs WHERE snapshot_date = '$snapshot_date';", - useLegacySql => false, - }; - - my $request = HTTP::Request->new('POST', "$base_url/$full_path", $http_headers); - $request->header('Content-Type' => 'application/json'); - $request->content(encode_json($query)); - - logger(encode_json($query), DEBUG_OUTPUT); - - my $res = $ua->request($request); - if (!$res->is_success) { - delete_lock(); - die 'Google Big Query query failure: ' . $res->content . "\n"; - } - - logger($res->content, DEBUG_OUTPUT); - - my $result = decode_json($res->content); - - my $row_count = $result->{rows}->[0]->{f}->[0]->{v}; - - # Do not export if we have any rows with this snapshot date. - if ($row_count) { - delete_lock(); - die "Duplicate data found for snapshot date $snapshot_date\n"; - } -} - -sub get_multi_group_value { - my ($bug) = @_; - - logger('Checking for multi group values.', DEBUG_OUTPUT); - - my $smallest_group_name = undef; - my $smallest_group_count = 0; - - foreach my $group (@{$bug->groups_in}) { - my $user_count = 0; - my $member_data = $group->members_complete; - foreach my $type (keys %{$member_data}) { - $user_count += scalar @{$member_data->{$type}}; - } - if ($user_count < $smallest_group_count) { - $smallest_group_count = $user_count; - $smallest_group_name = $group->name; - } - } - - logger("Smallest group members: $smallest_group_name") if $smallest_group_name; - - return $smallest_group_name; -} - -sub logger { - my ($message, $level) = @_; - - # Skip if -q or --quiet was specified - return if $quiet; - - # Skip if --debug was not specified and this is a DEBUG level entry - return if (defined $level && $level == DEBUG_OUTPUT && !$debug); - - # Otherwise just print - print time() . ": $message\n"; -} - -1; diff --git a/extensions/BMO/lib/Daemon.pm b/extensions/BMO/lib/Daemon.pm new file mode 100644 index 0000000000..087afdd84e --- /dev/null +++ b/extensions/BMO/lib/Daemon.pm @@ -0,0 +1,138 @@ +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. +# +# This Source Code Form is "Incompatible With Secondary Licenses", as +# defined by the Mozilla Public License, v. 2.0. + +package Bugzilla::Extension::BMO::Daemon; + +use 5.10.1; +use strict; +use warnings; + +use Bugzilla; +use Bugzilla::Constants; +use Bugzilla::Extension::BMO::ETL; +use Bugzilla::Extension::BMO::ETL::Export; +use Bugzilla::Logging; + +use Carp qw(confess); +use Daemon::Generic; +use File::Basename; +use Mojo::Util qw(dumper); +use Pod::Usage; + +sub start { + newdaemon(); +} + +sub gd_preconfig { + my $self = shift; + my $pidfile = $self->{gd_args}{pidfile}; + if (!$pidfile) { + $pidfile = bz_locations()->{datadir} . '/' . $self->{gd_progname} . '.pid'; + } + return (pidfile => $pidfile); +} + +sub gd_getopt { + my $self = shift; + $self->SUPER::gd_getopt(); + if ($self->{gd_args}{progname}) { + $self->{gd_progname} = $self->{gd_args}{progname}; + } + else { + $self->{gd_progname} = basename($0); + } + $self->{_original_zero} = $0; + $0 = $self->{gd_progname}; +} + +sub gd_postconfig { + my $self = shift; + $0 = delete $self->{_original_zero}; +} + +sub gd_more_opt { + my $self = shift; + return ( + 'pidfile=s' => \$self->{gd_args}{pidfile}, + 'n=s' => \$self->{gd_args}{progname}, + 't|test' => \$self->{gd_args}->{test}, + 's|snapshot-date=s' => \$self->{gd_args}->{snapshot_date}, + ); +} + +sub gd_usage { + pod2usage({-verbose => 0, -exitval => 'NOEXIT'}); + return 0; +} + +sub gd_redirect_output { + my $self = shift; + my $filename = bz_locations()->{datadir} . '/' . $self->{gd_progname} . '.log'; + open STDERR, '>>', $filename or (print "could not open stderr: $!" && exit 1); + close STDOUT; + open STDOUT, '>&', STDERR or die "redirect STDOUT -> STDERR: $!"; + $SIG{HUP} = sub { + close STDERR; + open STDERR, '>>', $filename or (print "could not open stderr: $!" && exit 1); + }; +} + +sub gd_setup_signals { + my $self = shift; + $self->SUPER::gd_setup_signals(); + $SIG{TERM} = sub { $self->gd_quit_event(); } +} + +sub gd_other_cmd { + my ($self) = shift; + if ($ARGV[0] eq 'once') { + try { + my $export = Bugzilla::Extension::BMO::ETL::Export->new( + debug => ($self->{debug} ? 1 : 0), + test => ($self->{gd_args}->{test} ? 1 : 0), + snapshot_date => ( + $self->{gd_args}->{snapshot_date} ? $self->{gd_args}->{snapshot_date} : 0 + ), + ); + $export->run_export; + exit 0; + } + catch { + FATAL($_); + } + finally { + try { + Bugzilla->_cleanup(); + } + catch { + FATAL("Error in _cleanup: $_"); + exit 1; + } + }; + } + $self->SUPER::gd_other_cmd(); +} + +sub gd_run { + my $self = shift; + $::SIG{__DIE__} = \&Carp::confess if $self->{debug}; + my $etl = Bugzilla::Extension::BMO::ETL->new; + $etl->is_daemon(1); + $etl->start( + debug => ($self->{debug} ? 1 : 0), + test => ($self->{gd_args}->{test} ? 1 : 0), + snapshot_date => ( + $self->{gd_args}->{snapshot_date} ? $self->{gd_args}->{snapshot_date} : 0 + ), + ); +} + +sub gd_check { + return 'OK'; +} + +1; diff --git a/extensions/BMO/lib/ETL.pm b/extensions/BMO/lib/ETL.pm new file mode 100644 index 0000000000..f266d3034a --- /dev/null +++ b/extensions/BMO/lib/ETL.pm @@ -0,0 +1,84 @@ +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. +# +# This Source Code Form is "Incompatible With Secondary Licenses", as +# defined by the Mozilla Public License, v. 2.0. +package Bugzilla::Extension::BMO::ETL; + +use 5.10.1; +use strict; +use warnings; +use lib qw(. lib local/lib/perl5); + +use Bugzilla; +use Bugzilla::Logging; +use Bugzilla::Extension::BMO::ETL::Export; + +use IO::Async::Timer::Periodic; +use IO::Async::Loop; +use IO::Async::Signal; +use Mojo::Util qw(dumper); +use Moo; +use Try::Tiny; + +has 'is_daemon' => (is => 'rw', default => 0); + +# ETL polling value +use constant ETL_TIMEOUT => 60; +use constant ETL_POLL_SECONDS => 5; +use constant ETL_RUN_HOUR => 23; + +sub start { + my ($self, %options) = @_; + + my $sig_alarm = IO::Async::Signal->new( + name => 'ALRM', + on_receipt => sub { + FATAL('Timeout reached while executing etl export'); + exit 1; + }, + ); + + my $etl_timer = IO::Async::Timer::Periodic->new( + first_interval => 0, + interval => ETL_POLL_SECONDS, + reschedule => 'drift', + on_tick => sub { + try { + alarm ETL_TIMEOUT; + + # Run the export only at the specified hour + my $datetime = DateTime->now(time_zone => 'local'); + if ($datetime->hour == ETL_RUN_HOUR) { + INFO('Starting export: ' . $datetime->ymd . ' ' . $datetime->hms); + my $export = Bugzilla::Extension::BMO::ETL::Export->new(%options); + $export->run_export; + } + else { + INFO('Skipping ETL export, not the scheduled hour: ' . $datetime->ymd . ' ' . $datetime->hms); + } + } + catch { + FATAL($_); + } + finally { + alarm 0; + try { + Bugzilla->_cleanup(); + } + catch { + FATAL("Error in _cleanup: $_"); + exit 1; + } + }; + }, + ); + + my $loop = IO::Async::Loop->new; + $loop->add($etl_timer); + $etl_timer->start; + $loop->run; +} + +1; diff --git a/extensions/BMO/lib/ETL/Export.pm b/extensions/BMO/lib/ETL/Export.pm new file mode 100644 index 0000000000..163f6f9baa --- /dev/null +++ b/extensions/BMO/lib/ETL/Export.pm @@ -0,0 +1,1062 @@ +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. +# +# This Source Code Form is "Incompatible With Secondary Licenses", as +# defined by the Mozilla Public License, v. 2.0. +package Bugzilla::Extension::BMO::ETL::Export; + +use 5.10.1; +use strict; +use warnings; +use lib qw(. lib local/lib/perl5); + +use Bugzilla; +use Bugzilla::Attachment; +use Bugzilla::Bug; +use Bugzilla::Constants; +use Bugzilla::Flag; +use Bugzilla::Group; +use Bugzilla::Logging; +use Bugzilla::User; +use Bugzilla::Util qw(with_writable_database); +use Bugzilla::Extension::Review::FlagStateActivity; + +use HTTP::Headers; +use HTTP::Request; +use IO::Compress::Gzip qw(gzip $GzipError); +use IO::Uncompress::Gunzip qw(gunzip $GunzipError); +use List::Util qw(any); +use LWP::UserAgent::Determined; +use Moo; +use Mojo::File qw(path); +use Mojo::JSON qw(decode_json encode_json false true); +use Try::Tiny; +use Types::Standard qw(Bool HashRef Object Str); + +has 'debug' => (is => 'ro', isa => Bool); +has 'excluded_bugs' => (is => 'rw', isa => HashRef); +has 'private_bugs' => (is => 'rw', isa => HashRef); +has 'snapshot_date' => (is => 'rw', isa => Str); +has 'test' => (is => 'ro', isa => Bool); +has 'ua' => (is => 'lazy', isa => Object); + +# BigQuery API cannot handle payloads larger than 10MB so +# we will send data in blocks. +use constant API_BLOCK_COUNT => 1000; + +# Products which we should not send data to ETL such as Legal, etc. +use constant EXCLUDE_PRODUCTS => ('Legal',); + +# Log levels +use constant DEBUG_OUTPUT => 1; + +sub run_export { + my ($self) = @_; + + # Sanity checks + my $params = Bugzilla->params; + if (!$params->{bmo_etl_enabled}) { + WARN('BMO ETL: Not enabled'); + return; + } + + if (!$params->{bmo_etl_base_url}) { + WARN('BMO ETL: Invalid BigQuery base URL'); + return; + } + + if (!$params->{bmo_etl_project_id}) { + WARN('BMO ETL: Invalid BigQuery product ID'); + return; + } + + if (!$params->{bmo_etl_dataset_id}) { + WARN('BMO ETL: Invalid BigQuery dataset ID'); + return; + } + + # Use replica if available + my $dbh = Bugzilla->switch_to_shadow_db(); + + # This date will be added to each object as it is being sent + if (!$self->snapshot_date) { + my $snapshot_date = $dbh->selectrow_array( + 'SELECT ' . $dbh->sql_date_format('LOCALTIMESTAMP(0)', '%Y-%m-%d') + ); + $self->snapshot_date($snapshot_date); + } + + # Process each table to be sent to ETL + $self->process_bugs(); + $self->process_attachments(); + $self->process_flags(); + $self->process_flag_state_activity(); + $self->process_tracking_flags(); + $self->process_keywords(); + $self->process_see_also(); + $self->process_users(); + + $self->process_two_columns( + 'bug_mentors', 'bug_mentors', + ['bug_id', 'user_id'], + ['bug_id', 'user_id'] + ); + $self->process_two_columns( + 'dependencies', 'bug_dependencies', + ['blocked', 'dependson'], + ['bug_id', 'depends_on_id'] + ); + $self->process_two_columns( + 'regressions', 'bug_regressions', + ['regresses', 'regressed_by'], + ['bug_id', 'regresses_id'] + ); + $self->process_two_columns( + 'duplicates', 'bug_duplicates', + ['dupe', 'dupe_of'], + ['bug_id', 'duplicate_of_id'] + ); +} + +### Functions + +sub process_bugs { + my ($self) = @_; + my $table_name = 'bugs'; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name"); + + my $sth + = $dbh->prepare( + 'SELECT bug_id AS id, delta_ts AS modification_time FROM bugs ORDER BY bug_id LIMIT ? OFFSET ?' + ); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($id, $mod_time) = $sth->fetchrow_array()) { + $self->logger("Processing id $id with mod_time of $mod_time."); + + # First check to see if we have a cached version with the same modification date + my $data = $self->get_cache($id, $table_name, $mod_time); + + if (!$data) { + $self->logger("$table_name id $id with time $mod_time not found in cache.", + DEBUG_OUTPUT); + + my $obj = Bugzilla::Bug->new($id); + + if (!$obj) { + $self->logger("Object $id not loaded from database or no longer exists"); + next; + } + + $self->logger("Object $id loaded from database", DEBUG_OUTPUT); + + my $bug_is_private = scalar @{$obj->groups_in}; + + if (any { $obj->product eq $_ } EXCLUDE_PRODUCTS) { + $self->excluded_bugs->{$obj->id} = 1; + $total_count++; + next; + } + + $self->private_bugs->{$obj->id} = 1 if $bug_is_private; + + # Standard non-sensitive fields + $data = { + id => $obj->id, + status => $obj->bug_status, + type => $obj->bug_type, + component => $obj->component, + creation_ts => $obj->creation_ts, + updated_ts => $obj->delta_ts, + op_sys => $obj->op_sys, + product => $obj->product, + platform => $obj->rep_platform, + reporter_id => $obj->reporter->id, + version => $obj->version, + team_name => $obj->component_obj->team_name, + classification => $obj->classification, + comment_count => $obj->comment_count, + vote_count => $obj->votes, + }; + + # Fields that require custom values based on criteria + $data->{assignee_id} + = $obj->assigned_to->login ne 'nobody@mozilla.org' + ? $obj->assigned_to->id + : undef; + $data->{url} + = (!$bug_is_private && $obj->bug_file_loc) ? $obj->bug_file_loc : undef; + $data->{severity} = $obj->bug_severity ne '--' ? $obj->bug_severity : undef; + $data->{crash_signature} + = (!$bug_is_private && $obj->cf_crash_signature) + ? $obj->cf_crash_signature + : undef; + $data->{priority} = $obj->priority ne '--' ? $obj->priority : undef; + $data->{resolution} = $obj->resolution ? $obj->resolution : undef; + $data->{summary} = !$bug_is_private ? $obj->short_desc : undef; + $data->{whiteboard} + = (!$bug_is_private && $obj->status_whiteboard) + ? $obj->status_whiteboard + : undef; + $data->{milestone} + = $obj->target_milestone ne '---' ? $obj->target_milestone : undef; + $data->{is_public} = $bug_is_private ? true : false; + $data->{cc_count} = scalar @{$obj->cc || []}; + + # If more than one group, then pick the one with the least of amount of members + if (!$bug_is_private) { + $data->{group} = undef; + } + elsif (scalar @{$obj->groups_in} == 1) { + my $groups = $obj->groups_in; + $data->{group} = $groups->[0]->name; + } + else { + $data->{group} = $self->get_multi_group_value($obj); + } + + # Store a copy of the data for use in later executions + $self->store_cache($obj->id, $table_name, $obj->delta_ts, $data); + } + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger("total count: $total_count, last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->send_data($table_name, \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the final rows to the server + $self->send_data($table_name, \@results, $total_count) if @results; +} + +sub process_attachments { + my ($self) = @_; + my $table_name = 'attachments'; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name."); + + my $sth + = $dbh->prepare( + 'SELECT attach_id, modification_time FROM attachments ORDER BY attach_id LIMIT ? OFFSET ?' + ); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($id, $mod_time) = $sth->fetchrow_array()) { + $self->logger("Processing id $id with mod_time of $mod_time."); + + # First check to see if we have a cached version with the same modification date + my $data = $self->get_cache($id, $table_name, $mod_time); + + if (!$data) { + $self->logger("$table_name id $id with time $mod_time not found in cache.", + DEBUG_OUTPUT); + + my $obj = Bugzilla::Attachment->new($id); + + if (!$obj) { + $self->logger("Object $id not loaded from database or no longer exists"); + next; + } + + $self->logger("Object $id loaded from database", DEBUG_OUTPUT); + + if ($self->excluded_bugs->{$obj->bug_id}) { + $total_count++; + next; + } + + # Standard non-sensitive fields + $data = { + id => $obj->id, + bug_id => $obj->bug_id, + creation_ts => $obj->attached, + content_type => $obj->contenttype, + updated_ts => $obj->modification_time, + submitter_id => $obj->attacher->id, + is_obsolete => ($obj->isobsolete ? true : false), + }; + + # Fields that require custom values based on criteria + my $bug_is_private = $self->private_bugs->{$obj->bug_id}; + $data->{description} = !$bug_is_private ? $obj->description : undef; + $data->{filename} = !$bug_is_private ? $obj->filename : undef; + + # Store a new copy of the data for use later + $self->store_cache($obj->id, $table_name, $obj->modification_time, $data); + } + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger("total count: $total_count, last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->send_data($table_name, \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the final rows to the server + $self->send_data($table_name, \@results, $total_count) if @results; +} + +sub process_flags { + my ($self) = @_; + my $table_name = 'flags'; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name."); + + my $sth = $dbh->prepare( + 'SELECT id, modification_date FROM flags ORDER BY id LIMIT ? OFFSET ?'); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($id, $mod_time) = $sth->fetchrow_array()) { + $self->logger("Processing id $id with mod_time of $mod_time."); + + # First check to see if we have a cached version with the same modification date + my $data = $self->get_cache($id, $table_name, $mod_time); + + if (!$data) { + $self->logger("$table_name id $id with time $mod_time not found in cache.", + DEBUG_OUTPUT); + + my $obj = Bugzilla::Flag->new($id); + + if (!$obj) { + $self->logger("Object $id not loaded from database or no longer exists"); + next; + } + + $self->logger("Object $id loaded from database", DEBUG_OUTPUT); + + if ($self->excluded_bugs->{$obj->bug_id}) { + $total_count++; + next; + } + + $data = { + id => $obj->id, + attachment_id => $obj->attach_id || undef, + bug_id => $obj->bug_id, + creation_ts => $obj->creation_date, + updated_ts => $obj->modification_date, + requestee_id => $obj->requestee_id, + setter_id => $obj->setter_id, + name => $obj->type->name, + value => $obj->status, + }; + + # Store a new copy of the data for use later + $self->store_cache($obj->id, $table_name, $obj->modification_date, $data); + } + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger("total count: $total_count, last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->send_data($table_name, \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the rows to the server + $self->send_data($table_name, \@results, $total_count) if @results; +} + +# Process flags that were removed today using the flag_state_activity table +# These entries will also go into the flags table in BigQuery. +sub process_flag_state_activity { + my ($self) = @_; + my $table_name = 'flag_state_activity'; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name."); + + my $sth + = $dbh->prepare( + 'SELECT id, flag_when FROM flag_state_activity WHERE status = \'X\' AND flag_when LIKE \'' + . $self->snapshot_date + . ' %\' ORDER BY id LIMIT ? OFFSET ?'); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($id, $mod_time) = $sth->fetchrow_array()) { + $self->logger("Processing id $id with mod_time of $mod_time."); + + # First check to see if we have a cached version with the same modification date + my $data = $self->get_cache($id, $table_name, $mod_time); + + if (!$data) { + $self->logger("$table_name id $id with time $mod_time not found in cache.", + DEBUG_OUTPUT); + + my $obj = Bugzilla::Extension::Review::FlagStateActivity->new($id); + + if (!$obj) { + $self->logger("Object $id not loaded from database or no longer exists"); + next; + } + + $self->logger("Object $id loaded from database", DEBUG_OUTPUT); + + if ($self->excluded_bugs->{$obj->bug_id}) { + $total_count++; + next; + } + + $data = { + attachment_id => $obj->attachment_id || undef, + bug_id => $obj->bug_id, + creation_ts => $obj->flag_when, + updated_ts => $obj->flag_when, + requestee_id => $obj->requestee_id, + setter_id => $obj->setter_id, + name => $obj->type->name, + value => $obj->status, + }; + + # Store a new copy of the data for use later + $self->store_cache($obj->id, $table_name, $obj->flag_when, $data); + } + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger("total count: $total_count, last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->send_data('flags', \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the rows to the server + $self->send_data('flags', \@results, $total_count) if @results; +} + +sub process_tracking_flags { + my ($self) = @_; + my $table_name = 'tracking_flags'; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name."); + + my $sth = $dbh->prepare( + 'SELECT tracking_flags_bugs.id, tracking_flags.name, tracking_flags_bugs.bug_id, tracking_flags_bugs.value + FROM tracking_flags_bugs + JOIN tracking_flags + ON tracking_flags_bugs.tracking_flag_id = tracking_flags.id + ORDER BY tracking_flags_bugs.id LIMIT ? OFFSET ?' + ); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($id, $name, $bug_id, $value) = $sth->fetchrow_array()) { + if ($self->excluded_bugs->{$bug_id}) { + $total_count++; + next; + } + + # Standard fields + my $data = {id => $id, bug_id => $bug_id}; + + # Fields that require custom values based on other criteria + if ($self->private_bugs->{$bug_id}) { + $data->{name} = undef; + $data->{value} = undef; + } + else { + $data->{name} = $name; + $data->{value} = $value; + } + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger("total count: $total_count last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->send_data($table_name, \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the final rows to the server + $self->send_data($table_name, \@results, $total_count) if @results; +} + +sub process_keywords { + my ($self) = @_; + my $table_name = 'keywords'; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name."); + + my $sth = $dbh->prepare( + 'SELECT bug_id, keyworddefs.name + FROM keywords + JOIN keyworddefs + ON keywords.keywordid = keyworddefs.id + ORDER BY bug_id LIMIT ? OFFSET ?' + ); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($bug_id, $keyword) = $sth->fetchrow_array()) { + if ($self->excluded_bugs->{$bug_id}) { + $total_count++; + next; + } + + # Standard fields + my $data = {bug_id => $bug_id}; + + # Fields that require custom values based on other criteria + $data->{keyword} = !$self->private_bugs->{$bug_id} ? $keyword : undef; + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger( + "total count: $total_count, count: $total_count, last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->send_data($table_name, \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the final rows to the server + $self->send_data($table_name, \@results, $total_count) if @results; +} + +sub process_see_also { + my ($self) = @_; + my $table_name = 'see_also'; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name."); + + my $sth + = $dbh->prepare( + 'SELECT bug_id, value, class FROM bug_see_also ORDER BY bug_id LIMIT ? OFFSET ?' + ); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($bug_id, $value, $class) = $sth->fetchrow_array()) { + if ($self->excluded_bugs->{$bug_id}) { + $total_count++; + next; + } + + # Standard fields + my $data = {bug_id => $bug_id,}; + + # Fields that require custom values based on other criteria + if ($self->private_bugs->{$bug_id}) { + $data->{url} = undef; + } + elsif ($class =~ /::Local/) { + $data->{url} = Bugzilla->localconfig->urlbase . 'show_bug.cgi?id=' . $value; + } + else { + $data->{url} = $value; + } + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger("total count: $total_count, last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->send_data($table_name, \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the final rows to the server + $self->send_data($table_name, \@results, $total_count) if @results; +} + +sub process_users { + my ($self) = @_; + my $table_name = 'users'; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name."); + + my $sth + = $dbh->prepare( + 'SELECT userid, modification_ts FROM profiles ORDER BY userid LIMIT ? OFFSET ?' + ); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($id, $mod_time) = $sth->fetchrow_array()) { + $self->logger("Processing id $id with mod_time of $mod_time."); + + # Set the mod time to an arbitrary value for caching purposes if its + # real mod time is not yet been set to a real value. + $mod_time = '1970-01-01 12:00:00' if !$mod_time; + + # First check to see if we have a cached version with the same modification date + my $data = $self->get_cache($id, $table_name, $mod_time); + + if (!$data) { + $self->logger("$table_name id $id with time $mod_time not found in cache.", + DEBUG_OUTPUT); + + my $obj = Bugzilla::User->new($id); + + if (!$obj) { + $self->logger("Object $id not loaded from database or no longer exists"); + next; + } + + $self->logger("Object $id loaded from database", DEBUG_OUTPUT); + + # Standard fields + $data = { + id => $obj->id, + last_seen => + ($obj->last_seen_date ? $obj->last_seen_date . ' 00:00:00' : undef), + email => $obj->email, + is_new => ($obj->is_new ? true : false), + }; + + # Fields that require custom values based on criteria + $data->{nick} = $obj->nick ? $obj->nick : undef; + $data->{name} = $obj->name ? $obj->name : undef; + $data->{is_staff} + = $obj->in_group('mozilla-employee-confidential') ? true : false; + $data->{is_trusted} = $obj->in_group('editbugs') ? true : false; + $data->{ldap_email} = $obj->ldap_email ? $obj->ldap_email : undef; + + # Store a new copy of the data for use later + $self->store_cache($obj->id, $table_name, $obj->modification_ts, $data); + } + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger("total count: $total_count, last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->send_data($table_name, \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the final rows to the server + $self->send_data($table_name, \@results, $total_count) if @results; +} + +sub process_two_columns { + my ($self, $table_name, $bq_name, $column_names, $data_names) = @_; + my $total_count = 0; + my $last_offset = 0; + my @results = (); + my $columns_string = join ', ', @{$column_names}; + my $order_by = $column_names->[0]; + my $dbh = Bugzilla->dbh; + + $self->logger("Processing $table_name."); + + my $sth = $dbh->prepare( + "SELECT $columns_string FROM $table_name ORDER BY $order_by LIMIT ? OFFSET ?"); + + $sth->execute(API_BLOCK_COUNT, $last_offset); + + while (my ($value1, $value2) = $sth->fetchrow_array()) { + if ($self->excluded_bugs->{$value1}) { + $total_count++; + next; + } + + $self->logger("Processing values $value1, $value2 for $table_name."); + + my $data = {$data_names->[0] => $value1, $data_names->[1] => $value2,}; + + push @results, $data; + + $total_count++; + + # Send a batch of data, reset counts and then run query for the next batch of rows + if (scalar @results >= API_BLOCK_COUNT) { + $self->logger("total count: $total_count, last offset: $last_offset", + DEBUG_OUTPUT); + + # Send the rows to the server + $self->$self->send_data($bq_name, \@results, $total_count); + + @results = (); + $last_offset += API_BLOCK_COUNT; + $sth->execute(API_BLOCK_COUNT, $last_offset); + } + } + + # Send the final rows to the server + $self->$self->send_data($bq_name, \@results, $total_count) if @results; +} + +sub get_cache { + my ($self, $id, $table, $timestamp) = @_; + my $dbh = Bugzilla->dbh; + + $self->logger("Retreiving data from $table for $id with time $timestamp.", + DEBUG_OUTPUT); + + try { + # Retrieve compressed JSON from cache table if it exists + my $gzipped_data = $dbh->selectrow_array( + 'SELECT data FROM bmo_etl_cache WHERE id = ? AND table_name = ? AND snapshot_date = ?', + undef, $id, $table, $timestamp + ); + return undef if !$gzipped_data; + + # First uncompress the JSON and then decode it back to Perl data + my $data; + unless (gunzip \$gzipped_data => \$data) { + die "gunzip failed: $GunzipError\n"; + } + return decode_json($data); + } + catch { + # Log the failure and return undef + WARN("ERROR: Unable to retrieve cached data from database: $_"); + return undef; + } +} + +sub store_cache { + my ($self, $id, $table, $timestamp, $data) = @_; + + $self->logger("Storing data into $table for $id with time $timestamp.", + DEBUG_OUTPUT); + + # Encode the perl data into JSON + $data = encode_json($data); + + # Compress the JSON to save space in the DB + my $gzipped_data; + unless (gzip \$data => \$gzipped_data) { + die "gzip failed: $GzipError\n"; + } + + # We need to use the main DB for write operations + my $dbh_main = Bugzilla->dbh_main; + try { + # Clean out outdated JSON + $dbh_main->do('DELETE FROM bmo_etl_cache WHERE id = ? AND table_name = ?', + undef, $id, $table); + + # Enter new cached JSON + $dbh_main->do( + 'INSERT INTO bmo_etl_cache (id, table_name, snapshot_date, data) VALUES (?, ?, ?, ?)', + undef, $id, $table, $timestamp, $gzipped_data + ); + } + catch { + # Log the failure and return undef + WARN("ERROR: Unable to store cached data into database: $_"); + return undef; + }; +} + +sub send_data { + my ($self, $table, $all_rows, $current_count) = @_; + my $project_id = Bugzilla->params->{bmo_etl_project_id}; + my $dataset_id = Bugzilla->params->{bmo_etl_dataset_id}; + my $snapshot_date = $self->snapshot_date; + + $self->logger( + 'Sending ' . scalar @{$all_rows} . " rows to table $table using BigQuery API"); + + # Add the same snapshot date to every row sent + foreach my $row (@{$all_rows}) { + $row->{snapshot_date} = $snapshot_date; + } + + my @json_rows = (); + foreach my $row (@{$all_rows}) { + push @json_rows, {json => $row}; + } + + my $query = {rows => \@json_rows}; + + if ($self->test) { + my $filename + = bz_locations()->{'datadir'} . '/' + . $snapshot_date . '-' + . $table . '-' + . $current_count . '.json'; + + $self->logger("Writing data to $filename."); + + my $fh = path($filename)->open('>>'); + print $fh encode_json($query) . "\n"; + unless (close $fh) { + die "Could not close $filename: $!\n"; + } + + return; + } + + my $path = sprintf 'projects/%s/datasets/%s/tables/%s/insertAll', $project_id, + $dataset_id, $table; + + my $result = $self->call_big_query('POST', $path, $query); + + if (exists $result->{insertErrors} && @{$result->{insertErrors}}) { + die 'Google Big Query insert failure: ' . encode_json($result); + } +} + +sub get_access_token { + my ($self) = @_; + state $access_token; # We should only need to get this once + state $token_expiry; + + $self->logger('Checking for acess token', DEBUG_OUTPUT); + + # If we already have a token and it has not expired yet, just return it + if ($access_token && time < $token_expiry) { + $self->logger('Previous access token found', DEBUG_OUTPUT); + return $access_token; + } + +# Google Kubernetes allows for the use of Workload Identity. This allows +# us to link two service accounts together and give special access for applications +# running under Kubernetes. We use the special access to get an OAuth2 access_token +# that can then be used for accessing the the Google API such as BigQuery. + my $url + = sprintf + 'http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/%s/token', + Bugzilla->params->{bmo_etl_service_account}; + + my $http_headers = HTTP::Headers->new; + $http_headers->header('Metadata-Flavor' => 'Google'); + + my $request = HTTP::Request->new('GET', $url, $http_headers); + + my $res = $self->ua->request($request); + + if (!$res->is_success) { + die 'Google access token failure: ' . $res->content . "\n"; + } + + my $result = decode_json($res->decoded_content); + $access_token = $result->{access_token}; + $token_expiry = time + $result->{expires_in}; + + $self->logger('New access token returned', DEBUG_OUTPUT); + + return $access_token; +} + +sub call_big_query { + my ($self, $method, $path, $data) = @_; + my $base_url = Bugzilla->params->{bmo_etl_base_url}; + + $self->logger("BigQuery request - method: $method, path: $path", DEBUG_OUTPUT); + + my $http_headers = HTTP::Headers->new; + + # Do not attempt to get access token if running in test environment + if ($base_url !~ /^http:\/\/[^\/]+:9050/) { + my $access_token = $self->get_access_token(); + $http_headers->header(Authorization => 'Bearer ' . $access_token); + } + + my $request = HTTP::Request->new($method, "$base_url/$path", $http_headers); + $request->header('Content-Type' => 'application/json'); + + $self->logger('Encoding content into JSON.', DEBUG_OUTPUT); + $self->logger(encode_json($data), DEBUG_OUTPUT); + $request->content(encode_json($data)); + + my $res = $self->ua->request($request); + $self->logger($res->content, DEBUG_OUTPUT); + + if (!$res->is_success) { + die 'Google Big Query query failure: ' . $res->content . "\n"; + } + + my $result = decode_json($res->content); +} + +sub check_duplicate_data { + my ($self, $table) = @_; + my $project_id = Bugzilla->params->{bmo_etl_project_id}; + my $dataset_id = Bugzilla->params->{bmo_etl_dataset_id}; + my $snapshot_date = $self->snapshot_date; + + return 0 if $self->test; # no need if just dumping test files + + $self->logger( + "Checking duplicate data for table $table for snapshot date $snapshot_date."); + + my $query = { + query => + "SELECT count(*) FROM ${project_id}.${dataset_id}.${table} WHERE snapshot_date = '$snapshot_date';", + useLegacySql => false + }; + + my $result + = $self->call_big_query('POST', "projects/$project_id/queries", $query); + + return $result->{rows}->[0]->{f}->[0]->{v} || 0; +} + +sub get_multi_group_value { + my ($self, $bug) = @_; + + $self->logger('Checking for multi group values.', DEBUG_OUTPUT); + + my $smallest_group_name = undef; + my $smallest_group_count = 0; + + foreach my $group (@{$bug->groups_in}) { + my $user_count = 0; + my $member_data = $group->members_complete; + foreach my $type (keys %{$member_data}) { + $user_count += scalar @{$member_data->{$type}}; + } + if ($user_count < $smallest_group_count) { + $smallest_group_count = $user_count; + $smallest_group_name = $group->name; + } + } + + $self->logger("Smallest group members: $smallest_group_name") + if $smallest_group_name; + + return $smallest_group_name; +} + +sub logger { + my ($self, $message, $level) = @_; + + # Skip if --debug was not specified and this is a DEBUG level entry + return if (defined $level && $level == DEBUG_OUTPUT && !$self->debug); + + # Otherwise just print + print time() . ": $message\n"; +} + +sub _build_ua { + my ($self) = @_; + + my $ua = LWP::UserAgent::Determined->new( + agent => 'Bugzilla', + keep_alive => 10, + requests_redirectable => [qw(GET HEAD DELETE PUT)], + ); + + $ua->timing('1,2,4,8,16,32'); + $ua->timeout(30); + + if (my $proxy = Bugzilla->params->{proxy_url}) { + $ua->proxy(['https', 'http'], $proxy); + } + + return $ua; +} + +1; diff --git a/extensions/BMO/t/bmo/bmo_etl.t b/extensions/BMO/t/bmo/bmo_etl.t index 0a7a1234f8..da9d8c889f 100644 --- a/extensions/BMO/t/bmo/bmo_etl.t +++ b/extensions/BMO/t/bmo/bmo_etl.t @@ -119,11 +119,12 @@ my ($attach_id) = keys %{$t->tx->res->json->{attachments}}; ### Section 4: Export data to test files my @cmd = ( - './extensions/BMO/bin/export_bmo_etl.pl', - '--test', '--snapshot-date', $snapshot_date, + './extensions/BMO/bin/bugzilla-etl.pl', + '--test', '--snapshot-date', $snapshot_date, 'once' ); my ($output, $error, $rv) = capture { system @cmd; }; +say "$output\n$error"; ok(!$rv, 'Data exported to test files without error'); ok(glob(bz_locations()->{'datadir'} . '/' . $snapshot_date . '-bugs-*.json'), 'Export test files exist'); @@ -131,11 +132,12 @@ ok(glob(bz_locations()->{'datadir'} . '/' . $snapshot_date . '-bugs-*.json'), ### Section 5: Export data to BigQuery test instance @cmd = ( - './extensions/BMO/bin/export_bmo_etl.pl', - '--snapshot-date', $snapshot_date, + './extensions/BMO/bin/bugzilla-etl.pl', + '--snapshot-date', $snapshot_date, 'once' ); ($output, $error, $rv) = capture { system @cmd; }; +say "$output\n$error"; ok(!$rv, 'Data exported to BigQuery test instance without error'); ### Section 6: Retrieve data from BigQuery instance and verify