Attachment #6902 for bug #5166

View | Details | Raw Unified | Return to bug 5166
Collapse All | Expand All




package C4::Catalog::Zebra;
#
# This file is part of Koha.
#
# Koha is free software; you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# Koha is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# Koha; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
# Suite 330, Boston, MA  02111-1307 USA

# Derived from rebuild_zebra.pl (2005-08-11) Paul Poulain and others
# Rewriten 02/03/2011 by Tomas Cohen Arazi (tomascohen@gmail.com)
#                      Universidad Nacional de Cordoba / Argentina

# Library for managing updates in zebra, usually from zebraqueue

use strict;
use warnings;
use C4::Context;
use Getopt::Long;
use File::Temp qw/ tempdir /;
use File::Path;
use Time::HiRes qw(time);
use C4::Biblio;
use C4::AuthoritiesMarc;

use vars qw($VERSION @ISA @EXPORT);

BEGIN {
	# set the version for version checking
	$VERSION = 0.01;

	require Exporter;
	@ISA = qw(Exporter);
	@EXPORT = qw(
		&UpdateAuths
		&UpdateBiblios
		&UpdateAuthsAndBiblios
		&IndexZebraqueueRecords
	);
}


=head1 NAME

C4::Catalog::Zebra

Comment:
	This should be used when merging the rest of the rebuild_zebra.pl indexing logic
	my $nosanitize				= (C4::Context->preference('ZebraNoSanitize')) ? 1 : 0;


=head2 UpdateAuths

  ( $num_records_updated ) = &UpdateAuths ();

returns the number of updated+deleted authority records 

=cut

sub UpdateAuths
{
	# Update authorities
	return IndexZebraqueueRecords('authority');
}

=head2 UpdateBiblios

  ( $num_records_updated ) = &UpdateBiblios ();

returns the number of updated+deleted biblio records 

=cut

sub UpdateBiblios
{
	# Update authorities
	return IndexZebraqueueRecords('biblio');
}

=head2 UpdateAuthsAndBiblios

  ( $num_records_updated ) = &UpdateAuthsAndBiblios ();

returns the number of updated+deleted authority and biblio records 

=cut

sub UpdateAuthsAndBiblios
{
	my $ret;
	# Update authorities
	$ret = UpdateAuths();

	# Update biblios
	$ret += UpdateBiblios();

	return $ret;
}

=head2 IndexZebraqueueRecords

  ( $num_records_updated ) = &IndexZebraqueueRecords ($record_type);

returns the number of updated+deleted $record_type records 

Comment :
$record_type can be either 'biblio' or 'authority'

=cut

sub IndexZebraqueueRecords
{
	my ($record_type) = @_;
	my $as_xml			= (C4::Context->preference('ZebraUseXml')) ? 1 : 0;
	my $noxml			= ($as_xml) ? 0 : 1;
	my $record_format	= ($as_xml) ? 'marcxml' : 'iso2709' ;

	my ($num_records_updated,$num_records_deleted);

	$num_records_deleted = (IndexZebraqueueByAction('deleted',$record_type,$record_format,$as_xml,$noxml)||0);
	$num_records_updated = (IndexZebraqueueByAction('updated',$record_type,$record_format,$as_xml,$noxml)||0);

	return $num_records_deleted + $num_records_updated;
}

=head2 IndexZebraqueueByAction

  ( $num_records_updated ) = &IndexZebraqueueByAction ($action,$record_type,
														$record_format,$as_xml,$noxml);

returns the number of updated+deleted $record_type records 

Comment :
$record_type can be 'biblio' or 'authority'
$record_format can be 'marcxml' or 'iso2709'
$action can be 'updated' or 'deleted'
$as_xml and $noxml are maintained for legacy reasons, one is enough. They
indicate whether to use marcxml for indexing in zebra or iso2709. They should
all be deduced from C4::Context->preference('ZebraUseXml').

=cut

sub IndexZebraqueueByAction
{
	my ($action,$record_type,$record_format,$as_xml,$noxml) = @_;
	my ($num_records_exported,$ret,$zaction);

	if ($action eq 'updated' or $action eq 'deleted') {
		# get records by action
		my $entries = select_zebraqueue_records($record_type, $action);
		# Create tmp dir
		my $directory = File::Temp->newdir();

		# get records from zebraqueue, export to file for zebraidx
		if ($action eq 'updated') {
			$zaction = 'update';
			$num_records_exported = export_marc_records_from_list($record_type, 
												$entries, "$directory", $as_xml, $noxml);
		} else {	
			# $action eq 'deleted'
			$zaction = 'delete';
			$num_records_exported = generate_deleted_marc_records($record_type,
												$entries, "$directory", $as_xml);
		}

		if ($num_records_exported) {
			# log export
			my $time = localtime(time);
			print "$time $num_records_exported $record_type record(s) exported for $zaction\n";
			# TODO error handling / and better logging
			$ret = DoIndexing($record_type,$zaction,"$directory",$record_format);
			if ($ret) {
				print "$time $num_records_exported $record_type record(s) $action\n";
				mark_zebraqueue_batch_done($entries);
				print "$time $num_records_exported $record_type record(s) marked done in zebraqueue\n";
			}
			# /TODO
		}
	} else {
		# Wrong action
		$ret = -1;
	}

	return $ret;
}


sub select_zebraqueue_records {
	my ($record_type, $update_type) = @_;

	my $dbh = C4::Context->dbh;
	my $server = ($record_type eq 'biblio') ? 'biblioserver' : 'authorityserver';
	my $op = ($update_type eq 'deleted') ? 'recordDelete' : 'specialUpdate';

	my $sth = $dbh->prepare(<<'SQL');
		SELECT id, biblio_auth_number 
		FROM zebraqueue
		WHERE server = ?
		AND   operation = ?
		AND   done = 0
		ORDER BY id DESC;
SQL

	$sth->execute($server, $op);
	my $entries = $sth->fetchall_arrayref({});
}

sub mark_zebraqueue_batch_done {
	my ($entries) = @_;

	my $dbh = C4::Context->dbh;

	$dbh->{AutoCommit} = 0;
	my $sth = $dbh->prepare("UPDATE zebraqueue SET done = 1 WHERE id = ?");
	$dbh->commit();
	foreach my $id (map { $_->{id} } @$entries) {
		$sth->execute($id);
	}
	$dbh->{AutoCommit} = 1;
}

sub export_marc_records_from_list {
	my ($record_type, $entries, $directory, $as_xml, $noxml) = @_;
	my $verbose_logging = (C4::Context->preference('ZebraqueueVerboseLogging')) ? 1 : 0;

	my $num_exported = 0;
	open (OUT, ">:utf8 ", "$directory/exported_records") or die $!;
	my $i = 0;
	my %found = ();
	foreach my $record_number ( map { $_->{biblio_auth_number} }
								grep { !$found{ $_->{biblio_auth_number} }++ }
								@$entries ) {
		print "." if ( $verbose_logging );
		print "\r$i" unless ($i++ %100 or !$verbose_logging);
		my ($marc) = get_corrected_marc_record($record_type, $record_number, $noxml);
		if (defined $marc) {
			# FIXME - when more than one record is exported and $as_xml is true,
			# the output file is not valid XML - it's just multiple <record> elements
			# strung together with no single root element.  zebraidx doesn't seem
			# to care, though, at least if you're using the GRS-1 filter.  It does
			# care if you're using the DOM filter, which requires valid XML file(s).
			print OUT ($as_xml) ? $marc->as_xml_record() : $marc->as_usmarc();
			$num_exported++;
		}
	}
	print "\nRecords exported: $num_exported\n" if ( $verbose_logging );
	close OUT;
	return $num_exported;
}

sub generate_deleted_marc_records {
	my ($record_type, $entries, $directory, $as_xml) = @_;
	my $verbose_logging = (C4::Context->preference('ZebraqueueVerboseLogging')) ? 1 : 0;

	my $num_exported = 0;
	open (OUT, ">:utf8 ", "$directory/exported_records") or die $!;
	my $i = 0;
	foreach my $record_number (map { $_->{biblio_auth_number} } @$entries ) {
		print "\r$i" unless ($i++ %100 or !$verbose_logging);
		print "." if ( $verbose_logging );

		my $marc = MARC::Record->new();
		if ($record_type eq 'biblio') {
			fix_biblio_ids($marc, $record_number, $record_number);
		} else {
			fix_authority_id($marc, $record_number);
		}
		if (C4::Context->preference("marcflavour") eq "UNIMARC") {
			fix_unimarc_100($marc);
		}

		print OUT ($as_xml) ? $marc->as_xml_record() : $marc->as_usmarc();
		$num_exported++;
	}
	print "\nRecords exported: $num_exported\n" if ( $verbose_logging );
	close OUT;
	return $num_exported;
}

sub get_corrected_marc_record {
	my ($record_type, $record_number, $noxml) = @_;

	my $marc = get_raw_marc_record($record_type, $record_number, $noxml); 

	if (defined $marc) {
		fix_leader($marc);
		if ($record_type eq 'biblio') {
			my $succeeded = fix_biblio_ids($marc, $record_number);
			return unless $succeeded;
		} else {
			fix_authority_id($marc, $record_number);
		}
		if (C4::Context->preference("marcflavour") eq "UNIMARC") {
			fix_unimarc_100($marc);
		}
	}

	return $marc;
}

sub get_raw_marc_record {
	my ($record_type, $record_number, $noxml) = @_;
	my $dbh = C4::Context->dbh;

	my $marc; 
	if ($record_type eq 'biblio') {
		if ($noxml) {
			my $fetch_sth = $dbh->prepare_cached("SELECT marc FROM biblioitems WHERE biblionumber = ?");
			$fetch_sth->execute($record_number);
			if (my ($blob) = $fetch_sth->fetchrow_array) {
				$marc = MARC::Record->new_from_usmarc($blob);
				$fetch_sth->finish();
			} else {
				return; # failure to find a bib is not a problem -
						# a delete could have been done before
						# trying to process a record update
			}
		} else {
			eval { $marc = GetMarcBiblio($record_number,1); };
			if ($@) {
				# here we do warn since catching an exception
				# means that the bib was found but failed
				# to be parsed
				warn "error retrieving biblio $record_number";
				return;
			}
		}
	} else {
		eval { $marc = GetAuthority($record_number); };
		if ($@) {
			warn "error retrieving authority $record_number";
			return;
		}
	}
	return $marc;
}

sub fix_leader {
    # FIXME - this routine is suspect
    # It blanks the Leader/00-05 and Leader/12-16 to
    # force them to be recalculated correct when
    # the $marc->as_usmarc() or $marc->as_xml() is called.
    # But why is this necessary?  It would be a serious bug
    # in MARC::Record (definitely) and MARC::File::XML (arguably) 
    # if they are emitting incorrect leader values.
    my $marc = shift;

    my $leader = $marc->leader;
    substr($leader,  0, 5) = '     ';
    substr($leader, 10, 7) = '22     ';
    $marc->leader(substr($leader, 0, 24));
}

sub fix_biblio_ids {
	# FIXME - it is essential to ensure that the biblionumber is present,
	#         otherwise, Zebra will choke on the record.  However, this
	#         logic belongs in the relevant C4::Biblio APIs.
	my $marc = shift;
	my $biblionumber = shift;
	my $biblioitemnumber;
	my $dbh = C4::Context->dbh;

	if (@_) {
		$biblioitemnumber = shift;
	} else {    
		my $sth = $dbh->prepare(
			"SELECT biblioitemnumber FROM biblioitems WHERE biblionumber=?");
		$sth->execute($biblionumber);
		($biblioitemnumber) = $sth->fetchrow_array;
		$sth->finish;
		unless ($biblioitemnumber) {
			warn "failed to get biblioitemnumber for biblio $biblionumber";
			return 0;
		}
	}

	# FIXME - this is cheating on two levels
	# 1. C4::Biblio::_koha_marc_update_bib_ids is meant to be an internal function
	# 2. Making sure that the biblionumber and biblioitemnumber are correct and
	#    present in the MARC::Record object ought to be part of GetMarcBiblio.
	#
	# On the other hand, this better for now than what rebuild_zebra.pl used to
	# do, which was duplicate the code for inserting the biblionumber 
	# and biblioitemnumber
	C4::Biblio::_koha_marc_update_bib_ids($marc, '', $biblionumber, $biblioitemnumber);

	return 1;
}

sub fix_authority_id {
	# FIXME - as with fix_biblio_ids, the authid must be present
	#         for Zebra's sake.  However, this really belongs
	#         in C4::AuthoritiesMarc.
	my ($marc, $authid) = @_;
	unless ($marc->field('001') and $marc->field('001')->data() eq $authid){
		$marc->delete_field($marc->field('001'));
		$marc->insert_fields_ordered(MARC::Field->new('001',$authid));
	}
}

sub fix_unimarc_100 {
	# FIXME - again, if this is necessary, it belongs in C4::AuthoritiesMarc.
	my $marc = shift;

	my $string;
	if ( length($marc->subfield( 100, "a" )) == 35 ) {
		$string = $marc->subfield( 100, "a" );
		my $f100 = $marc->field(100);
		$marc->delete_field($f100);
	}
	else {
		$string = POSIX::strftime( "%Y%m%d", localtime );
		$string =~ s/\-//g;
		$string = sprintf( "%-*s", 35, $string );
	}
	substr( $string, 22, 6, "frey50" );
	unless ( length($marc->subfield( 100, "a" )) == 35 ) {
		$marc->delete_field($marc->field(100));
		$marc->insert_grouped_field(MARC::Field->new( 100, "", "", "a" => $string ));
	}
}

=head2 DoIndexing

  ( $error_code ) = &DoIndexing($record_type,$op,$record_dir,$record_format);

returns the corresponding zebraidx error code

Comment :
$record_type can be 'biblio' or 'authority'
$zaction can be 'delete' or 'update'
$record_dir is the directory where the exported records are
$record_format can be 'marcxml' or 'iso2709'

=cut

sub DoIndexing {
	my ($record_type, $zaction, $record_dir, $record_format) = @_;
	my $zebra_server	= ($record_type eq 'biblio') ? 'biblioserver' : 'authorityserver';
	my $zebra_db_name	= ($record_type eq 'biblio') ? 'biblios' : 'authorities';
	my $zebra_config	= C4::Context->zebraconfig($zebra_server)->{'config'};
	my $zebra_db_dir	= C4::Context->zebraconfig($zebra_server)->{'directory'};
	my $noshadow		= (C4::Context->preference('ZebraNoshadow')) ? '-n' : '';
	my $zebraidx_log_opt		= " -v none,fatal ";

	# TODO better error handling!!
	system("zebraidx -c $zebra_config $zebraidx_log_opt $noshadow -g $record_format -d $zebra_db_name $zaction $record_dir");
	system("zebraidx -c $zebra_config $zebraidx_log_opt -g $record_format -d $zebra_db_name commit") unless $noshadow;
	# /TODO
	
	return 1;
}


END { }

1;
__END__

=head1 AUTHOR

Koha Development Team <http://koha-community.org/>

Tomas Cohen Arazi tomascohen@gmail.com

=cut




#! /usr/bin/perl
use strict;
use warnings;
use C4::Context;
my $dbh=C4::Context->dbh;


$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraUseXml','1','Tell Zebra to use MARCXML instead of ISO2907 for indexing. Very important for libraries with records bigger than the allowed by ISO2907 (e.g. with lots of items in a single record).',NULL,'YesNo')");
$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraNoshadow','0','Tell Zebra to use shadow records when updating. Prevents locking on records while updating the database. Refer to zebra documentation for more info on the drawbacks.',NULL,'YesNo')");
$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraqueueVerboseLogging','0','Tell zebraqueue daemon to be more verbose on logging.',NULL,'YesNo')");
$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraBiblioUpdateRatio','6','By default, tell zebraqueue daemon to search for updates every ZebraBiblioUpdateRatio*ZebraAuthUpdateRatio seconds in the biblios database.',NULL,'Integer')");
$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraAuthUpdateRatio','10','By default, tell zebraqueue daemon to search for updates every ZebraAuthUpdateRatio seconds in the authorities  database.',NULL,'Integer')");

print "Upgrade done (Add sysprefs to control zebraqueue_daemon scripts: ZebraUseXml, ZebraNoshadow, ZebraqueueVerboseLogging, ZebraBiblioUpdateRatio, ZebraAuthUpdateRatio)\n";

Lines 328-331 INSERT INTO systempreferences (variable,value,explanation,options,type) VALUES(' Link Here

(-)a/installer/data/mysql/sysprefs.sql (+5 lines)
328	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('OpacKohaUrl','1',"Show 'Powered by Koha' text on OPAC footer.",NULL,NULL);	328	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('OpacKohaUrl','1',"Show 'Powered by Koha' text on OPAC footer.",NULL,NULL);
329	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES('EasyAnalyticalRecords','0','If on, display in the catalogue screens tools to easily setup analytical record relationships','','YesNo');	329	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES('EasyAnalyticalRecords','0','If on, display in the catalogue screens tools to easily setup analytical record relationships','','YesNo');
330	INSERT INTO systempreferences (variable,value,explanation,options,type) VALUES('OpacShowRecentComments',0,'If ON a link to recent comments will appear in the OPAC masthead',NULL,'YesNo');	330	INSERT INTO systempreferences (variable,value,explanation,options,type) VALUES('OpacShowRecentComments',0,'If ON a link to recent comments will appear in the OPAC masthead',NULL,'YesNo');
		331	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraUseXml','1','Tell Zebra to use MARCXML instead of ISO2907 for indexing. Very important for libraries with records bigger than the allowed by ISO2907 (e.g. with lots of items in a single record).',NULL,'YesNo');
		332	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraNoshadow','0','Tell Zebra to use shadow records when updating. Prevents locking on records while updating the database. Refer to zebra documentation for more info on the drawbacks.',NULL,'YesNo');
		333	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraqueueVerboseLogging','0','Tell zebraqueue daemon to be more verbose on logging.',NULL,'YesNo');
		334	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraBiblioUpdateRatio','6','By default, tell zebraqueue daemon to search for updates every ZebraBiblioUpdateRatio*ZebraAuthUpdateRatio seconds in the biblios database.',NULL,'Integer');
		335	INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraAuthUpdateRatio','10','By default, tell zebraqueue daemon to search for updates every ZebraAuthUpdateRatio seconds in the authorities database.',NULL,'Integer');
331		336




                  yes: Include
                  no: "Don't include"
            - subdivisions for searches generated by clicking on subject tracings.
        -
            - pref: ZebraUseXml
              type: boolean
              choices:
                  no: "Don't use"
                  yes: Use
            - MARCXML instead of ISO2907 for indexing in Zebra. Very important for libraries with records bigger than the allowed by ISO2907 (e.g. with lots of items in a single record).
        -
            - pref: ZebraNoshadow
              type: boolean
              choices:
                  yes: "Don't use"
                  no: Use
            - shadow records when updating Zebra indexes. Prevents locking on records while updating the database. Refer to zebra documentation for more info on the drawbacks.
        -
            - Set to
            - pref: ZebraAuthUpdateRatio
              class: integer
              default: 10
            - seconds the interval for searching authority record updates.
        -
            - Set to
            - pref: ZebraBiblioUpdateRatio
              class: integer
              default: 6
            - x ZebraAuthUpdateRatio seconds the interval for searching record updates.
        -
            - pref: ZebraqueueVerboseLogging
              type: boolean
              choices:
                no: "Don't use"
                yes: Use
            -  verbose logging in zebraqueue daemon.
    Search Form:
        -
            - Show checkboxes to search by




case "$1" in
    start)
      echo "Starting Zebraqueue Daemon"
      daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER -- perl -I $PERL5LIB $ZEBRAQUEUE
      ;;
    stop)
      echo "Stopping Zebraqueue Daemon"
      daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER --stop -- perl -I $PERL5LIB $ZEBRAQUEUE
      ;;
    restart)
      echo "Restarting the Zebraqueue Daemon"
      daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER --restart -- perl -I $PERL5LIB $ZEBRAQUEUE
      ;;
    *)
      echo "Usage: /etc/init.d/$NAME {start|stop|restart}"




#!/usr/bin/perl -w

# This file is part of Koha.
#
# Koha is free software; you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# Koha is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# Koha; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
# Suite 330, Boston, MA  02111-1307 USA

# Writen 02/03/2011 by Tomas Cohen Arazi (tomascohen@gmail.com)
#                      Universidad Nacional de Cordoba / Argentina

# Daemon to watch the zebraqueue table and update zebra indexes as needed

use strict;
#use warnings; FIXME - Bug 2505
BEGIN {
    # find Koha's Perl modules
    # test carefully before changing this
    use FindBin;
    eval { require "$FindBin::Bin/../kohalib.pl" };
}
use POE;
use Time::HiRes qw(time);
use Unix::Syslog qw(:macros);

use C4::Context;
use C4::Catalog::Zebra;
use C4::Search;
use C4::AuthoritiesMarc;
use XML::Simple;
use POSIX;
use utf8;


# wait periods governing connection attempts
my $min_connection_wait =    1; # start off at 1 second
my $max_connection_wait = 1024; # max about 17 minutes

# keep separate wait period for bib and authority Zebra databases
my %zoom_connection_waits = (); 

my $db_connection_wait = $min_connection_wait;

# ZOOM and Z39.50 errors that are potentially
# resolvable by connecting again and retrying
# the operation
my %retriable_zoom_errors = (
    10000 => 'ZOOM_ERROR_CONNECT',
    10001 => 'ZOOM_ERROR_MEMORY',
    10002 => 'ZOOM_ERROR_ENCODE',
    10003 => 'ZOOM_ERROR_DECODE',
    10004 => 'ZOOM_ERROR_CONNECTION_LOST',
    10005 => 'ZOOM_ERROR_INIT',
    10006 => 'ZOOM_ERROR_INTERNAL',
    10007 => 'ZOOM_ERROR_TIMEOUT',
);

# structure to store updates that have
# failed and are to be retrieved.  The
# structure is a hashref of hashrefs, 
# e.g.,
#
# $postoned_updates->{$server}->{$record_number} = 1;
#
# If an operation is attempted and fails because
# of a retriable error (see above), the daemon
# will try several times to recover as follows:
#
# 1. close and reopen the connection to the
#    Zebra server, unless the error was a timeout,
#    in which case
# 2. retry the operation
#
# If, after trying this five times, the operation still
# fails, the daemon will mark the record number as
# postponed, and try to process other entries in 
# zebraqueue.  When an update is postponed, the 
# error will be reported to syslog. 
#
# If more than 100 postponed updates are 
# accumulated, the daemon will assume that 
# something is seriously wrong, complain loudly,
# and abort.  If running under the daemon(1) command, 
# this means that the daemon will respawn.
#
my $num_postponed_updates = 0;
my $postponed_updates = {};

my $max_operation_attempts =   5;
my $max_postponed_updates  = 100;

# Zebra connection timeout
my $zconn_timeout            =  30;
my $zconn_timeout_multiplier = 1.5;
my $max_zconn_timeout        = 120;

my $ident = "Koha Zebraqueue ";

my $debug = 0;
Unix::Syslog::openlog $ident, LOG_PID, LOG_LOCAL0;

Unix::Syslog::syslog LOG_INFO, "Starting Zebraqueue log at " . scalar localtime(time) . "\n";

sub handler_start {

    # Starts session. Only ever called once only really used to set an alias
    # for the POE kernel
    my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];

    my $time = localtime(time);
    Unix::Syslog::syslog LOG_INFO, "$time POE Session ", $session->ID, " has started.\n";

    # check status
#    $kernel->yield('status_check');
    $kernel->yield('sleep');
}

sub handler_sleep {

    # can be used to slow down loop execution if needed
    my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
    use Time::HiRes qw (sleep);
    Time::HiRes::sleep(0.5);
    #sleep 1;
    $kernel->yield('status_check');
}

sub handler_check {
    # check if we need to do anything, at the moment just checks the zebraqueue, it could check other things too
    my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
    my $dbh = get_db_connection();
    my $sth = $dbh->prepare("SELECT count(*) AS opcount FROM zebraqueue WHERE done = 0");
    $sth->execute;
    my $data = $sth->fetchrow_hashref();
    if ($data->{'opcount'} > 0) {
        Unix::Syslog::syslog LOG_INFO, "$data->{'opcount'} operations waiting to be run\n";
        $sth->finish();
        $dbh->commit(); # needed so that we get current state of zebraqueue next time
                        # we enter handler_check
        $kernel->yield('do_ops');
    }
    else {
        $sth->finish();
        $dbh->commit(); # needed so that we get current state of zebraqueue next time
                        # we enter handler_check
        $kernel->yield('sleep');
    }
}

sub zebraop {
    # execute operations waiting in the zebraqueue
    my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
    my $dbh = get_db_connection();
    my $readsth = $dbh->prepare("SELECT id, biblio_auth_number, operation, server FROM zebraqueue WHERE done = 0 ORDER BY id DESC");
    $readsth->execute();
    Unix::Syslog::syslog LOG_INFO, "Executing zebra operations\n";

    my $completed_updates = {};
    ZEBRAQUEUE: while (my $data = $readsth->fetchrow_hashref()) {
        warn "Inside while loop" if $debug;

        my $id = $data->{'id'};
        my $op = $data->{'operation'};
        $op = 'recordDelete' if $op =~ /delete/i; # delete ops historically have been coded
                                                  # either delete_record or recordDelete
        my $record_number = $data->{'biblio_auth_number'};
        my $server = $data->{'server'};

        next ZEBRAQUEUE if exists $postponed_updates->{$server}->{$record_number};
        next ZEBRAQUEUE if exists $completed_updates->{$server}->{$record_number}->{$op};

        my $ok = 0;
        my $record;
        if ($op eq 'recordDelete') {
            $ok = process_delete($dbh, $server, $record_number);
        }
        else {
            $ok = process_update($dbh, $server, $record_number, $id);
        }
        if ($ok == 1) {
            mark_done($dbh, $record_number, $op, $server);
            $completed_updates->{$server}->{$record_number}->{$op} = 1;
            if ($op eq 'recordDelete') {
                $completed_updates->{$server}->{$record_number}->{'specialUpdate'} = 1;
            }
        }                            
    }
    $readsth->finish();
    $dbh->commit();
    $kernel->yield('sleep');
}

sub process_delete {
    my $dbh = shift;
    my $server = shift;
    my $record_number = shift;

    my $record;
    my $ok = 0;
    eval {
        warn "Searching for record to delete" if $debug;
        # 1st read the record in zebra, we have to get it from zebra as its no longer in the db
        my $Zconn =  get_zebra_connection($server);
        my $results = $Zconn->search_pqf( '@attr 1=Local-number '.$record_number);
        $results->option(elementSetName => 'marcxml');
        $record = $results->record(0)->raw();
    };
    if ($@) {
        # this doesn't exist, so no need to wail on zebra to delete it
        if ($@->code() eq 13) {
            $ok = 1;
        } else {
            # caught a ZOOM::Exception
            my $message = _format_zoom_error_message($@);
            postpone_update($server, $record_number, $message);
        }
    } else {
        # then, delete the record
        warn "Deleting record" if $debug;
        $ok = zebrado($record, 'recordDelete', $server, $record_number);
    }
    return $ok;
}

sub process_update {
    my $dbh = shift;
    my $server = shift;
    my $record_number = shift;
    my $id = shift;

    my $record;
    my $ok = 0;

    warn "Updating record" if $debug;
    # get the XML
    my $marcxml;
    if ($server eq "biblioserver") {
        my $marc = GetMarcBiblio($record_number);
        $marcxml = $marc->as_xml_record() if $marc;
    } 
    elsif ($server eq "authorityserver") {
        $marcxml = C4::AuthoritiesMarc::GetAuthorityXML($record_number);
    }
    # check it's XML, just in case
    eval {
        my $hashed = XMLin($marcxml);
    }; ### is it a proper xml? broken xml may crash ZEBRA- slow but safe
    ## it's Broken XML-- Should not reach here-- but if it does -lets protect ZEBRA
    if ($@) {
        Unix::Syslog::syslog LOG_ERR, "$server record $record_number is malformed: $@";
        mark_done_by_id($dbh, $id, $server);
        $ok = 0;
    } else {
        # ok, we have everything, do the operation in zebra !
        $ok = zebrado($marcxml, 'specialUpdate', $server, $record_number);
    }
    return $ok;
}

sub mark_done_by_id {
    my $dbh = shift;
    my $id = shift;
    my $server = shift;
    my $delsth = $dbh->prepare("UPDATE zebraqueue SET done = 1 WHERE id = ? AND server = ? AND done = 0");
    $delsth->execute($id, $server);
}

sub mark_done {
    my $dbh = shift;
    my $record_number = shift;
    my $op = shift;
    my $server = shift;

    my $delsth;
    if ($op eq 'recordDelete') {
        # if it's a deletion, we can delete every request on this biblio : in case the user
        # did a modif (or item deletion) just before biblio deletion, there are some specialUpdate
        # that are pending and can't succeed, as we don't have the XML anymore
        # so, delete everything for this biblionumber
        $delsth = $dbh->prepare_cached("UPDATE zebraqueue SET done = 1 
                                        WHERE biblio_auth_number = ? 
                                        AND server = ?
                                        AND done = 0");
        $delsth->execute($record_number, $server);
    } else {
        # if it's not a deletion, delete every pending specialUpdate for this biblionumber
        # in case the user add biblio, then X items, before this script runs
        # this avoid indexing X+1 times where just 1 is enough.
        $delsth = $dbh->prepare("UPDATE zebraqueue SET done = 1 
                                 WHERE biblio_auth_number = ? 
                                 AND operation = 'specialUpdate'
                                 AND server = ?
                                 AND done = 0");
        $delsth->execute($record_number, $server);
    }
}

my $authUpdateRatio;
my $biblioUpdateRatio;
my $tickCounter;

    unless ($record) {
        my $message = "error updating index for $server $record $record_number: no source record";
        postpone_update($server, $record_number, $message);
        return 0;
    }

sub handler_start
{
	my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
	my $time = localtime(time);
        warn "Attempt $attempts for $op for $server $record_number" if $debug;
        my $Zconn = get_zebra_connection($server);

	print "$time Zebraqueue daemon started\n";
        $Zpackage->option(action => $op);
        $Zpackage->option(record => $record);

	# Initialize counter
	$tickCounter  = 0;
            my $message = _format_zoom_error_message($@);
            my $error = $@->code();
            if (exists $retriable_zoom_errors{$error}) {
                warn "reattempting operation $op for $server $record_number" if $debug;
                warn "last Zebra error was $message" if $debug;
                $Zpackage->destroy();

                if ($error == 10007 and $zconn_timeout < $max_zconn_timeout) {
                    # bump up connection timeout
                    $zconn_timeout = POSIX::ceil($zconn_timeout * $zconn_timeout_multiplier);
                    $zconn_timeout = $max_zconn_timeout if $zconn_timeout > $max_zconn_timeout;
                    Unix::Syslog::syslog LOG_INFO, "increased Zebra connection timeout to $zconn_timeout\n";
                    warn "increased Zebra connection timeout to $zconn_timeout" if $debug;
                }
                next ATTEMPT;
            } else {
                postpone_update($server, $record_number, $message);
            }
        }
        # FIXME - would be more efficient to send a ES commit
        # after a batch of records, rather than commiting after
        # each one - Zebra handles updates relatively slowly.
        eval { $Zpackage->send('commit'); };
        if ($@) {
            # operation succeeded, but commit
            # did not - we have a problem
            my $message = _format_zoom_error_message($@);
            postpone_update($server, $record_number, $message);
        } else {
            $ok = 1;
            last ATTEMPT;
        }
    }

    unless ($ok) {
        my $message = "Made $attempts attempts to index $server record $record_number without success";
        postpone_update($server, $record_number, $message);
    }

    return $ok; 
}

	# Get timer settings
	$authUpdateRatio	= (C4::Context->preference("ZebraAuthUpdateRatio")||10);
	$biblioUpdateRatio	= (C4::Context->preference("ZebraBiblioUpdateRatio")||6);
    $message .= "\n" unless $message =~ /\n$/;
    Unix::Syslog::syslog LOG_ERR, $message;
    $postponed_updates->{$server}->{$record_number} = 1;

	# Log
	my $authPrefsString = (C4::Context->preference("ZebraAuthUpdateRatio") ? 'syspref' : 'default');
	print "$time Authorities update ratio (secs): $authUpdateRatio ($authPrefsString)\n";
	my $biblioUpdateSecs = $biblioUpdateRatio * $authUpdateRatio;
	my $biblioPrefsString = (C4::Context->preference("ZebraBiblioUpdateRatio") ? 'syspref' : 'default');
	print "$time Biblios update ratio (secs): $biblioUpdateSecs ($biblioPrefsString)\n";
    }
}

	$kernel->delay(tick => $authUpdateRatio);
    my $heap = $_[HEAP];
    my $time = localtime(time);
    Unix::Syslog::syslog LOG_INFO, "$time Session ", $_[SESSION]->ID, " has stopped.\n";
    delete $heap->{session};
}

# get a DB connection
sub get_db_connection {
    my $dbh;

    $db_connection_wait = $min_connection_wait unless defined $db_connection_wait;
    while (1) {
        eval {
            # note that C4::Context caches the
            # DB handle; C4::Context->dbh() will
            # check that handle first before returning
            # it.  If the connection is bad, it
            # then tries (once) to create a new one.
            $dbh = C4::Context->dbh();
        };

sub handler_stop
{
	my $heap = $_[HEAP];
	my $time = localtime(time);
	# Log
	print "$time Zebraqueue daemon stopped - POE Session ended\n";
	delete $heap->{session};
        }

        # connection failed
        my $error = "failed to connect to DB: $DBI::errstr";
        warn $error if $debug;
        Unix::Syslog::syslog LOG_ERR, $error;
        sleep $db_connection_wait;
        $db_connection_wait *= 2 unless $db_connection_wait >= $max_connection_wait;
    }
}

# get a Zebra connection
sub get_zebra_connection {
    my $server = shift;

    # start connection retry wait queue if necessary
    $zoom_connection_waits{$server} = $min_connection_wait unless exists  $zoom_connection_waits{$server};

sub handler_tick
{
	my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
	my $ret = 0;
	$tickCounter  = $tickCounter + 1;
        $Zconn->option('timeout' => $zconn_timeout);

        # it is important to note that if the existing connection
        # stored by C4::Context has an error (any type of error)
        # from the last transaction, C4::Context->Zconn closes
        # it and establishes a new one.  Therefore, the
        # following check will succeed if we have a new, good 
        # connection or we're using a previously established
        # connection that has experienced no errors.
        if ($Zconn->errcode() == 0) {
            $zoom_connection_waits{$server} = $min_connection_wait;
            return $Zconn;
        }

	# Calculate if we have to update biblios too
	# Check: biblioUpdateRatio ?= tickCounter
	if ($biblioUpdateRatio == $tickCounter) {
		# Update biblios and auths
		$ret = C4::Catalog::Zebra::UpdateAuthsAndBiblios();
		# Reset counter
		$tickCounter  = 0;
	} else {
		# Update only auths
		$ret = C4::Catalog::Zebra::UpdateAuths();
	}
# a human-reaable error message
sub _format_zoom_error_message {
    my $err = shift;

	$kernel->delay(tick => $authUpdateRatio);
    if (ref($err) eq 'ZOOM::Connection') {
        $message = $err->errmsg() . " (" . $err->diagset . " " . $err->errcode() . ") " . $err->addinfo();
    } elsif (ref($err) eq 'ZOOM::Exception') {
        $message = $err->message() . " (" . $err->diagset . " " .  $err->code() . ") " . $err->addinfo();
    }
    return $message; 
}

POE::Session->create(
	inline_states => {
		_start       => \&handler_start,
		tick         => \&handler_tick,
		_stop        => \&handler_stop,
	},
        _stop        => \&handler_stop,
    },
);

POE::Kernel->run();
exit 0;

Unix::Syslog::closelog;

exit;
- 

Return to bug 5166

Line 0 Link Here

(-)a/C4/Catalog/Zebra.pm (+474 lines)
		1	package C4::Catalog::Zebra;
		2	#
		3	# This file is part of Koha.
		4	#
		5	# Koha is free software; you can redistribute it and/or modify it under the
		6	# terms of the GNU General Public License as published by the Free Software
		7	# Foundation; either version 2 of the License, or (at your option) any later
		8	# version.
		9	#
		10	# Koha is distributed in the hope that it will be useful, but WITHOUT ANY
		11	# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
		12	# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
		13	#
		14	# You should have received a copy of the GNU General Public License along with
		15	# Koha; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
		16	# Suite 330, Boston, MA 02111-1307 USA
		17
		18	# Derived from rebuild_zebra.pl (2005-08-11) Paul Poulain and others
		19	# Rewriten 02/03/2011 by Tomas Cohen Arazi (tomascohen@gmail.com)
		20	# Universidad Nacional de Cordoba / Argentina
		21
		22	# Library for managing updates in zebra, usually from zebraqueue
		23
		24	use strict;
		25	use warnings;
		26	use C4::Context;
		27	use Getopt::Long;
		28	use File::Temp qw/ tempdir /;
		29	use File::Path;
		30	use Time::HiRes qw(time);
		31	use C4::Biblio;
		32	use C4::AuthoritiesMarc;
		33
		34	use vars qw($VERSION @ISA @EXPORT);
		35
		36	BEGIN {
		37	# set the version for version checking
		38	$VERSION = 0.01;
		39
		40	require Exporter;
		41	@ISA = qw(Exporter);
		42	@EXPORT = qw(
		43	&UpdateAuths
		44	&UpdateBiblios
		45	&UpdateAuthsAndBiblios
		46	&IndexZebraqueueRecords
		47	);
		48	}
		49
		50
		51	=head1 NAME
		52
		53	C4::Catalog::Zebra
		54
		55	Comment:
		56	This should be used when merging the rest of the rebuild_zebra.pl indexing logic
		57	my $nosanitize = (C4::Context->preference('ZebraNoSanitize')) ? 1 : 0;
		58
		59
		60	=head2 UpdateAuths
		61
		62	( $num_records_updated ) = &UpdateAuths ();
		63
		64	returns the number of updated+deleted authority records
65
66	=cut
67
68	sub UpdateAuths
69	{
70	# Update authorities
71	return IndexZebraqueueRecords('authority');
72	}
73
74	=head2 UpdateBiblios
75
76	( $num_records_updated ) = &UpdateBiblios ();
77
78	returns the number of updated+deleted biblio records
79
80	=cut
81
82	sub UpdateBiblios
83	{
84	# Update authorities
85	return IndexZebraqueueRecords('biblio');
86	}
87
88	=head2 UpdateAuthsAndBiblios
89
90	( $num_records_updated ) = &UpdateAuthsAndBiblios ();
91
92	returns the number of updated+deleted authority and biblio records
93
94	=cut
95
96	sub UpdateAuthsAndBiblios
97	{
98	my $ret;
99	# Update authorities
100	$ret = UpdateAuths();
101
102	# Update biblios
103	$ret += UpdateBiblios();
104
105	return $ret;
106	}
107
108	=head2 IndexZebraqueueRecords
109
110	( $num_records_updated ) = &IndexZebraqueueRecords ($record_type);
111
112	returns the number of updated+deleted $record_type records
113
114	Comment :
115	$record_type can be either 'biblio' or 'authority'
116
117	=cut
118
119	sub IndexZebraqueueRecords
120	{
121	my ($record_type) = @_;
122	my $as_xml = (C4::Context->preference('ZebraUseXml')) ? 1 : 0;
123	my $noxml = ($as_xml) ? 0 : 1;
124	my $record_format = ($as_xml) ? 'marcxml' : 'iso2709' ;
125
126	my ($num_records_updated,$num_records_deleted);
127
128	$num_records_deleted = (IndexZebraqueueByAction('deleted',$record_type,$record_format,$as_xml,$noxml)\|\|0);
129	$num_records_updated = (IndexZebraqueueByAction('updated',$record_type,$record_format,$as_xml,$noxml)\|\|0);
130
131	return $num_records_deleted + $num_records_updated;
132	}
133
134	=head2 IndexZebraqueueByAction
135
136	( $num_records_updated ) = &IndexZebraqueueByAction ($action,$record_type,
137	$record_format,$as_xml,$noxml);
138
139	returns the number of updated+deleted $record_type records
140
141	Comment :
142	$record_type can be 'biblio' or 'authority'
143	$record_format can be 'marcxml' or 'iso2709'
144	$action can be 'updated' or 'deleted'
145	$as_xml and $noxml are maintained for legacy reasons, one is enough. They
146	indicate whether to use marcxml for indexing in zebra or iso2709. They should
147	all be deduced from C4::Context->preference('ZebraUseXml').
148
149	=cut
150
151	sub IndexZebraqueueByAction
152	{
153	my ($action,$record_type,$record_format,$as_xml,$noxml) = @_;
154	my ($num_records_exported,$ret,$zaction);
155
156	if ($action eq 'updated' or $action eq 'deleted') {
157	# get records by action
158	my $entries = select_zebraqueue_records($record_type, $action);
159	# Create tmp dir
160	my $directory = File::Temp->newdir();
161
162	# get records from zebraqueue, export to file for zebraidx
163	if ($action eq 'updated') {
164	$zaction = 'update';
165	$num_records_exported = export_marc_records_from_list($record_type,
166	$entries, "$directory", $as_xml, $noxml);
167	} else {
168	# $action eq 'deleted'
169	$zaction = 'delete';
170	$num_records_exported = generate_deleted_marc_records($record_type,
171	$entries, "$directory", $as_xml);
172	}
173
174	if ($num_records_exported) {
175	# log export
176	my $time = localtime(time);
177	print "$time $num_records_exported $record_type record(s) exported for $zaction\n";
178	# TODO error handling / and better logging
179	$ret = DoIndexing($record_type,$zaction,"$directory",$record_format);
180	if ($ret) {
181	print "$time $num_records_exported $record_type record(s) $action\n";
182	mark_zebraqueue_batch_done($entries);
183	print "$time $num_records_exported $record_type record(s) marked done in zebraqueue\n";
184	}
185	# /TODO
186	}
187	} else {
188	# Wrong action
189	$ret = -1;
190	}
191
192	return $ret;
193	}
194
195
196	sub select_zebraqueue_records {
197	my ($record_type, $update_type) = @_;
198
199	my $dbh = C4::Context->dbh;
200	my $server = ($record_type eq 'biblio') ? 'biblioserver' : 'authorityserver';
201	my $op = ($update_type eq 'deleted') ? 'recordDelete' : 'specialUpdate';
202
203	my $sth = $dbh->prepare(<<'SQL');
204	SELECT id, biblio_auth_number
205	FROM zebraqueue
206	WHERE server = ?
207	AND operation = ?
208	AND done = 0
209	ORDER BY id DESC;
210	SQL
211
212	$sth->execute($server, $op);
213	my $entries = $sth->fetchall_arrayref({});
214	}
215
216	sub mark_zebraqueue_batch_done {
217	my ($entries) = @_;
218
219	my $dbh = C4::Context->dbh;
220
221	$dbh->{AutoCommit} = 0;
222	my $sth = $dbh->prepare("UPDATE zebraqueue SET done = 1 WHERE id = ?");
223	$dbh->commit();
224	foreach my $id (map { $_->{id} } @$entries) {
225	$sth->execute($id);
226	}
227	$dbh->{AutoCommit} = 1;
228	}
229
230	sub export_marc_records_from_list {
231	my ($record_type, $entries, $directory, $as_xml, $noxml) = @_;
232	my $verbose_logging = (C4::Context->preference('ZebraqueueVerboseLogging')) ? 1 : 0;
233
234	my $num_exported = 0;
235	open (OUT, ">:utf8 ", "$directory/exported_records") or die $!;
236	my $i = 0;
237	my %found = ();
238	foreach my $record_number ( map { $_->{biblio_auth_number} }
239	grep { !$found{ $_->{biblio_auth_number} }++ }
240	@$entries ) {
241	print "." if ( $verbose_logging );
242	print "\r$i" unless ($i++ %100 or !$verbose_logging);
243	my ($marc) = get_corrected_marc_record($record_type, $record_number, $noxml);
244	if (defined $marc) {
245	# FIXME - when more than one record is exported and $as_xml is true,
246	# the output file is not valid XML - it's just multiple <record> elements
247	# strung together with no single root element. zebraidx doesn't seem
248	# to care, though, at least if you're using the GRS-1 filter. It does
249	# care if you're using the DOM filter, which requires valid XML file(s).
250	print OUT ($as_xml) ? $marc->as_xml_record() : $marc->as_usmarc();
251	$num_exported++;
252	}
253	}
254	print "\nRecords exported: $num_exported\n" if ( $verbose_logging );
255	close OUT;
256	return $num_exported;
257	}
258
259	sub generate_deleted_marc_records {
260	my ($record_type, $entries, $directory, $as_xml) = @_;
261	my $verbose_logging = (C4::Context->preference('ZebraqueueVerboseLogging')) ? 1 : 0;
262
263	my $num_exported = 0;
264	open (OUT, ">:utf8 ", "$directory/exported_records") or die $!;
265	my $i = 0;
266	foreach my $record_number (map { $_->{biblio_auth_number} } @$entries ) {
267	print "\r$i" unless ($i++ %100 or !$verbose_logging);
268	print "." if ( $verbose_logging );
269
270	my $marc = MARC::Record->new();
271	if ($record_type eq 'biblio') {
272	fix_biblio_ids($marc, $record_number, $record_number);
273	} else {
274	fix_authority_id($marc, $record_number);
275	}
276	if (C4::Context->preference("marcflavour") eq "UNIMARC") {
277	fix_unimarc_100($marc);
278	}
279
280	print OUT ($as_xml) ? $marc->as_xml_record() : $marc->as_usmarc();
281	$num_exported++;
282	}
283	print "\nRecords exported: $num_exported\n" if ( $verbose_logging );
284	close OUT;
285	return $num_exported;
286	}
287
288	sub get_corrected_marc_record {
289	my ($record_type, $record_number, $noxml) = @_;
290
291	my $marc = get_raw_marc_record($record_type, $record_number, $noxml);
292
293	if (defined $marc) {
294	fix_leader($marc);
295	if ($record_type eq 'biblio') {
296	my $succeeded = fix_biblio_ids($marc, $record_number);
297	return unless $succeeded;
298	} else {
299	fix_authority_id($marc, $record_number);
300	}
301	if (C4::Context->preference("marcflavour") eq "UNIMARC") {
302	fix_unimarc_100($marc);
303	}
304	}
305
306	return $marc;
307	}
308
309	sub get_raw_marc_record {
310	my ($record_type, $record_number, $noxml) = @_;
311	my $dbh = C4::Context->dbh;
312
313	my $marc;
314	if ($record_type eq 'biblio') {
315	if ($noxml) {
316	my $fetch_sth = $dbh->prepare_cached("SELECT marc FROM biblioitems WHERE biblionumber = ?");
317	$fetch_sth->execute($record_number);
318	if (my ($blob) = $fetch_sth->fetchrow_array) {
319	$marc = MARC::Record->new_from_usmarc($blob);
320	$fetch_sth->finish();
321	} else {
322	return; # failure to find a bib is not a problem -
323	# a delete could have been done before
324	# trying to process a record update
325	}
326	} else {
327	eval { $marc = GetMarcBiblio($record_number,1); };
328	if ($@) {
329	# here we do warn since catching an exception
330	# means that the bib was found but failed
331	# to be parsed
332	warn "error retrieving biblio $record_number";
333	return;
334	}
335	}
336	} else {
337	eval { $marc = GetAuthority($record_number); };
338	if ($@) {
339	warn "error retrieving authority $record_number";
340	return;
341	}
342	}
343	return $marc;
344	}
345
346	sub fix_leader {
347	# FIXME - this routine is suspect
348	# It blanks the Leader/00-05 and Leader/12-16 to
349	# force them to be recalculated correct when
350	# the $marc->as_usmarc() or $marc->as_xml() is called.
351	# But why is this necessary? It would be a serious bug
352	# in MARC::Record (definitely) and MARC::File::XML (arguably)
353	# if they are emitting incorrect leader values.
354	my $marc = shift;
355
356	my $leader = $marc->leader;
357	substr($leader, 0, 5) = ' ';
358	substr($leader, 10, 7) = '22 ';
359	$marc->leader(substr($leader, 0, 24));
360	}
361
362	sub fix_biblio_ids {
363	# FIXME - it is essential to ensure that the biblionumber is present,
364	# otherwise, Zebra will choke on the record. However, this
365	# logic belongs in the relevant C4::Biblio APIs.
366	my $marc = shift;
367	my $biblionumber = shift;
368	my $biblioitemnumber;
369	my $dbh = C4::Context->dbh;
370
371	if (@_) {
372	$biblioitemnumber = shift;
373	} else {
374	my $sth = $dbh->prepare(
375	"SELECT biblioitemnumber FROM biblioitems WHERE biblionumber=?");
376	$sth->execute($biblionumber);
377	($biblioitemnumber) = $sth->fetchrow_array;
378	$sth->finish;
379	unless ($biblioitemnumber) {
380	warn "failed to get biblioitemnumber for biblio $biblionumber";
381	return 0;
382	}
383	}
384
385	# FIXME - this is cheating on two levels
386	# 1. C4::Biblio::_koha_marc_update_bib_ids is meant to be an internal function
387	# 2. Making sure that the biblionumber and biblioitemnumber are correct and
388	# present in the MARC::Record object ought to be part of GetMarcBiblio.
389	#
390	# On the other hand, this better for now than what rebuild_zebra.pl used to
391	# do, which was duplicate the code for inserting the biblionumber
392	# and biblioitemnumber
393	C4::Biblio::_koha_marc_update_bib_ids($marc, '', $biblionumber, $biblioitemnumber);
394
395	return 1;
396	}
397
398	sub fix_authority_id {
399	# FIXME - as with fix_biblio_ids, the authid must be present
400	# for Zebra's sake. However, this really belongs
401	# in C4::AuthoritiesMarc.
402	my ($marc, $authid) = @_;
403	unless ($marc->field('001') and $marc->field('001')->data() eq $authid){
404	$marc->delete_field($marc->field('001'));
405	$marc->insert_fields_ordered(MARC::Field->new('001',$authid));
406	}
407	}
408
409	sub fix_unimarc_100 {
410	# FIXME - again, if this is necessary, it belongs in C4::AuthoritiesMarc.
411	my $marc = shift;
412
413	my $string;
414	if ( length($marc->subfield( 100, "a" )) == 35 ) {
415	$string = $marc->subfield( 100, "a" );
416	my $f100 = $marc->field(100);
417	$marc->delete_field($f100);
418	}
419	else {
420	$string = POSIX::strftime( "%Y%m%d", localtime );
421	$string =~ s/\-//g;
422	$string = sprintf( "%-*s", 35, $string );
423	}
424	substr( $string, 22, 6, "frey50" );
425	unless ( length($marc->subfield( 100, "a" )) == 35 ) {
426	$marc->delete_field($marc->field(100));
427	$marc->insert_grouped_field(MARC::Field->new( 100, "", "", "a" => $string ));
428	}
429	}
430
431	=head2 DoIndexing
432
433	( $error_code ) = &DoIndexing($record_type,$op,$record_dir,$record_format);
434
435	returns the corresponding zebraidx error code
436
437	Comment :
438	$record_type can be 'biblio' or 'authority'
439	$zaction can be 'delete' or 'update'
440	$record_dir is the directory where the exported records are
441	$record_format can be 'marcxml' or 'iso2709'
442
443	=cut
444
445	sub DoIndexing {
446	my ($record_type, $zaction, $record_dir, $record_format) = @_;
447	my $zebra_server = ($record_type eq 'biblio') ? 'biblioserver' : 'authorityserver';
448	my $zebra_db_name = ($record_type eq 'biblio') ? 'biblios' : 'authorities';
449	my $zebra_config = C4::Context->zebraconfig($zebra_server)->{'config'};
450	my $zebra_db_dir = C4::Context->zebraconfig($zebra_server)->{'directory'};
451	my $noshadow = (C4::Context->preference('ZebraNoshadow')) ? '-n' : '';
452	my $zebraidx_log_opt = " -v none,fatal ";
453
454	# TODO better error handling!!
455	system("zebraidx -c $zebra_config $zebraidx_log_opt $noshadow -g $record_format -d $zebra_db_name $zaction $record_dir");
456	system("zebraidx -c $zebra_config $zebraidx_log_opt -g $record_format -d $zebra_db_name commit") unless $noshadow;
457	# /TODO
458
459	return 1;
460	}
461
462
463	END { }
464
465	1;
466	__END__
467
468	=head1 AUTHOR
469
470	Koha Development Team <http://koha-community.org/>
471
472	Tomas Cohen Arazi tomascohen@gmail.com
473
474	=cut

Line 0 Link Here

(-)a/installer/data/mysql/atomicupdate/bug_5166-zebraqueuedaemon_is_back.pl (+14 lines)
	1	#! /usr/bin/perl
	2	use strict;
	3	use warnings;
	4	use C4::Context;
	5	my $dbh=C4::Context->dbh;
	6
	7
	8	$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraUseXml','1','Tell Zebra to use MARCXML instead of ISO2907 for indexing. Very important for libraries with records bigger than the allowed by ISO2907 (e.g. with lots of items in a single record).',NULL,'YesNo')");
	9	$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraNoshadow','0','Tell Zebra to use shadow records when updating. Prevents locking on records while updating the database. Refer to zebra documentation for more info on the drawbacks.',NULL,'YesNo')");
	10	$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraqueueVerboseLogging','0','Tell zebraqueue daemon to be more verbose on logging.',NULL,'YesNo')");
	11	$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraBiblioUpdateRatio','6','By default, tell zebraqueue daemon to search for updates every ZebraBiblioUpdateRatio*ZebraAuthUpdateRatio seconds in the biblios database.',NULL,'Integer')");
	12	$dbh->do("INSERT INTO `systempreferences` (variable,value,explanation,options,type) VALUES ('ZebraAuthUpdateRatio','10','By default, tell zebraqueue daemon to search for updates every ZebraAuthUpdateRatio seconds in the authorities database.',NULL,'Integer')");
	13
	14	print "Upgrade done (Add sysprefs to control zebraqueue_daemon scripts: ZebraUseXml, ZebraNoshadow, ZebraqueueVerboseLogging, ZebraBiblioUpdateRatio, ZebraAuthUpdateRatio)\n";

Lines 69-74 Searching: Link Here

(-)a/koha-tmpl/intranet-tmpl/prog/en/modules/admin/preferences/searching.pref (+33 lines)
69	yes: Include	69	yes: Include
70	no: "Don't include"	70	no: "Don't include"
71	- subdivisions for searches generated by clicking on subject tracings.	71	- subdivisions for searches generated by clicking on subject tracings.
		72	-
		73	- pref: ZebraUseXml
		74	type: boolean
		75	choices:
		76	no: "Don't use"
		77	yes: Use
		78	- MARCXML instead of ISO2907 for indexing in Zebra. Very important for libraries with records bigger than the allowed by ISO2907 (e.g. with lots of items in a single record).
		79	-
		80	- pref: ZebraNoshadow
		81	type: boolean
		82	choices:
		83	yes: "Don't use"
		84	no: Use
		85	- shadow records when updating Zebra indexes. Prevents locking on records while updating the database. Refer to zebra documentation for more info on the drawbacks.
		86	-
		87	- Set to
		88	- pref: ZebraAuthUpdateRatio
		89	class: integer
		90	default: 10
		91	- seconds the interval for searching authority record updates.
		92	-
		93	- Set to
		94	- pref: ZebraBiblioUpdateRatio
		95	class: integer
		96	default: 6
		97	- x ZebraAuthUpdateRatio seconds the interval for searching record updates.
		98	-
		99	- pref: ZebraqueueVerboseLogging
		100	type: boolean
		101	choices:
		102	no: "Don't use"
		103	yes: Use
		104	- verbose logging in zebraqueue daemon.
72	Search Form:	105	Search Form:
73	-	106	-
74	- Show checkboxes to search by	107	- Show checkboxes to search by

Lines 23-37 fi Link Here

(-)a/misc/bin/koha-zebraqueue-ctl.sh (-3 / +3 lines)
23	case "$1" in	23	case "$1" in
24	start)	24	start)
25	echo "Starting Zebraqueue Daemon"	25	echo "Starting Zebraqueue Daemon"
26	daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER -- perl -I $PERL5LIB $ZEBRAQUEUE -f $KOHA_CONF	26	daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER -- perl -I $PERL5LIB $ZEBRAQUEUE
27	;;	27	;;
28	stop)	28	stop)
29	echo "Stopping Zebraqueue Daemon"	29	echo "Stopping Zebraqueue Daemon"
30	daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER --stop -- perl -I $PERL5LIB $ZEBRAQUEUE -f $KOHA_CONF	30	daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER --stop -- perl -I $PERL5LIB $ZEBRAQUEUE
31	;;	31	;;
32	restart)	32	restart)
33	echo "Restarting the Zebraqueue Daemon"	33	echo "Restarting the Zebraqueue Daemon"
34	daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER --restart -- perl -I $PERL5LIB $ZEBRAQUEUE -f $KOHA_CONF	34	daemon --name=$NAME --errlog=$ERRLOG --stdout=$STDOUT --output=$OUTPUT --verbose=1 --respawn --delay=30 $OTHERUSER --restart -- perl -I $PERL5LIB $ZEBRAQUEUE
35	;;	35	;;
36	*)	36	*)
37	echo "Usage: /etc/init.d/$NAME {start\|stop\|restart}"	37	echo "Usage: /etc/init.d/$NAME {start\|stop\|restart}"

Lines 1-475 Link Here

(-)a/misc/bin/zebraqueue_daemon.pl (-444 / +75 lines)
1	#!/usr/bin/perl	1	#!/usr/bin/perl -w
2		2
3	# daemon to watch the zebraqueue and update zebra as needed	3	# This file is part of Koha.
		4	#
		5	# Koha is free software; you can redistribute it and/or modify it under the
		6	# terms of the GNU General Public License as published by the Free Software
		7	# Foundation; either version 2 of the License, or (at your option) any later
		8	# version.
		9	#
		10	# Koha is distributed in the hope that it will be useful, but WITHOUT ANY
		11	# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
		12	# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
		13	#
		14	# You should have received a copy of the GNU General Public License along with
		15	# Koha; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
		16	# Suite 330, Boston, MA 02111-1307 USA
		17
		18	# Writen 02/03/2011 by Tomas Cohen Arazi (tomascohen@gmail.com)
		19	# Universidad Nacional de Cordoba / Argentina
		20
		21	# Daemon to watch the zebraqueue table and update zebra indexes as needed
4		22
5	use strict;	23	use strict;
6	#use warnings; FIXME - Bug 2505
7	BEGIN {	24	BEGIN {
8	# find Koha's Perl modules	25	# find Koha's Perl modules
9	# test carefully before changing this	26	# test carefully before changing this
10	use FindBin;	27	use FindBin;
11	eval { require "$FindBin::Bin/kohalib.pl" };	28	eval { require "$FindBin::Bin/../kohalib.pl" };
12	}	29	}
13		30	use POE;
14	use POE qw(Wheel::SocketFactory Wheel::ReadWrite Filter::Stream Driver::SysRW);	31	use Time::HiRes qw(time);
15	use Unix::Syslog qw(:macros);
16
17	use C4::Context;	32	use C4::Context;
18	use C4::Biblio;	33	use C4::Catalog::Zebra;
19	use C4::Search;
20	use C4::AuthoritiesMarc;
21	use XML::Simple;
22	use POSIX;
23	use utf8;
24
25
26	# wait periods governing connection attempts
27	my $min_connection_wait = 1; # start off at 1 second
28	my $max_connection_wait = 1024; # max about 17 minutes
29
30	# keep separate wait period for bib and authority Zebra databases
31	my %zoom_connection_waits = ();
32
33	my $db_connection_wait = $min_connection_wait;
34
35	# ZOOM and Z39.50 errors that are potentially
36	# resolvable by connecting again and retrying
37	# the operation
38	my %retriable_zoom_errors = (
39	10000 => 'ZOOM_ERROR_CONNECT',
40	10001 => 'ZOOM_ERROR_MEMORY',
41	10002 => 'ZOOM_ERROR_ENCODE',
42	10003 => 'ZOOM_ERROR_DECODE',
43	10004 => 'ZOOM_ERROR_CONNECTION_LOST',
44	10005 => 'ZOOM_ERROR_INIT',
45	10006 => 'ZOOM_ERROR_INTERNAL',
46	10007 => 'ZOOM_ERROR_TIMEOUT',
47	);
48
49	# structure to store updates that have
50	# failed and are to be retrieved. The
51	# structure is a hashref of hashrefs,
52	# e.g.,
53	#
54	# $postoned_updates->{$server}->{$record_number} = 1;
55	#
56	# If an operation is attempted and fails because
57	# of a retriable error (see above), the daemon
58	# will try several times to recover as follows:
59	#
60	# 1. close and reopen the connection to the
61	# Zebra server, unless the error was a timeout,
62	# in which case
63	# 2. retry the operation
64	#
65	# If, after trying this five times, the operation still
66	# fails, the daemon will mark the record number as
67	# postponed, and try to process other entries in
68	# zebraqueue. When an update is postponed, the
69	# error will be reported to syslog.
70	#
71	# If more than 100 postponed updates are
72	# accumulated, the daemon will assume that
73	# something is seriously wrong, complain loudly,
74	# and abort. If running under the daemon(1) command,
75	# this means that the daemon will respawn.
76	#
77	my $num_postponed_updates = 0;
78	my $postponed_updates = {};
79
80	my $max_operation_attempts = 5;
81	my $max_postponed_updates = 100;
82
83	# Zebra connection timeout
84	my $zconn_timeout = 30;
85	my $zconn_timeout_multiplier = 1.5;
86	my $max_zconn_timeout = 120;
87
88	my $ident = "Koha Zebraqueue ";
89
90	my $debug = 0;
91	Unix::Syslog::openlog $ident, LOG_PID, LOG_LOCAL0;
92
93	Unix::Syslog::syslog LOG_INFO, "Starting Zebraqueue log at " . scalar localtime(time) . "\n";
94
95	sub handler_start {
96
97	# Starts session. Only ever called once only really used to set an alias
98	# for the POE kernel
99	my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
100
101	my $time = localtime(time);
102	Unix::Syslog::syslog LOG_INFO, "$time POE Session ", $session->ID, " has started.\n";
103
104	# check status
105	# $kernel->yield('status_check');
106	$kernel->yield('sleep');
107	}
108
109	sub handler_sleep {
110
111	# can be used to slow down loop execution if needed
112	my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
113	use Time::HiRes qw (sleep);
114	Time::HiRes::sleep(0.5);
115	#sleep 1;
116	$kernel->yield('status_check');
117	}
118
119	sub handler_check {
120	# check if we need to do anything, at the moment just checks the zebraqueue, it could check other things too
121	my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
122	my $dbh = get_db_connection();
123	my $sth = $dbh->prepare("SELECT count(*) AS opcount FROM zebraqueue WHERE done = 0");
124	$sth->execute;
125	my $data = $sth->fetchrow_hashref();
126	if ($data->{'opcount'} > 0) {
127	Unix::Syslog::syslog LOG_INFO, "$data->{'opcount'} operations waiting to be run\n";
128	$sth->finish();
129	$dbh->commit(); # needed so that we get current state of zebraqueue next time
130	# we enter handler_check
131	$kernel->yield('do_ops');
132	}
133	else {
134	$sth->finish();
135	$dbh->commit(); # needed so that we get current state of zebraqueue next time
136	# we enter handler_check
137	$kernel->yield('sleep');
138	}
139	}
140
141	sub zebraop {
142	# execute operations waiting in the zebraqueue
143	my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
144	my $dbh = get_db_connection();
145	my $readsth = $dbh->prepare("SELECT id, biblio_auth_number, operation, server FROM zebraqueue WHERE done = 0 ORDER BY id DESC");
146	$readsth->execute();
147	Unix::Syslog::syslog LOG_INFO, "Executing zebra operations\n";
148
149	my $completed_updates = {};
150	ZEBRAQUEUE: while (my $data = $readsth->fetchrow_hashref()) {
151	warn "Inside while loop" if $debug;
152
153	my $id = $data->{'id'};
154	my $op = $data->{'operation'};
155	$op = 'recordDelete' if $op =~ /delete/i; # delete ops historically have been coded
156	# either delete_record or recordDelete
157	my $record_number = $data->{'biblio_auth_number'};
158	my $server = $data->{'server'};
159
160	next ZEBRAQUEUE if exists $postponed_updates->{$server}->{$record_number};
161	next ZEBRAQUEUE if exists $completed_updates->{$server}->{$record_number}->{$op};
162
163	my $ok = 0;
164	my $record;
165	if ($op eq 'recordDelete') {
166	$ok = process_delete($dbh, $server, $record_number);
167	}
168	else {
169	$ok = process_update($dbh, $server, $record_number, $id);
170	}
171	if ($ok == 1) {
172	mark_done($dbh, $record_number, $op, $server);
173	$completed_updates->{$server}->{$record_number}->{$op} = 1;
174	if ($op eq 'recordDelete') {
175	$completed_updates->{$server}->{$record_number}->{'specialUpdate'} = 1;
176	}
177	}
178	}
179	$readsth->finish();
180	$dbh->commit();
181	$kernel->yield('sleep');
182	}
183
184	sub process_delete {
185	my $dbh = shift;
186	my $server = shift;
187	my $record_number = shift;
188
189	my $record;
190	my $ok = 0;
191	eval {
192	warn "Searching for record to delete" if $debug;
193	# 1st read the record in zebra, we have to get it from zebra as its no longer in the db
194	my $Zconn = get_zebra_connection($server);
195	my $results = $Zconn->search_pqf( '@attr 1=Local-number '.$record_number);
196	$results->option(elementSetName => 'marcxml');
197	$record = $results->record(0)->raw();
198	};
199	if ($@) {
200	# this doesn't exist, so no need to wail on zebra to delete it
201	if ($@->code() eq 13) {
202	$ok = 1;
203	} else {
204	# caught a ZOOM::Exception
205	my $message = _format_zoom_error_message($@);
206	postpone_update($server, $record_number, $message);
207	}
208	} else {
209	# then, delete the record
210	warn "Deleting record" if $debug;
211	$ok = zebrado($record, 'recordDelete', $server, $record_number);
212	}
213	return $ok;
214	}
215
216	sub process_update {
217	my $dbh = shift;
218	my $server = shift;
219	my $record_number = shift;
220	my $id = shift;
221
222	my $record;
223	my $ok = 0;
224
225	warn "Updating record" if $debug;
226	# get the XML
227	my $marcxml;
228	if ($server eq "biblioserver") {
229	my $marc = GetMarcBiblio($record_number);
230	$marcxml = $marc->as_xml_record() if $marc;
231	}
232	elsif ($server eq "authorityserver") {
233	$marcxml = C4::AuthoritiesMarc::GetAuthorityXML($record_number);
234	}
235	# check it's XML, just in case
236	eval {
237	my $hashed = XMLin($marcxml);
238	}; ### is it a proper xml? broken xml may crash ZEBRA- slow but safe
239	## it's Broken XML-- Should not reach here-- but if it does -lets protect ZEBRA
240	if ($@) {
241	Unix::Syslog::syslog LOG_ERR, "$server record $record_number is malformed: $@";
242	mark_done_by_id($dbh, $id, $server);
243	$ok = 0;
244	} else {
245	# ok, we have everything, do the operation in zebra !
246	$ok = zebrado($marcxml, 'specialUpdate', $server, $record_number);
247	}
248	return $ok;
249	}
250		34
251	sub mark_done_by_id {
252	my $dbh = shift;
253	my $id = shift;
254	my $server = shift;
255	my $delsth = $dbh->prepare("UPDATE zebraqueue SET done = 1 WHERE id = ? AND server = ? AND done = 0");
256	$delsth->execute($id, $server);
257	}
258
259	sub mark_done {
260	my $dbh = shift;
261	my $record_number = shift;
262	my $op = shift;
263	my $server = shift;
264
265	my $delsth;
266	if ($op eq 'recordDelete') {
267	# if it's a deletion, we can delete every request on this biblio : in case the user
268	# did a modif (or item deletion) just before biblio deletion, there are some specialUpdate
269	# that are pending and can't succeed, as we don't have the XML anymore
270	# so, delete everything for this biblionumber
271	$delsth = $dbh->prepare_cached("UPDATE zebraqueue SET done = 1
272	WHERE biblio_auth_number = ?
273	AND server = ?
274	AND done = 0");
275	$delsth->execute($record_number, $server);
276	} else {
277	# if it's not a deletion, delete every pending specialUpdate for this biblionumber
278	# in case the user add biblio, then X items, before this script runs
279	# this avoid indexing X+1 times where just 1 is enough.
280	$delsth = $dbh->prepare("UPDATE zebraqueue SET done = 1
281	WHERE biblio_auth_number = ?
282	AND operation = 'specialUpdate'
283	AND server = ?
284	AND done = 0");
285	$delsth->execute($record_number, $server);
286	}
287	}
288		35
289	sub zebrado {	36	my $authUpdateRatio;
290	###Accepts a $server variable thus we can use it to update biblios, authorities or other zebra dbs	37	my $biblioUpdateRatio;
291	my ($record, $op, $server, $record_number) = @_;	38	my $tickCounter;
292		39
293	unless ($record) {
294	my $message = "error updating index for $server $record $record_number: no source record";
295	postpone_update($server, $record_number, $message);
296	return 0;
297	}
298		40
299	my $attempts = 0;	41	sub handler_start
300	my $ok = 0;	42	{
301	ATTEMPT: while ($attempts < $max_operation_attempts) {	43	my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
302	$attempts++;	44	my $time = localtime(time);
303	warn "Attempt $attempts for $op for $server $record_number" if $debug;
304	my $Zconn = get_zebra_connection($server);
305		45
306	my $Zpackage = $Zconn->package();	46	print "$time Zebraqueue daemon started\n";
307	$Zpackage->option(action => $op);
308	$Zpackage->option(record => $record);
309		47
310	eval { $Zpackage->send("update") };	48	# Initialize counter
311	if ($@ && $@->isa("ZOOM::Exception")) {	49	$tickCounter = 0;
312	my $message = _format_zoom_error_message($@);
313	my $error = $@->code();
314	if (exists $retriable_zoom_errors{$error}) {
315	warn "reattempting operation $op for $server $record_number" if $debug;
316	warn "last Zebra error was $message" if $debug;
317	$Zpackage->destroy();
318
319	if ($error == 10007 and $zconn_timeout < $max_zconn_timeout) {
320	# bump up connection timeout
321	$zconn_timeout = POSIX::ceil($zconn_timeout * $zconn_timeout_multiplier);
322	$zconn_timeout = $max_zconn_timeout if $zconn_timeout > $max_zconn_timeout;
323	Unix::Syslog::syslog LOG_INFO, "increased Zebra connection timeout to $zconn_timeout\n";
324	warn "increased Zebra connection timeout to $zconn_timeout" if $debug;
325	}
326	next ATTEMPT;
327	} else {
328	postpone_update($server, $record_number, $message);
329	}
330	}
331	# FIXME - would be more efficient to send a ES commit
332	# after a batch of records, rather than commiting after
333	# each one - Zebra handles updates relatively slowly.
334	eval { $Zpackage->send('commit'); };
335	if ($@) {
336	# operation succeeded, but commit
337	# did not - we have a problem
338	my $message = _format_zoom_error_message($@);
339	postpone_update($server, $record_number, $message);
340	} else {
341	$ok = 1;
342	last ATTEMPT;
343	}
344	}
345
346	unless ($ok) {
347	my $message = "Made $attempts attempts to index $server record $record_number without success";
348	postpone_update($server, $record_number, $message);
349	}
350
351	return $ok;
352	}
353		50
354	sub postpone_update {	51	# Get timer settings
355	my ($server, $record_number, $message) = @_;	52	$authUpdateRatio = (C4::Context->preference("ZebraAuthUpdateRatio")\|\|10);
356	warn $message if $debug;	53	$biblioUpdateRatio = (C4::Context->preference("ZebraBiblioUpdateRatio")\|\|6);
357	$message .= "\n" unless $message =~ /\n$/;
358	Unix::Syslog::syslog LOG_ERR, $message;
359	$postponed_updates->{$server}->{$record_number} = 1;
360		54
361	$num_postponed_updates++;	55	# Log
362	if ($num_postponed_updates > $max_postponed_updates) {	56	my $authPrefsString = (C4::Context->preference("ZebraAuthUpdateRatio") ? 'syspref' : 'default');
363	warn "exiting, over $max_postponed_updates postponed indexing updates";	57	print "$time Authorities update ratio (secs): $authUpdateRatio ($authPrefsString)\n";
364	Unix::Syslog::syslog LOG_ERR, "exiting, over $max_postponed_updates postponed indexing updates";	58	my $biblioUpdateSecs = $biblioUpdateRatio * $authUpdateRatio;
365	Unix::Syslog::closelog;	59	my $biblioPrefsString = (C4::Context->preference("ZebraBiblioUpdateRatio") ? 'syspref' : 'default');
366	exit;	60	print "$time Biblios update ratio (secs): $biblioUpdateSecs ($biblioPrefsString)\n";
367	}
368	}
369		61
370	sub handler_stop {	62	$kernel->delay(tick => $authUpdateRatio);
371	my $heap = $_[HEAP];
372	my $time = localtime(time);
373	Unix::Syslog::syslog LOG_INFO, "$time Session ", $_[SESSION]->ID, " has stopped.\n";
374	delete $heap->{session};
375	}	63	}
376		64
377	# get a DB connection
378	sub get_db_connection {
379	my $dbh;
380
381	$db_connection_wait = $min_connection_wait unless defined $db_connection_wait;
382	while (1) {
383	eval {
384	# note that C4::Context caches the
385	# DB handle; C4::Context->dbh() will
386	# check that handle first before returning
387	# it. If the connection is bad, it
388	# then tries (once) to create a new one.
389	$dbh = C4::Context->dbh();
390	};
391		65
392	unless ($@) {	66	sub handler_stop
393	# C4::Context->dbh dies if it cannot	67	{
394	# establish a connection	68	my $heap = $_[HEAP];
395	$db_connection_wait = $min_connection_wait;	69	my $time = localtime(time);
396	$dbh->{AutoCommit} = 0; # do this to reduce number of	70	# Log
397	# commits to zebraqueue	71	print "$time Zebraqueue daemon stopped - POE Session ended\n";
398	return $dbh;	72	delete $heap->{session};
399	}
400
401	# connection failed
402	my $error = "failed to connect to DB: $DBI::errstr";
403	warn $error if $debug;
404	Unix::Syslog::syslog LOG_ERR, $error;
405	sleep $db_connection_wait;
406	$db_connection_wait *= 2 unless $db_connection_wait >= $max_connection_wait;
407	}
408	}	73	}
409		74
410	# get a Zebra connection
411	sub get_zebra_connection {
412	my $server = shift;
413
414	# start connection retry wait queue if necessary
415	$zoom_connection_waits{$server} = $min_connection_wait unless exists $zoom_connection_waits{$server};
416		75
417	# try to connect to Zebra forever until we succeed	76	sub handler_tick
418	while (1) {	77	{
419	# what follows assumes that C4::Context->Zconn	78	my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
420	# makes only one attempt to create a new connection;	79	my $ret = 0;
421	my $Zconn = C4::Context->Zconn($server, 0, 1, '', 'xml');	80	$tickCounter = $tickCounter + 1;
422	$Zconn->option('timeout' => $zconn_timeout);
423		81
424	# it is important to note that if the existing connection
425	# stored by C4::Context has an error (any type of error)
426	# from the last transaction, C4::Context->Zconn closes
427	# it and establishes a new one. Therefore, the
428	# following check will succeed if we have a new, good
429	# connection or we're using a previously established
430	# connection that has experienced no errors.
431	if ($Zconn->errcode() == 0) {
432	$zoom_connection_waits{$server} = $min_connection_wait;
433	return $Zconn;
434	}
435		82
436	# connection failed	83	# Calculate if we have to update biblios too
437	my $error = _format_zoom_error_message($Zconn);	84	# Check: biblioUpdateRatio ?= tickCounter
438	warn $error if $debug;	85	if ($biblioUpdateRatio == $tickCounter) {
439	Unix::Syslog::syslog LOG_ERR, $error;	86	# Update biblios and auths
440	sleep $zoom_connection_waits{$server};	87	$ret = C4::Catalog::Zebra::UpdateAuthsAndBiblios();
441	$zoom_connection_waits{$server} *= 2 unless $zoom_connection_waits{$server} >= $max_connection_wait;	88	# Reset counter
442	}	89	$tickCounter = 0;
443	}	90	} else {
444		91	# Update only auths
445	# given a ZOOM::Exception or	92	$ret = C4::Catalog::Zebra::UpdateAuths();
446	# ZOOM::Connection object, generate	93	}
447	# a human-reaable error message
448	sub _format_zoom_error_message {
449	my $err = shift;
450		94
451	my $message = "";	95	$kernel->delay(tick => $authUpdateRatio);
452	if (ref($err) eq 'ZOOM::Connection') {
453	$message = $err->errmsg() . " (" . $err->diagset . " " . $err->errcode() . ") " . $err->addinfo();
454	} elsif (ref($err) eq 'ZOOM::Exception') {
455	$message = $err->message() . " (" . $err->diagset . " " . $err->code() . ") " . $err->addinfo();
456	}
457	return $message;
458	}	96	}
459		97
460	POE::Session->create(	98	POE::Session->create(
461	inline_states => {	99	inline_states => {
462	_start => \&handler_start,	100	_start => \&handler_start,
463	sleep => \&handler_sleep,	101	tick => \&handler_tick,
464	status_check => \&handler_check,	102	_stop => \&handler_stop,
465	do_ops => \&zebraop,	103	},
466	_stop => \&handler_stop,
467	},
468	);	104	);
469		105
470	# start the kernel	106	POE::Kernel->run();
471	$poe_kernel->run();	107	exit 0;
472
473	Unix::Syslog::closelog;
474
475	exit;
476	-