misc/bin/zebraqueue_daemon.pl

   1 #!/usr/bin/perl
   2
   3 # daemon to watch the zebraqueue and update zebra as needed
   4
   5 use strict;
   6 BEGIN {
   7     # find Koha's Perl modules
   8     # test carefully before changing this
   9     use FindBin;
  10     eval { require "$FindBin::Bin/kohalib.pl" };
  11 }
  12 use POE qw(Wheel::SocketFactory Wheel::ReadWrite Filter::Stream Driver::SysRW);
  13 use Unix::Syslog qw(:macros);
  14
  15 use C4::Context;
  16 use C4::Biblio;
  17 use C4::Search;
  18 use C4::AuthoritiesMarc;
  19 use XML::Simple;
  20 use utf8;
  21
  22 my $dbh = C4::Context->dbh;
  23 my $ident = "Koha Zebraqueue ";
  24
  25 my $debug = 1;
  26 Unix::Syslog::openlog $ident, LOG_PID, LOG_LOCAL0;
  27
  28 Unix::Syslog::syslog LOG_INFO, "Starting Zebraqueue log at " . scalar localtime(time) . "\n";
  29
  30 sub handler_start {
  31
  32     # Starts session. Only ever called once only really used to set an alias
  33     # for the POE kernel
  34     my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
  35
  36     my $time = localtime(time);
  37     Unix::Syslog::syslog LOG_INFO, "$time POE Session ", $session->ID, " has started.\n";
  38
  39     # check status
  40 #    $kernel->yield('status_check');
  41     $kernel->yield('sleep');
  42 }
  43
  44 sub handler_sleep {
  45
  46     # can be used to slow down loop execution if needed
  47     my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
  48     use Time::HiRes qw (sleep);
  49     Time::HiRes::sleep(0.01);
  50     #sleep 1;
  51     $kernel->yield('status_check');
  52 }
  53
  54 sub handler_check {
  55     # check if we need to do anything, at the moment just checks the zebraqueue, it could check other things too
  56     my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
  57     my $dbh = C4::Context->dbh;
  58     my $sth = $dbh->prepare("SELECT count(*) AS opcount FROM zebraqueue WHERE done = 0");
  59     $sth->execute;
  60     my $data = $sth->fetchrow_hashref();
  61     if ($data->{'opcount'} > 0) {
  62         Unix::Syslog::syslog LOG_INFO, "$data->{'opcount'} operations waiting to be run\n";
  63         $sth->finish();
  64         $kernel->yield('do_ops');
  65     }
  66     else {
  67         $sth->finish();
  68         $kernel->yield('sleep');
  69     }
  70 }
  71
  72 sub zebraop {
  73     # execute operations waiting in the zebraqueue
  74     my ( $kernel, $heap, $session ) = @_[ KERNEL, HEAP, SESSION ];
  75     my $dbh = C4::Context->dbh;
  76     my $readsth = $dbh->prepare("SELECT id, biblio_auth_number, operation, server FROM zebraqueue WHERE done = 0");
  77     $readsth->execute();
  78     Unix::Syslog::syslog LOG_INFO, "Executing zebra operations\n";
  79     while (my $data = $readsth->fetchrow_hashref()) {
  80         warn "Inside while loop" if $debug;
  81         eval {
  82             my $ok = 0;
  83             my $record;
  84             if ($data->{'operation'} =~ /delete/i ) {
  85                 eval {
  86
  87                     warn "Searching for record to delete" if $debug;
  88                     # 1st read the record in zebra, we have to get it from zebra as its no longer in the db
  89                     my $Zconn = C4::Context->Zconn($data->{'server'}, 0, 1, '', 'xml');
  90                     my $results = $Zconn->search_pqf( '@attr 1=Local-number '.$data->{'biblio_auth_number'});
  91                     $results->option(elementSetName => 'marcxml');
  92                     $record = $results->record(0)->raw();
  93                 };
  94                 if ($@) {
  95                     # this doesn't exist, so no need to wail on zebra to delete it
  96                     if ($@->code() eq 13) {
  97                         $ok = 1;
  98                     } else {
  99                         # caught a ZOOM::Exception
 100                         my $error =
 101                             $@->message() . " ("
 102                             . $@->code() . ") "
 103                             . $@->addinfo() . " "
 104                             . $@->diagset();
 105                         warn "ERROR: $error";
 106                     }
 107                 } else {
 108                     # then, delete the record
 109                     warn "Deleting record" if $debug;
 110                     $ok = zebrado($record, $data->{'operation'}, $data->{'server'}, $data->{'biblio_auth_number'});
 111                 }
 112             }
 113             else {
 114                 # it is an update
 115                 warn "Updating record" if $debug;
 116                 # get the XML
 117                 my $marcxml;
 118                 if ($data->{'server'} eq "biblioserver") {
 119                     my $marc = GetMarcBiblio($data->{'biblio_auth_number'});
 120                     $marcxml = $marc->as_xml_record() if $marc;
 121                 }
 122                 elsif ($data->{'server'} eq "authorityserver") {
 123                     $marcxml = C4::AuthoritiesMarc::GetAuthorityXML($data->{'biblio_auth_number'});
 124                 }
 125                 # check it's XML, just in case
 126                 eval {
 127                     my $hashed = XMLin($marcxml);
 128                 }; ### is it a proper xml? broken xml may crash ZEBRA- slow but safe
 129                 ## it's Broken XML-- Should not reach here-- but if it does -lets protect ZEBRA
 130                 if ($@) {
 131                     Unix::Syslog::syslog LOG_ERR, "$@";
 132                     my $delsth = $dbh->prepare("UPDATE zebraqueue SET done = 1 WHERE id = ?");
 133                     $delsth->execute($data->{'id'});
 134                     next;
 135                 }
 136                 # ok, we have everything, do the operation in zebra !
 137                 $ok = zebrado($marcxml, $data->{'operation'}, $data->{'server'}, $data->{'biblio_auth_number'});
 138             }
 139             if ($ok == 1) {
 140                 $dbh = C4::Context->dbh;
 141                 my $delsth;
 142                 # if it's a deletion, we can delete every request on this biblio : in case the user
 143                 # did a modif (or item deletion) just before biblio deletion, there are some specialUpdate
 144                 # that are pending and can't succeed, as we don't have the XML anymore
 145                 # so, delete everything for this biblionumber
 146                 if ($data->{'operation'} =~ /delete/i) {
 147                     $delsth = $dbh->prepare("UPDATE zebraqueue SET done=1 WHERE biblio_auth_number = ?");
 148                     $delsth->execute($data->{'biblio_auth_number'});
 149                     # if it's not a deletion, delete every pending specialUpdate for this biblionumber
 150                     # in case the user add biblio, then X items, before this script runs
 151                     # this avoid indexing X+1 times where just 1 is enough.
 152                 } else {
 153                     $delsth = $dbh->prepare("UPDATE zebraqueue SET done = 1
 154                                              WHERE biblio_auth_number = ? and operation = 'specialUpdate'");
 155                     $delsth->execute($data->{'biblio_auth_number'});
 156                 }
 157             }
 158         };
 159         if ($@) {
 160             Unix::Syslog::syslog LOG_ERR, "$@";
 161         }
 162     }
 163     $readsth->finish();
 164     $kernel->yield('status_check');
 165 }
 166
 167 sub zebrado {
 168     ###Accepts a $server variable thus we can use it to update  biblios, authorities or other zebra dbs
 169     my ($record, $op, $server, $biblionumber) = @_;
 170
 171     warn "In zebrado" if $debug;
 172     my @port;
 173
 174     my $tried = 0;
 175     my $recon = 0;
 176     my $reconnect = 0;
 177 #    $record=Encode::encode("UTF-8",$record);
 178     my $shadow = $server."shadow";
 179     $op = 'recordDelete' if $op eq 'delete_record';
 180
 181 reconnect:
 182     warn "At reconnect" if $debug;
 183     my $Zconn = C4::Context->Zconn($server, 0, 1, '', 'xml');
 184     if ($record) {
 185         warn "Record found" if $debug;
 186         my $Zpackage = $Zconn->package();
 187         $Zpackage->option(action => $op);
 188         $Zpackage->option(record => $record);
 189 #       $Zpackage->option(recordIdOpaque => $biblionumber) if $biblionumber;
 190 retry:
 191         warn "At Retry" if $debug;
 192         eval { $Zpackage->send("update") };
 193         if ($@ && $@->isa("ZOOM::Exception")) {
 194         print "Oops!  ", $@->message(), "\n";
 195         return $@->code();
 196         }
 197         my($error, $errmsg, $addinfo, $diagset) = $Zconn->error_x();
 198         if ($error == 10007 && $tried < 3) {## timeout --another 30 looonng seconds for this update
 199             sleep 1;    ##  wait a sec!
 200             $tried++;
 201             goto "retry";
 202         } elsif ($error == 2 && $tried < 2) {## timeout --temporary zebra error !whatever that means
 203             sleep 2;    ##  wait two seconds!
 204             $tried++;
 205             goto "retry";
 206         } elsif ($error==10004 && $recon == 0) {##Lost connection -reconnect
 207             sleep 1;    ##  wait a sec!
 208             $recon = 1;
 209             $Zpackage->destroy();
 210             $Zconn->destroy();
 211             goto "reconnect";
 212         } elsif ($error) {
 213             $Zpackage->destroy();
 214             $Zconn->destroy();
 215             return 0;
 216         }
 217         $Zpackage->send('commit');
 218         return 1;
 219     }
 220     return 0;
 221 }
 222
 223
 224 sub handler_stop {
 225     my $heap = $_[HEAP];
 226     my $time = localtime(time);
 227     Unix::Syslog::syslog LOG_INFO, "$time Session ", $_[SESSION]->ID, " has stopped.\n";
 228     delete $heap->{session};
 229 }
 230
 231 POE::Session->create(
 232     inline_states => {
 233         _start       => \&handler_start,
 234         sleep        => \&handler_sleep,
 235         status_check => \&handler_check,
 236         do_ops       => \&zebraop,
 237         _stop        => \&handler_stop,
 238     },
 239 );
 240
 241 # start the kernel
 242 $poe_kernel->run();
 243
 244 Unix::Syslog::closelog;
 245
 246 exit;